hatchet/tests/graphframe.py
def check_filter_squash(gf, filter_func, expected_graph, expected_inc_time):
    """Ensure filtering and squashing results in the right Graph and GraphFrame."""
    orig_graph = gf.graph.copy()

    filtered = gf.filter(filter_func)
    index_names = filtered.dataframe.index.names
    filtered.dataframe.reset_index(inplace=True)
    assert filtered.graph is gf.graph
    assert filtered.graph == orig_graph
    assert all(n in filtered.graph.traverse() for n in filtered.dataframe["node"])
    filtered.dataframe.set_index(index_names, inplace=True)

    squashed = filtered.squash()
    index_names = squashed.dataframe.index.names
    squashed.dataframe.reset_index(inplace=True, drop=False)
    assert filtered.graph is not squashed.graph
    assert squashed.graph == expected_graph
    assert len(squashed.dataframe.index) == len(expected_graph)
    squashed_node_names = list(expected_graph.traverse(attrs="name"))
    assert all(
        n.frame["name"] in squashed_node_names for n in squashed.dataframe["node"]
    )
    squashed.dataframe.set_index(index_names, inplace=True)

    # verify inclusive metrics at different nodes
    nodes = list(squashed.graph.traverse())
    assert len(nodes) == len(expected_inc_time)

    assert expected_inc_time == [
        squashed.dataframe.loc[node, "time (inc)"] for node in nodes
    ]
Ensure filtering and squashing results in the right Graph and GraphFrame.
Ensure filtering and squashing results in the right Graph and GraphFrame.
Ensure filtering and squashing results in the right Graph and GraphFrame .
def check_filter_squash(gf, filter_func, expected_graph, expected_inc_time):
    orig_graph = gf.graph.copy()
    filtered = gf.filter(filter_func)
    index_names = filtered.dataframe.index.names
    filtered.dataframe.reset_index(inplace=True)
    assert filtered.graph is gf.graph
    assert filtered.graph == orig_graph
    assert all(n in filtered.graph.traverse() for n in filtered.dataframe["node"])
    filtered.dataframe.set_index(index_names, inplace=True)
    squashed = filtered.squash()
    index_names = squashed.dataframe.index.names
    squashed.dataframe.reset_index(inplace=True, drop=False)
    assert filtered.graph is not squashed.graph
    assert squashed.graph == expected_graph
    assert len(squashed.dataframe.index) == len(expected_graph)
    squashed_node_names = list(expected_graph.traverse(attrs="name"))
    assert all(
        n.frame["name"] in squashed_node_names for n in squashed.dataframe["node"]
    )
    squashed.dataframe.set_index(index_names, inplace=True)
    nodes = list(squashed.graph.traverse())
    assert len(nodes) == len(expected_inc_time)
    assert expected_inc_time == [
        squashed.dataframe.loc[node, "time (inc)"] for node in nodes
    ]
def check_filter_squash ( gf , filter_func , expected_graph , expected_inc_time ) : orig_graph = gf . graph . copy ( ) filtered = gf . filter ( filter_func ) index_names = filtered . dataframe . index . names filtered . dataframe . reset_index ( inplace = True ) assert filtered . graph is gf . graph assert filtered . graph == orig_graph assert all ( n in filtered . graph . traverse ( ) for n in filtered . dataframe [ "node" ] ) filtered . dataframe . set_index ( index_names , inplace = True ) squashed = filtered . squash ( ) index_names = squashed . dataframe . index . names squashed . dataframe . reset_index ( inplace = True , drop = False ) assert filtered . graph is not squashed . graph assert squashed . graph == expected_graph assert len ( squashed . dataframe . index ) == len ( expected_graph ) squashed_node_names = list ( expected_graph . traverse ( attrs = "name" ) ) assert all ( n . frame [ "name" ] in squashed_node_names for n in squashed . dataframe [ "node" ] ) squashed . dataframe . set_index ( index_names , inplace = True ) nodes = list ( squashed . graph . traverse ( ) ) assert len ( nodes ) == len ( expected_inc_time ) assert expected_inc_time == [ squashed . dataframe . loc [ node , "time (inc)" ] for node in nodes ]
Ensure filtering and squashing results in the right Graph and GraphFrame.
Ensure filtering and squashing results in the right Graph and GraphFrame .
"""Ensure filtering and squashing results in the right Graph and GraphFrame.""" # verify inclusive metrics at different nodes
gf
filter_func
expected_graph
expected_inc_time