Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

adding sort shortest to longest and vice versa #41

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
170 changes: 115 additions & 55 deletions tests/test_pathway_analysis.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,8 @@ def test_ndp():


@pytest.mark.parametrize("source,target", [('import', 'heu_pu_collector'),
('mine', 'export'),
('import', 'export')])
('mine', 'export'),
('import', 'export')])
def test_ndp_raise_exception(source, target):
G = nx.DiGraph()
edges = [('mine', 'civ_enrich'), ('mine', 'mil_enrich'),
Expand All @@ -36,7 +36,7 @@ def test_ndp_raise_exception(source, target):
('mil_enrich', 'heu_pu_collector')]
G.add_edges_from(edges)

with pytest.raises(ValueError) as excinfo:
with pytest.raises(ValueError) as excinfo:
obj = pa.find_node_disjoint_paths(G, source, target)
assert 'Source and/or target not in graph G!' in str(excinfo.value)

Expand Down Expand Up @@ -101,7 +101,7 @@ def test_has_multiedges_not_multigraph():
exp = None
G = nx.DiGraph()
G.add_edge('a', 'b')

obs = pa.has_multiedges(G)
assert obs == exp

Expand All @@ -119,6 +119,17 @@ def test_transform_to_digraph():
assert obs == exp


def test_transform_to_digraph_has_multiedges():
G = nx.MultiDiGraph()
G.add_edge('a', 'b')
G.add_edge('a', 'b')
exp = False

(obs_H, obs_safe) = pa.transform_to_digraph(G)

assert obs_safe == exp


@pytest.mark.parametrize("G", [(nx.Graph()), (nx.MultiGraph())])
def test_transform_to_digraph_unsafe(G):
exp = False
Expand All @@ -130,7 +141,7 @@ def test_transform_to_digraph_unsafe(G):
assert obs_safe == exp


