Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/networkx/algorithms/connectivity/tests/test_disjoint_paths.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 import pytest | |
2 | |
3 import networkx as nx | |
4 from networkx.algorithms import flow | |
5 from networkx.utils import pairwise | |
6 | |
7 flow_funcs = [ | |
8 flow.boykov_kolmogorov, | |
9 flow.edmonds_karp, | |
10 flow.dinitz, | |
11 flow.preflow_push, | |
12 flow.shortest_augmenting_path, | |
13 ] | |
14 | |
15 | |
16 def is_path(G, path): | |
17 return all(v in G[u] for u, v in pairwise(path)) | |
18 | |
19 | |
20 def are_edge_disjoint_paths(G, paths): | |
21 if not paths: | |
22 return False | |
23 for path in paths: | |
24 assert is_path(G, path) | |
25 paths_edges = [list(pairwise(p)) for p in paths] | |
26 num_of_edges = sum(len(e) for e in paths_edges) | |
27 num_unique_edges = len(set.union(*[set(es) for es in paths_edges])) | |
28 if num_of_edges == num_unique_edges: | |
29 return True | |
30 return False | |
31 | |
32 | |
33 def are_node_disjoint_paths(G, paths): | |
34 if not paths: | |
35 return False | |
36 for path in paths: | |
37 assert is_path(G, path) | |
38 # first and last nodes are source and target | |
39 st = {paths[0][0], paths[0][-1]} | |
40 num_of_nodes = len([n for path in paths for n in path if n not in st]) | |
41 num_unique_nodes = len({n for path in paths for n in path if n not in st}) | |
42 if num_of_nodes == num_unique_nodes: | |
43 return True | |
44 return False | |
45 | |
46 | |
47 def test_graph_from_pr_2053(): | |
48 G = nx.Graph() | |
49 G.add_edges_from( | |
50 [ | |
51 ("A", "B"), | |
52 ("A", "D"), | |
53 ("A", "F"), | |
54 ("A", "G"), | |
55 ("B", "C"), | |
56 ("B", "D"), | |
57 ("B", "G"), | |
58 ("C", "D"), | |
59 ("C", "E"), | |
60 ("C", "Z"), | |
61 ("D", "E"), | |
62 ("D", "F"), | |
63 ("E", "F"), | |
64 ("E", "Z"), | |
65 ("F", "Z"), | |
66 ("G", "Z"), | |
67 ] | |
68 ) | |
69 for flow_func in flow_funcs: | |
70 kwargs = dict(flow_func=flow_func) | |
71 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
72 # edge disjoint paths | |
73 edge_paths = list(nx.edge_disjoint_paths(G, "A", "Z", **kwargs)) | |
74 assert are_edge_disjoint_paths(G, edge_paths), errmsg | |
75 assert nx.edge_connectivity(G, "A", "Z") == len(edge_paths), errmsg | |
76 # node disjoint paths | |
77 node_paths = list(nx.node_disjoint_paths(G, "A", "Z", **kwargs)) | |
78 assert are_node_disjoint_paths(G, node_paths), errmsg | |
79 assert nx.node_connectivity(G, "A", "Z") == len(node_paths), errmsg | |
80 | |
81 | |
82 def test_florentine_families(): | |
83 G = nx.florentine_families_graph() | |
84 for flow_func in flow_funcs: | |
85 kwargs = dict(flow_func=flow_func) | |
86 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
87 # edge disjoint paths | |
88 edge_dpaths = list(nx.edge_disjoint_paths(G, "Medici", "Strozzi", **kwargs)) | |
89 assert are_edge_disjoint_paths(G, edge_dpaths), errmsg | |
90 assert nx.edge_connectivity(G, "Medici", "Strozzi") == len(edge_dpaths), errmsg | |
91 # node disjoint paths | |
92 node_dpaths = list(nx.node_disjoint_paths(G, "Medici", "Strozzi", **kwargs)) | |
93 assert are_node_disjoint_paths(G, node_dpaths), errmsg | |
94 assert nx.node_connectivity(G, "Medici", "Strozzi") == len(node_dpaths), errmsg | |
95 | |
96 | |
97 def test_karate(): | |
98 G = nx.karate_club_graph() | |
99 for flow_func in flow_funcs: | |
100 kwargs = dict(flow_func=flow_func) | |
101 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
102 # edge disjoint paths | |
103 edge_dpaths = list(nx.edge_disjoint_paths(G, 0, 33, **kwargs)) | |
104 assert are_edge_disjoint_paths(G, edge_dpaths), errmsg | |
105 assert nx.edge_connectivity(G, 0, 33) == len(edge_dpaths), errmsg | |
106 # node disjoint paths | |
107 node_dpaths = list(nx.node_disjoint_paths(G, 0, 33, **kwargs)) | |
108 assert are_node_disjoint_paths(G, node_dpaths), errmsg | |
109 assert nx.node_connectivity(G, 0, 33) == len(node_dpaths), errmsg | |
110 | |
111 | |
112 def test_petersen_disjoint_paths(): | |
113 G = nx.petersen_graph() | |
114 for flow_func in flow_funcs: | |
115 kwargs = dict(flow_func=flow_func) | |
116 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
117 # edge disjoint paths | |
118 edge_dpaths = list(nx.edge_disjoint_paths(G, 0, 6, **kwargs)) | |
119 assert are_edge_disjoint_paths(G, edge_dpaths), errmsg | |
120 assert 3 == len(edge_dpaths), errmsg | |
121 # node disjoint paths | |
122 node_dpaths = list(nx.node_disjoint_paths(G, 0, 6, **kwargs)) | |
123 assert are_node_disjoint_paths(G, node_dpaths), errmsg | |
124 assert 3 == len(node_dpaths), errmsg | |
125 | |
126 | |
127 def test_octahedral_disjoint_paths(): | |
128 G = nx.octahedral_graph() | |
129 for flow_func in flow_funcs: | |
130 kwargs = dict(flow_func=flow_func) | |
131 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
132 # edge disjoint paths | |
133 edge_dpaths = list(nx.edge_disjoint_paths(G, 0, 5, **kwargs)) | |
134 assert are_edge_disjoint_paths(G, edge_dpaths), errmsg | |
135 assert 4 == len(edge_dpaths), errmsg | |
136 # node disjoint paths | |
137 node_dpaths = list(nx.node_disjoint_paths(G, 0, 5, **kwargs)) | |
138 assert are_node_disjoint_paths(G, node_dpaths), errmsg | |
139 assert 4 == len(node_dpaths), errmsg | |
140 | |
141 | |
142 def test_icosahedral_disjoint_paths(): | |
143 G = nx.icosahedral_graph() | |
144 for flow_func in flow_funcs: | |
145 kwargs = dict(flow_func=flow_func) | |
146 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
147 # edge disjoint paths | |
148 edge_dpaths = list(nx.edge_disjoint_paths(G, 0, 6, **kwargs)) | |
149 assert are_edge_disjoint_paths(G, edge_dpaths), errmsg | |
150 assert 5 == len(edge_dpaths), errmsg | |
151 # node disjoint paths | |
152 node_dpaths = list(nx.node_disjoint_paths(G, 0, 6, **kwargs)) | |
153 assert are_node_disjoint_paths(G, node_dpaths), errmsg | |
154 assert 5 == len(node_dpaths), errmsg | |
155 | |
156 | |
157 def test_cutoff_disjoint_paths(): | |
158 G = nx.icosahedral_graph() | |
159 for flow_func in flow_funcs: | |
160 kwargs = dict(flow_func=flow_func) | |
161 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
162 for cutoff in [2, 4]: | |
163 kwargs["cutoff"] = cutoff | |
164 # edge disjoint paths | |
165 edge_dpaths = list(nx.edge_disjoint_paths(G, 0, 6, **kwargs)) | |
166 assert are_edge_disjoint_paths(G, edge_dpaths), errmsg | |
167 assert cutoff == len(edge_dpaths), errmsg | |
168 # node disjoint paths | |
169 node_dpaths = list(nx.node_disjoint_paths(G, 0, 6, **kwargs)) | |
170 assert are_node_disjoint_paths(G, node_dpaths), errmsg | |
171 assert cutoff == len(node_dpaths), errmsg | |
172 | |
173 | |
174 def test_missing_source_edge_paths(): | |
175 with pytest.raises(nx.NetworkXError): | |
176 G = nx.path_graph(4) | |
177 list(nx.edge_disjoint_paths(G, 10, 1)) | |
178 | |
179 | |
180 def test_missing_source_node_paths(): | |
181 with pytest.raises(nx.NetworkXError): | |
182 G = nx.path_graph(4) | |
183 list(nx.node_disjoint_paths(G, 10, 1)) | |
184 | |
185 | |
186 def test_missing_target_edge_paths(): | |
187 with pytest.raises(nx.NetworkXError): | |
188 G = nx.path_graph(4) | |
189 list(nx.edge_disjoint_paths(G, 1, 10)) | |
190 | |
191 | |
192 def test_missing_target_node_paths(): | |
193 with pytest.raises(nx.NetworkXError): | |
194 G = nx.path_graph(4) | |
195 list(nx.node_disjoint_paths(G, 1, 10)) | |
196 | |
197 | |
198 def test_not_weakly_connected_edges(): | |
199 with pytest.raises(nx.NetworkXNoPath): | |
200 G = nx.DiGraph() | |
201 nx.add_path(G, [1, 2, 3]) | |
202 nx.add_path(G, [4, 5]) | |
203 list(nx.edge_disjoint_paths(G, 1, 5)) | |
204 | |
205 | |
206 def test_not_weakly_connected_nodes(): | |
207 with pytest.raises(nx.NetworkXNoPath): | |
208 G = nx.DiGraph() | |
209 nx.add_path(G, [1, 2, 3]) | |
210 nx.add_path(G, [4, 5]) | |
211 list(nx.node_disjoint_paths(G, 1, 5)) | |
212 | |
213 | |
214 def test_not_connected_edges(): | |
215 with pytest.raises(nx.NetworkXNoPath): | |
216 G = nx.Graph() | |
217 nx.add_path(G, [1, 2, 3]) | |
218 nx.add_path(G, [4, 5]) | |
219 list(nx.edge_disjoint_paths(G, 1, 5)) | |
220 | |
221 | |
222 def test_not_connected_nodes(): | |
223 with pytest.raises(nx.NetworkXNoPath): | |
224 G = nx.Graph() | |
225 nx.add_path(G, [1, 2, 3]) | |
226 nx.add_path(G, [4, 5]) | |
227 list(nx.node_disjoint_paths(G, 1, 5)) | |
228 | |
229 | |
230 def test_isolated_edges(): | |
231 with pytest.raises(nx.NetworkXNoPath): | |
232 G = nx.Graph() | |
233 G.add_node(1) | |
234 nx.add_path(G, [4, 5]) | |
235 list(nx.edge_disjoint_paths(G, 1, 5)) | |
236 | |
237 | |
238 def test_isolated_nodes(): | |
239 with pytest.raises(nx.NetworkXNoPath): | |
240 G = nx.Graph() | |
241 G.add_node(1) | |
242 nx.add_path(G, [4, 5]) | |
243 list(nx.node_disjoint_paths(G, 1, 5)) | |
244 | |
245 | |
246 def test_invalid_auxiliary(): | |
247 with pytest.raises(nx.NetworkXError): | |
248 G = nx.complete_graph(5) | |
249 list(nx.node_disjoint_paths(G, 0, 3, auxiliary=G)) |