Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/networkx/algorithms/connectivity/tests/test_cuts.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.algorithms.connectivity import minimum_st_edge_cut | |
6 from networkx.algorithms.connectivity import minimum_st_node_cut | |
7 from networkx.utils import arbitrary_element | |
8 | |
9 flow_funcs = [ | |
10 flow.boykov_kolmogorov, | |
11 flow.dinitz, | |
12 flow.edmonds_karp, | |
13 flow.preflow_push, | |
14 flow.shortest_augmenting_path, | |
15 ] | |
16 | |
17 # Tests for node and edge cutsets | |
18 | |
19 | |
20 def _generate_no_biconnected(max_attempts=50): | |
21 attempts = 0 | |
22 while True: | |
23 G = nx.fast_gnp_random_graph(100, 0.0575, seed=42) | |
24 if nx.is_connected(G) and not nx.is_biconnected(G): | |
25 attempts = 0 | |
26 yield G | |
27 else: | |
28 if attempts >= max_attempts: | |
29 msg = f"Tried {attempts} times: no suitable Graph." | |
30 raise Exception(msg) | |
31 else: | |
32 attempts += 1 | |
33 | |
34 | |
35 def test_articulation_points(): | |
36 Ggen = _generate_no_biconnected() | |
37 for flow_func in flow_funcs: | |
38 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
39 for i in range(1): # change 1 to 3 or more for more realizations. | |
40 G = next(Ggen) | |
41 cut = nx.minimum_node_cut(G, flow_func=flow_func) | |
42 assert len(cut) == 1, errmsg | |
43 assert cut.pop() in set(nx.articulation_points(G)), errmsg | |
44 | |
45 | |
46 def test_brandes_erlebach_book(): | |
47 # Figure 1 chapter 7: Connectivity | |
48 # http://www.informatik.uni-augsburg.de/thi/personen/kammer/Graph_Connectivity.pdf | |
49 G = nx.Graph() | |
50 G.add_edges_from( | |
51 [ | |
52 (1, 2), | |
53 (1, 3), | |
54 (1, 4), | |
55 (1, 5), | |
56 (2, 3), | |
57 (2, 6), | |
58 (3, 4), | |
59 (3, 6), | |
60 (4, 6), | |
61 (4, 7), | |
62 (5, 7), | |
63 (6, 8), | |
64 (6, 9), | |
65 (7, 8), | |
66 (7, 10), | |
67 (8, 11), | |
68 (9, 10), | |
69 (9, 11), | |
70 (10, 11), | |
71 ] | |
72 ) | |
73 for flow_func in flow_funcs: | |
74 kwargs = dict(flow_func=flow_func) | |
75 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
76 # edge cutsets | |
77 assert 3 == len(nx.minimum_edge_cut(G, 1, 11, **kwargs)), errmsg | |
78 edge_cut = nx.minimum_edge_cut(G, **kwargs) | |
79 # Node 5 has only two edges | |
80 assert 2 == len(edge_cut), errmsg | |
81 H = G.copy() | |
82 H.remove_edges_from(edge_cut) | |
83 assert not nx.is_connected(H), errmsg | |
84 # node cuts | |
85 assert {6, 7} == minimum_st_node_cut(G, 1, 11, **kwargs), errmsg | |
86 assert {6, 7} == nx.minimum_node_cut(G, 1, 11, **kwargs), errmsg | |
87 node_cut = nx.minimum_node_cut(G, **kwargs) | |
88 assert 2 == len(node_cut), errmsg | |
89 H = G.copy() | |
90 H.remove_nodes_from(node_cut) | |
91 assert not nx.is_connected(H), errmsg | |
92 | |
93 | |
94 def test_white_harary_paper(): | |
95 # Figure 1b white and harary (2001) | |
96 # http://eclectic.ss.uci.edu/~drwhite/sm-w23.PDF | |
97 # A graph with high adhesion (edge connectivity) and low cohesion | |
98 # (node connectivity) | |
99 G = nx.disjoint_union(nx.complete_graph(4), nx.complete_graph(4)) | |
100 G.remove_node(7) | |
101 for i in range(4, 7): | |
102 G.add_edge(0, i) | |
103 G = nx.disjoint_union(G, nx.complete_graph(4)) | |
104 G.remove_node(G.order() - 1) | |
105 for i in range(7, 10): | |
106 G.add_edge(0, i) | |
107 for flow_func in flow_funcs: | |
108 kwargs = dict(flow_func=flow_func) | |
109 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
110 # edge cuts | |
111 edge_cut = nx.minimum_edge_cut(G, **kwargs) | |
112 assert 3 == len(edge_cut), errmsg | |
113 H = G.copy() | |
114 H.remove_edges_from(edge_cut) | |
115 assert not nx.is_connected(H), errmsg | |
116 # node cuts | |
117 node_cut = nx.minimum_node_cut(G, **kwargs) | |
118 assert {0} == node_cut, errmsg | |
119 H = G.copy() | |
120 H.remove_nodes_from(node_cut) | |
121 assert not nx.is_connected(H), errmsg | |
122 | |
123 | |
124 def test_petersen_cutset(): | |
125 G = nx.petersen_graph() | |
126 for flow_func in flow_funcs: | |
127 kwargs = dict(flow_func=flow_func) | |
128 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
129 # edge cuts | |
130 edge_cut = nx.minimum_edge_cut(G, **kwargs) | |
131 assert 3 == len(edge_cut), errmsg | |
132 H = G.copy() | |
133 H.remove_edges_from(edge_cut) | |
134 assert not nx.is_connected(H), errmsg | |
135 # node cuts | |
136 node_cut = nx.minimum_node_cut(G, **kwargs) | |
137 assert 3 == len(node_cut), errmsg | |
138 H = G.copy() | |
139 H.remove_nodes_from(node_cut) | |
140 assert not nx.is_connected(H), errmsg | |
141 | |
142 | |
143 def test_octahedral_cutset(): | |
144 G = nx.octahedral_graph() | |
145 for flow_func in flow_funcs: | |
146 kwargs = dict(flow_func=flow_func) | |
147 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
148 # edge cuts | |
149 edge_cut = nx.minimum_edge_cut(G, **kwargs) | |
150 assert 4 == len(edge_cut), errmsg | |
151 H = G.copy() | |
152 H.remove_edges_from(edge_cut) | |
153 assert not nx.is_connected(H), errmsg | |
154 # node cuts | |
155 node_cut = nx.minimum_node_cut(G, **kwargs) | |
156 assert 4 == len(node_cut), errmsg | |
157 H = G.copy() | |
158 H.remove_nodes_from(node_cut) | |
159 assert not nx.is_connected(H), errmsg | |
160 | |
161 | |
162 def test_icosahedral_cutset(): | |
163 G = nx.icosahedral_graph() | |
164 for flow_func in flow_funcs: | |
165 kwargs = dict(flow_func=flow_func) | |
166 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
167 # edge cuts | |
168 edge_cut = nx.minimum_edge_cut(G, **kwargs) | |
169 assert 5 == len(edge_cut), errmsg | |
170 H = G.copy() | |
171 H.remove_edges_from(edge_cut) | |
172 assert not nx.is_connected(H), errmsg | |
173 # node cuts | |
174 node_cut = nx.minimum_node_cut(G, **kwargs) | |
175 assert 5 == len(node_cut), errmsg | |
176 H = G.copy() | |
177 H.remove_nodes_from(node_cut) | |
178 assert not nx.is_connected(H), errmsg | |
179 | |
180 | |
181 def test_node_cutset_exception(): | |
182 G = nx.Graph() | |
183 G.add_edges_from([(1, 2), (3, 4)]) | |
184 for flow_func in flow_funcs: | |
185 pytest.raises(nx.NetworkXError, nx.minimum_node_cut, G, flow_func=flow_func) | |
186 | |
187 | |
188 def test_node_cutset_random_graphs(): | |
189 for flow_func in flow_funcs: | |
190 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
191 for i in range(3): | |
192 G = nx.fast_gnp_random_graph(50, 0.25, seed=42) | |
193 if not nx.is_connected(G): | |
194 ccs = iter(nx.connected_components(G)) | |
195 start = arbitrary_element(next(ccs)) | |
196 G.add_edges_from((start, arbitrary_element(c)) for c in ccs) | |
197 cutset = nx.minimum_node_cut(G, flow_func=flow_func) | |
198 assert nx.node_connectivity(G) == len(cutset), errmsg | |
199 G.remove_nodes_from(cutset) | |
200 assert not nx.is_connected(G), errmsg | |
201 | |
202 | |
203 def test_edge_cutset_random_graphs(): | |
204 for flow_func in flow_funcs: | |
205 errmsg = f"Assertion failed in function: {flow_func.__name__}" | |
206 for i in range(3): | |
207 G = nx.fast_gnp_random_graph(50, 0.25, seed=42) | |
208 if not nx.is_connected(G): | |
209 ccs = iter(nx.connected_components(G)) | |
210 start = arbitrary_element(next(ccs)) | |
211 G.add_edges_from((start, arbitrary_element(c)) for c in ccs) | |
212 cutset = nx.minimum_edge_cut(G, flow_func=flow_func) | |
213 assert nx.edge_connectivity(G) == len(cutset), errmsg | |
214 G.remove_edges_from(cutset) | |
215 assert not nx.is_connected(G), errmsg | |
216 | |
217 | |
218 def test_empty_graphs(): | |
219 G = nx.Graph() | |
220 D = nx.DiGraph() | |
221 for interface_func in [nx.minimum_node_cut, nx.minimum_edge_cut]: | |
222 for flow_func in flow_funcs: | |
223 pytest.raises( | |
224 nx.NetworkXPointlessConcept, interface_func, G, flow_func=flow_func | |
225 ) | |
226 pytest.raises( | |
227 nx.NetworkXPointlessConcept, interface_func, D, flow_func=flow_func | |
228 ) | |
229 | |
230 | |
231 def test_unbounded(): | |
232 G = nx.complete_graph(5) | |
233 for flow_func in flow_funcs: | |
234 assert 4 == len(minimum_st_edge_cut(G, 1, 4, flow_func=flow_func)) | |
235 | |
236 | |
237 def test_missing_source(): | |
238 G = nx.path_graph(4) | |
239 for interface_func in [nx.minimum_edge_cut, nx.minimum_node_cut]: | |
240 for flow_func in flow_funcs: | |
241 pytest.raises( | |
242 nx.NetworkXError, interface_func, G, 10, 1, flow_func=flow_func | |
243 ) | |
244 | |
245 | |
246 def test_missing_target(): | |
247 G = nx.path_graph(4) | |
248 for interface_func in [nx.minimum_edge_cut, nx.minimum_node_cut]: | |
249 for flow_func in flow_funcs: | |
250 pytest.raises( | |
251 nx.NetworkXError, interface_func, G, 1, 10, flow_func=flow_func | |
252 ) | |
253 | |
254 | |
255 def test_not_weakly_connected(): | |
256 G = nx.DiGraph() | |
257 nx.add_path(G, [1, 2, 3]) | |
258 nx.add_path(G, [4, 5]) | |
259 for interface_func in [nx.minimum_edge_cut, nx.minimum_node_cut]: | |
260 for flow_func in flow_funcs: | |
261 pytest.raises(nx.NetworkXError, interface_func, G, flow_func=flow_func) | |
262 | |
263 | |
264 def test_not_connected(): | |
265 G = nx.Graph() | |
266 nx.add_path(G, [1, 2, 3]) | |
267 nx.add_path(G, [4, 5]) | |
268 for interface_func in [nx.minimum_edge_cut, nx.minimum_node_cut]: | |
269 for flow_func in flow_funcs: | |
270 pytest.raises(nx.NetworkXError, interface_func, G, flow_func=flow_func) | |
271 | |
272 | |
273 def tests_min_cut_complete(): | |
274 G = nx.complete_graph(5) | |
275 for interface_func in [nx.minimum_edge_cut, nx.minimum_node_cut]: | |
276 for flow_func in flow_funcs: | |
277 assert 4 == len(interface_func(G, flow_func=flow_func)) | |
278 | |
279 | |
280 def tests_min_cut_complete_directed(): | |
281 G = nx.complete_graph(5) | |
282 G = G.to_directed() | |
283 for interface_func in [nx.minimum_edge_cut, nx.minimum_node_cut]: | |
284 for flow_func in flow_funcs: | |
285 assert 4 == len(interface_func(G, flow_func=flow_func)) | |
286 | |
287 | |
288 def tests_minimum_st_node_cut(): | |
289 G = nx.Graph() | |
290 G.add_nodes_from([0, 1, 2, 3, 7, 8, 11, 12]) | |
291 G.add_edges_from([(7, 11), (1, 11), (1, 12), (12, 8), (0, 1)]) | |
292 nodelist = minimum_st_node_cut(G, 7, 11) | |
293 assert nodelist == {} | |
294 | |
295 | |
296 def test_invalid_auxiliary(): | |
297 G = nx.complete_graph(5) | |
298 pytest.raises(nx.NetworkXError, minimum_st_node_cut, G, 0, 3, auxiliary=G) | |
299 | |
300 | |
301 def test_interface_only_source(): | |
302 G = nx.complete_graph(5) | |
303 for interface_func in [nx.minimum_node_cut, nx.minimum_edge_cut]: | |
304 pytest.raises(nx.NetworkXError, interface_func, G, s=0) | |
305 | |
306 | |
307 def test_interface_only_target(): | |
308 G = nx.complete_graph(5) | |
309 for interface_func in [nx.minimum_node_cut, nx.minimum_edge_cut]: | |
310 pytest.raises(nx.NetworkXError, interface_func, G, t=3) |