@pytest.mark.parametrize("G,exp", [(nx.Graph(), None),
@pytest.mark.parametrize("G,exp", [(nx.Graph(), None),
(nx.MultiGraph(), None)])
def test_transform_to_digraph_formats(G, exp):
G.add_edge('a', 'b')
Expand All @@ -139,7 +150,7 @@ def test_transform_to_digraph_formats(G, exp):
assert obs_H == exp


@pytest.mark.parametrize("G,exp", [(nx.Graph(), False),
@pytest.mark.parametrize("G,exp", [(nx.Graph(), False),
(nx.DiGraph(), True),
(nx.MultiGraph(), False),
(nx.MultiDiGraph(), True)])
Expand Down Expand Up @@ -176,7 +187,7 @@ def test_maximum_flow_several_paths():


def test_maximum_flow_multigraph():
'''NetworkX retains only the most recently added edge if multiple edges
'''NetworkX retains only the most recently added edge if multiple edges
exit and a MultiGraph is converted into a DiGraph'''
exp = 1
G = nx.DiGraph()
Expand All @@ -190,7 +201,7 @@ def test_maximum_flow_multigraph():


def test_maximum_flow_multigraph_order_flipped():
'''NetworkX retains only the most recently added edge if multiple edges
'''NetworkX retains only the most recently added edge if multiple edges
exit and a MultiGraph is converted into a DiGraph'''
exp = 2
G = nx.DiGraph()
Expand Down Expand Up @@ -221,8 +232,8 @@ def test_maximum_flow_directed_only():
def test_maximum_flow_not_directed_or_multi():
exp = None
G = nx.Graph()
G.add_edge('a', 'b', capacity = 2)
G.add_edge('b', 'c', capacity = 5)
G.add_edge('a', 'b', capacity=2)
G.add_edge('b', 'c', capacity=5)

with pytest.raises(TypeError) as excinfo:
(obs_path, obs) = pa.find_maximum_flow(G, 'a', 'c')
Expand All @@ -232,14 +243,14 @@ def test_maximum_flow_not_directed_or_multi():
def test_find_pathway_flow():
G = nx.DiGraph()
edges = [(0, 1, {'capacity': 3}),
(1, 2, {'capacity': 5}),
(2, 3, {'capacity': 3}),
(3, 4, {'capacity': 4}),
(0, 5, {'capacity': 2}),
(5, 6, {'capacity': 3}),
(6, 3, {'capacity': 2}),
(3, 7, {'capacity': 1}),
(7, 4, {'capacity': 2})]
(1, 2, {'capacity': 5}),
(2, 3, {'capacity': 3}),
(3, 4, {'capacity': 4}),
(0, 5, {'capacity': 2}),
(5, 6, {'capacity': 3}),
(6, 3, {'capacity': 2}),
(3, 7, {'capacity': 1}),
(7, 4, {'capacity': 2})]
G.add_edges_from(edges)
path = (0, 5, 6, 3, 7)

Expand All @@ -248,15 +259,16 @@ def test_find_pathway_flow():

assert obs == exp


def test_find_pathway_flow_no_capacity():
G = nx.DiGraph()
edges = [(0,1), (1,2), (2,3)]
edges = [(0, 1), (1, 2), (2, 3)]
G.add_edges_from(edges)
path = (0, 1, 2, 3)

with pytest.raises(nx.exception.NetworkXUnbounded) as excinfo:
obj = pa.find_pathway_flow(G, path)
assert 'Infinite capacity path, flow unbounded above.' in str(excinfo.value)
with pytest.raises(nx.exception.NetworkXUnbounded) as excinfo:
obj = pa.find_pathway_flow(G, path)
assert 'Infinite capacity path' in str(excinfo.value)


def test_find_pathway_flow_single_infinite():
Expand All @@ -281,14 +293,14 @@ def test_find_pathway_flow_all_infinite():
G.add_edges_from(edges)
path = (0, 1, 2, 3)

with pytest.raises(nx.exception.NetworkXUnbounded) as excinfo:
obj = pa.find_pathway_flow(G, path)
assert 'Infinite capacity path, flow unbounded above.' in str(excinfo.value)
with pytest.raises(nx.exception.NetworkXUnbounded) as excinfo:
obj = pa.find_pathway_flow(G, path)
assert 'Infinite capacity path' in str(excinfo.value)


def test_find_pathway_flow_multiedges():
G = nx.MultiDiGraph()
edges = [(0,1), (1,2), (1,2), (2,3)]
edges = [(0, 1), (1, 2), (1, 2), (2, 3)]
G.add_edges_from(edges)
pathway = (0, 1, 2, 3)

Expand All @@ -299,7 +311,7 @@ def test_find_pathway_flow_multiedges():

def test_find_pathway_flow_other_type():
G = nx.Graph()
edges = [(0,1), (1,2), (2,3)]
edges = [(0, 1), (1, 2), (2, 3)]
G.add_edges_from(edges)
pathway = (0, 1, 2, 3)

Expand All @@ -310,7 +322,15 @@ def test_find_pathway_flow_other_type():

@pytest.mark.parametrize("steps,exp", [(('FacilityA', 'FacilityB'), 1),
(('FacilityA', 'Sink'), -1)])
def test_check_if_sublist(steps,exp):
def test_check_if_sublist(steps, exp):
path = ('Source', 'FacilityA', 'FacilityB', 'Sink')
pos = pa.check_if_sublist(path, steps)
assert pos == exp


@pytest.mark.parametrize("path,steps,exp", [(('Source', 'Sink'), (), -1),
((), ('Sink'), -1)])
def test_check_if_sublist_len_zero(path, steps, exp):
path = ('Source', 'FacilityA', 'FacilityB', 'Sink')
pos = pa.check_if_sublist(path, steps)
assert pos == exp
Expand All @@ -331,7 +351,7 @@ def test_roll_cycle(cycle, exp):
def test_insert_cycle_single():
path = (0, 1, 3, 4, 7, 8)
rolled_cycles = {(7, 3, 4)}

exp = (0, 1, 3, 4, (7, 3, 4), 7, 8)
obs = pa.insert_cycles(path, rolled_cycles)
assert obs == exp
Expand All @@ -340,7 +360,7 @@ def test_insert_cycle_single():
def test_insert_cycle_loop():
path = (0, 1, 3, 4, 7, 8)
rolled_cycles = {(7,)}

exp = (0, 1, 3, 4, (7,), 7, 8)
obs = pa.insert_cycles(path, rolled_cycles)
assert obs == exp
Expand All @@ -356,10 +376,10 @@ def test_insert_cycle_multiple():


def test_insert_cycle_multiple_same_index():
path = ('Source' , 'A', 'B', 'C', 'Sink')
path = ('Source', 'A', 'B', 'C', 'Sink')
rolled_cycles = (('C', 'D', 'B'), ('C', 'D', 'E', 'A', 'B'))
exp = ('Source' , 'A', 'B', ('C', 'D', 'B'), ('C', 'D', 'E', 'A', 'B'),

exp = ('Source', 'A', 'B', ('C', 'D', 'B'), ('C', 'D', 'E', 'A', 'B'),
'C', 'Sink')
obs = pa.insert_cycles(path, rolled_cycles)
assert obs == exp
Expand Down Expand Up @@ -393,7 +413,7 @@ def test_get_pathways_with_cycles_multiple():
(0, 2, (6, 5), 6, (7, 3, 4), 7, 8)}

obs = pa.get_pathways_with_cycles(pathways, sc)
#assert all(x in obs for x in exp)
# assert all(x in obs for x in exp)
assert obs == exp


Expand Down Expand Up @@ -422,10 +442,10 @@ def data():

@pytest.mark.parametrize("source,exp", [("SourceA",
{("SourceA", "Facility", "SinkA"),
("SourceA", "Facility", "SinkB")}),
("SourceA", "Facility", "SinkB")}),
("SourceC", set()),
("Facility", set())])
def test_find_paths_with_source(source,exp):
def test_find_paths_with_source(source, exp):
pathways = testdata[2][4]

obs_subset = pa.find_paths_with_source(pathways, source)
Expand All @@ -441,10 +461,10 @@ def test_find_paths_with_source_none():

@pytest.mark.parametrize("sink,exp", [("SinkB",
{("SourceA", "Facility", "SinkB"),
("SourceB", "Facility", "SinkB")}),
("SourceB", "Facility", "SinkB")}),
("SinkC", set()),
("SourceA", set())])
def test_find_paths_with_sink(sink,exp):
def test_find_paths_with_sink(sink, exp):
pathways = testdata[2][4]

obs_subset = pa.find_paths_with_sink(pathways, sink)
Expand All @@ -458,9 +478,9 @@ def test_find_paths_with_sink_none():
assert obs == set()


@pytest.mark.parametrize("contain,exp", [("SourceB",
@pytest.mark.parametrize("contain,exp", [("SourceB",
{("SourceB", "Facility", "SinkA"),
("SourceB", "Facility", "SinkB")}),
("SourceB", "Facility", "SinkB")}),
(["SourceB", "SinkB"],
{("SourceB", "Facility", "SinkB")}),
(["SourceB", "SourceA"], set()),
Expand All @@ -472,17 +492,18 @@ def test_find_paths_containing_all(contain, exp):
assert obs == exp


@pytest.mark.parametrize("contain,exp", [(5, {(0,1,5,6,4)}),
([5], {(0,1,5,6,4)}),
([1, 5], {(0,1,5,6,4)}),
([0, 3], {(0,1,2,3,4), (0,2,3,4)}),
@pytest.mark.parametrize("contain,exp", [(5, {(0, 1, 5, 6, 4)}),
([5], {(0, 1, 5, 6, 4)}),
([1, 5], {(0, 1, 5, 6, 4)}),
([0, 3], {(0, 1, 2, 3, 4),
(0, 2, 3, 4)}),
([5, 7], set()),
([8], set()),
([7,8], set()),
([7, 8], set()),
([], set())])
def test_find_paths_containing_all_int(contain, exp):
pathways = {(0,1,2,3,4), (0,2,3,4), (0,1,5,6,4), (0,1,7,4)}
pathways = {(0, 1, 2, 3, 4), (0, 2, 3, 4), (0, 1, 5, 6, 4), (0, 1, 7, 4)}

obs = pa.find_paths_containing_all(pathways, contain)
assert obs == exp

Expand All @@ -496,11 +517,11 @@ def test_find_paths_containing_all_none():

@pytest.mark.parametrize("contain,exp", [("SourceA",
{("SourceA", "Facility", "SinkA"),
("SourceA", "Facility", "SinkB")}),
("SourceA", "Facility", "SinkB")}),
(["SourceA", "SinkA"],
{("SourceA", "Facility", "SinkA"),
("SourceA", "Facility", "SinkB"),
("SourceB", "Facility", "SinkA")}),
("SourceA", "Facility", "SinkB"),
("SourceB", "Facility", "SinkA")}),
(["SourceC"], set()),
([], set())])
def test_find_paths_containing_one_of(contain, exp):
Expand All @@ -510,15 +531,16 @@ def test_find_paths_containing_one_of(contain, exp):
assert obs == exp


@pytest.mark.parametrize("contain,exp", [(5, {(0,1,5,6,4)}),
([5], {(0,1,5,6,4)}),
([5, 7], {(0,1,5,6,4),(0,1,7,4)}),
@pytest.mark.parametrize("contain,exp", [(5, {(0, 1, 5, 6, 4)}),
([5], {(0, 1, 5, 6, 4)}),
([5, 7], {(0, 1, 5, 6, 4),
(0, 1, 7, 4)}),
(8, set()),
([8], set()),
([7,8], {(0,1,7,4)}),
([7, 8], {(0, 1, 7, 4)}),
([], set())])
def test_find_paths_containing_one_of_int(contain, exp):
pathways = {(0,1,2,3,4), (0,2,3,4),(0,1,5,6,4), (0,1,7,4)}
pathways = {(0, 1, 2, 3, 4), (0, 2, 3, 4), (0, 1, 5, 6, 4), (0, 1, 7, 4)}

obs = pa.find_paths_containing_one_of(pathways, contain)
assert obs == exp
Expand Down Expand Up @@ -552,3 +574,41 @@ def test_get_longest_paths(name, short, long, edges, paths, sc):
exp = {path for path in paths if len(path) == long}
obs = pa.get_longest_path(paths)
assert obs == exp


def test_sort_shortest():
pathways = {(2, 3), (3, 4, 7, 4, 3, 32, 3), (10000, 10000, 0)}
exp = [(2, 3), (10000, 10000, 0), (3, 4, 7, 4, 3, 32, 3)]

obs = pa.sort_by_shortest(pathways)
assert obs == exp


def test_sort_longest():
pathways = {(2, 3), (3, 4, 7, 4, 3, 32, 3), (10000, 10000, 0)}
exp = [(3, 4, 7, 4, 3, 32, 3), (10000, 10000, 0), (2, 3)]

obs = pa.sort_by_longest(pathways)
assert obs == exp


@pytest.mark.parametrize("pathways", [{("a", "b")},
{("a", "b"), (1, "b")},
{(1, 4)},
{(4, 5), (6, 5.9)},
{(6.3, "a")}])
def test_check_for_invalid_pathways(pathways):
obs = pa.check_for_invalid_pathways(pathways)
assert obs is None
Comment on lines +601 to +602
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does it make more sense to some how test that there is no exception?



@pytest.mark.parametrize("pathways", [{("a")},
{("a", "b"), "c"},
{(1)},
{(4, 5), 6},
{(6.3, "a"), 4.6},
{(5.3)}])
def test_check_for_invalid_pathways_type_error(pathways):
with pytest.raises(TypeError) as excinfo:
obj = pa.sort_by_longest(pathways)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why not just call check_for_invalid_pathways here?

assert 'pathways contains' in str(excinfo.value)
Loading