1import pytest 2import networkx as nx 3import os 4 5 6@pytest.fixture 7def simple_flow_graph(): 8 G = nx.DiGraph() 9 G.add_node("a", demand=0) 10 G.add_node("b", demand=-5) 11 G.add_node("c", demand=50000000) 12 G.add_node("d", demand=-49999995) 13 G.add_edge("a", "b", weight=3, capacity=4) 14 G.add_edge("a", "c", weight=6, capacity=10) 15 G.add_edge("b", "d", weight=1, capacity=9) 16 G.add_edge("c", "d", weight=2, capacity=5) 17 return G 18 19 20@pytest.fixture 21def simple_no_flow_graph(): 22 G = nx.DiGraph() 23 G.add_node("s", demand=-5) 24 G.add_node("t", demand=5) 25 G.add_edge("s", "a", weight=1, capacity=3) 26 G.add_edge("a", "b", weight=3) 27 G.add_edge("a", "c", weight=-6) 28 G.add_edge("b", "d", weight=1) 29 G.add_edge("c", "d", weight=-2) 30 G.add_edge("d", "t", weight=1, capacity=3) 31 return G 32 33 34def get_flowcost_from_flowdict(G, flowDict): 35 """Returns flow cost calculated from flow dictionary""" 36 flowCost = 0 37 for u in flowDict.keys(): 38 for v in flowDict[u].keys(): 39 flowCost += flowDict[u][v] * G[u][v]["weight"] 40 return flowCost 41 42 43def test_infinite_demand_raise(simple_flow_graph): 44 G = simple_flow_graph 45 inf = float("inf") 46 nx.set_node_attributes(G, {"a": {"demand": inf}}) 47 pytest.raises(nx.NetworkXError, nx.network_simplex, G) 48 49 50def test_neg_infinite_demand_raise(simple_flow_graph): 51 G = simple_flow_graph 52 inf = float("inf") 53 nx.set_node_attributes(G, {"a": {"demand": -inf}}) 54 pytest.raises(nx.NetworkXError, nx.network_simplex, G) 55 56 57def test_infinite_weight_raise(simple_flow_graph): 58 G = simple_flow_graph 59 inf = float("inf") 60 nx.set_edge_attributes( 61 G, {("a", "b"): {"weight": inf}, ("b", "d"): {"weight": inf}} 62 ) 63 pytest.raises(nx.NetworkXError, nx.network_simplex, G) 64 65 66def test_nonzero_net_demand_raise(simple_flow_graph): 67 G = simple_flow_graph 68 nx.set_node_attributes(G, {"b": {"demand": -4}}) 69 pytest.raises(nx.NetworkXUnfeasible, nx.network_simplex, G) 70 71 72def test_negative_capacity_raise(simple_flow_graph): 73 G = simple_flow_graph 74 nx.set_edge_attributes(G, {("a", "b"): {"weight": 1}, ("b", "d"): {"capacity": -9}}) 75 pytest.raises(nx.NetworkXUnfeasible, nx.network_simplex, G) 76 77 78def test_no_flow_satisfying_demands(simple_no_flow_graph): 79 G = simple_no_flow_graph 80 pytest.raises(nx.NetworkXUnfeasible, nx.network_simplex, G) 81 82 83def test_sum_demands_not_zero(simple_no_flow_graph): 84 G = simple_no_flow_graph 85 nx.set_node_attributes(G, {"t": {"demand": 4}}) 86 pytest.raises(nx.NetworkXUnfeasible, nx.network_simplex, G) 87 88 89def test_google_or_tools_example(): 90 """ 91 https://developers.google.com/optimization/flow/mincostflow 92 """ 93 G = nx.DiGraph() 94 start_nodes = [0, 0, 1, 1, 1, 2, 2, 3, 4] 95 end_nodes = [1, 2, 2, 3, 4, 3, 4, 4, 2] 96 capacities = [15, 8, 20, 4, 10, 15, 4, 20, 5] 97 unit_costs = [4, 4, 2, 2, 6, 1, 3, 2, 3] 98 supplies = [20, 0, 0, -5, -15] 99 answer = 150 100 101 for i in range(len(supplies)): 102 G.add_node(i, demand=(-1) * supplies[i]) # supplies are negative of demand 103 104 for i in range(len(start_nodes)): 105 G.add_edge( 106 start_nodes[i], 107 end_nodes[i], 108 weight=unit_costs[i], 109 capacity=capacities[i], 110 ) 111 112 flowCost, flowDict = nx.network_simplex(G) 113 assert flowCost == answer 114 assert flowCost == get_flowcost_from_flowdict(G, flowDict) 115 116 117def test_google_or_tools_example2(): 118 """ 119 https://developers.google.com/optimization/flow/mincostflow 120 """ 121 G = nx.DiGraph() 122 start_nodes = [0, 0, 1, 1, 1, 2, 2, 3, 4, 3] 123 end_nodes = [1, 2, 2, 3, 4, 3, 4, 4, 2, 5] 124 capacities = [15, 8, 20, 4, 10, 15, 4, 20, 5, 10] 125 unit_costs = [4, 4, 2, 2, 6, 1, 3, 2, 3, 4] 126 supplies = [23, 0, 0, -5, -15, -3] 127 answer = 183 128 129 for i in range(len(supplies)): 130 G.add_node(i, demand=(-1) * supplies[i]) # supplies are negative of demand 131 132 for i in range(len(start_nodes)): 133 G.add_edge( 134 start_nodes[i], 135 end_nodes[i], 136 weight=unit_costs[i], 137 capacity=capacities[i], 138 ) 139 140 flowCost, flowDict = nx.network_simplex(G) 141 assert flowCost == answer 142 assert flowCost == get_flowcost_from_flowdict(G, flowDict) 143 144 145def test_large(): 146 fname = os.path.join(os.path.dirname(__file__), "netgen-2.gpickle.bz2") 147 G = nx.read_gpickle(fname) 148 flowCost, flowDict = nx.network_simplex(G) 149 assert 6749969302 == flowCost 150 assert 6749969302 == nx.cost_of_flow(G, flowDict) 151 152 153def test_simple_digraph(): 154 G = nx.DiGraph() 155 G.add_node("a", demand=-5) 156 G.add_node("d", demand=5) 157 G.add_edge("a", "b", weight=3, capacity=4) 158 G.add_edge("a", "c", weight=6, capacity=10) 159 G.add_edge("b", "d", weight=1, capacity=9) 160 G.add_edge("c", "d", weight=2, capacity=5) 161 flowCost, H = nx.network_simplex(G) 162 soln = {"a": {"b": 4, "c": 1}, "b": {"d": 4}, "c": {"d": 1}, "d": {}} 163 assert flowCost == 24 164 assert nx.min_cost_flow_cost(G) == 24 165 assert H == soln 166 167 168def test_negcycle_infcap(): 169 G = nx.DiGraph() 170 G.add_node("s", demand=-5) 171 G.add_node("t", demand=5) 172 G.add_edge("s", "a", weight=1, capacity=3) 173 G.add_edge("a", "b", weight=3) 174 G.add_edge("c", "a", weight=-6) 175 G.add_edge("b", "d", weight=1) 176 G.add_edge("d", "c", weight=-2) 177 G.add_edge("d", "t", weight=1, capacity=3) 178 pytest.raises(nx.NetworkXUnfeasible, nx.network_simplex, G) 179 180 181def test_transshipment(): 182 G = nx.DiGraph() 183 G.add_node("a", demand=1) 184 G.add_node("b", demand=-2) 185 G.add_node("c", demand=-2) 186 G.add_node("d", demand=3) 187 G.add_node("e", demand=-4) 188 G.add_node("f", demand=-4) 189 G.add_node("g", demand=3) 190 G.add_node("h", demand=2) 191 G.add_node("r", demand=3) 192 G.add_edge("a", "c", weight=3) 193 G.add_edge("r", "a", weight=2) 194 G.add_edge("b", "a", weight=9) 195 G.add_edge("r", "c", weight=0) 196 G.add_edge("b", "r", weight=-6) 197 G.add_edge("c", "d", weight=5) 198 G.add_edge("e", "r", weight=4) 199 G.add_edge("e", "f", weight=3) 200 G.add_edge("h", "b", weight=4) 201 G.add_edge("f", "d", weight=7) 202 G.add_edge("f", "h", weight=12) 203 G.add_edge("g", "d", weight=12) 204 G.add_edge("f", "g", weight=-1) 205 G.add_edge("h", "g", weight=-10) 206 flowCost, H = nx.network_simplex(G) 207 soln = { 208 "a": {"c": 0}, 209 "b": {"a": 0, "r": 2}, 210 "c": {"d": 3}, 211 "d": {}, 212 "e": {"r": 3, "f": 1}, 213 "f": {"d": 0, "g": 3, "h": 2}, 214 "g": {"d": 0}, 215 "h": {"b": 0, "g": 0}, 216 "r": {"a": 1, "c": 1}, 217 } 218 assert flowCost == 41 219 assert H == soln 220 221 222def test_digraph1(): 223 # From Bradley, S. P., Hax, A. C. and Magnanti, T. L. Applied 224 # Mathematical Programming. Addison-Wesley, 1977. 225 G = nx.DiGraph() 226 G.add_node(1, demand=-20) 227 G.add_node(4, demand=5) 228 G.add_node(5, demand=15) 229 G.add_edges_from( 230 [ 231 (1, 2, {"capacity": 15, "weight": 4}), 232 (1, 3, {"capacity": 8, "weight": 4}), 233 (2, 3, {"weight": 2}), 234 (2, 4, {"capacity": 4, "weight": 2}), 235 (2, 5, {"capacity": 10, "weight": 6}), 236 (3, 4, {"capacity": 15, "weight": 1}), 237 (3, 5, {"capacity": 5, "weight": 3}), 238 (4, 5, {"weight": 2}), 239 (5, 3, {"capacity": 4, "weight": 1}), 240 ] 241 ) 242 flowCost, H = nx.network_simplex(G) 243 soln = { 244 1: {2: 12, 3: 8}, 245 2: {3: 8, 4: 4, 5: 0}, 246 3: {4: 11, 5: 5}, 247 4: {5: 10}, 248 5: {3: 0}, 249 } 250 assert flowCost == 150 251 assert nx.min_cost_flow_cost(G) == 150 252 assert H == soln 253 254 255def test_zero_capacity_edges(): 256 """Address issue raised in ticket #617 by arv.""" 257 G = nx.DiGraph() 258 G.add_edges_from( 259 [ 260 (1, 2, {"capacity": 1, "weight": 1}), 261 (1, 5, {"capacity": 1, "weight": 1}), 262 (2, 3, {"capacity": 0, "weight": 1}), 263 (2, 5, {"capacity": 1, "weight": 1}), 264 (5, 3, {"capacity": 2, "weight": 1}), 265 (5, 4, {"capacity": 0, "weight": 1}), 266 (3, 4, {"capacity": 2, "weight": 1}), 267 ] 268 ) 269 G.nodes[1]["demand"] = -1 270 G.nodes[2]["demand"] = -1 271 G.nodes[4]["demand"] = 2 272 273 flowCost, H = nx.network_simplex(G) 274 soln = {1: {2: 0, 5: 1}, 2: {3: 0, 5: 1}, 3: {4: 2}, 4: {}, 5: {3: 2, 4: 0}} 275 assert flowCost == 6 276 assert nx.min_cost_flow_cost(G) == 6 277 assert H == soln 278 279 280def test_digon(): 281 """Check if digons are handled properly. Taken from ticket 282 #618 by arv.""" 283 nodes = [(1, {}), (2, {"demand": -4}), (3, {"demand": 4})] 284 edges = [ 285 (1, 2, {"capacity": 3, "weight": 600000}), 286 (2, 1, {"capacity": 2, "weight": 0}), 287 (2, 3, {"capacity": 5, "weight": 714285}), 288 (3, 2, {"capacity": 2, "weight": 0}), 289 ] 290 G = nx.DiGraph(edges) 291 G.add_nodes_from(nodes) 292 flowCost, H = nx.network_simplex(G) 293 soln = {1: {2: 0}, 2: {1: 0, 3: 4}, 3: {2: 0}} 294 assert flowCost == 2857140 295 296 297def test_deadend(): 298 """Check if one-node cycles are handled properly. Taken from ticket 299 #2906 from @sshraven.""" 300 G = nx.DiGraph() 301 302 G.add_nodes_from(range(5), demand=0) 303 G.nodes[4]["demand"] = -13 304 G.nodes[3]["demand"] = 13 305 306 G.add_edges_from([(0, 2), (0, 3), (2, 1)], capacity=20, weight=0.1) 307 pytest.raises(nx.NetworkXUnfeasible, nx.network_simplex, G) 308 309 310def test_infinite_capacity_neg_digon(): 311 """An infinite capacity negative cost digon results in an unbounded 312 instance.""" 313 nodes = [(1, {}), (2, {"demand": -4}), (3, {"demand": 4})] 314 edges = [ 315 (1, 2, {"weight": -600}), 316 (2, 1, {"weight": 0}), 317 (2, 3, {"capacity": 5, "weight": 714285}), 318 (3, 2, {"capacity": 2, "weight": 0}), 319 ] 320 G = nx.DiGraph(edges) 321 G.add_nodes_from(nodes) 322 pytest.raises(nx.NetworkXUnbounded, nx.network_simplex, G) 323 324 325def test_multidigraph(): 326 """Multidigraphs are acceptable.""" 327 G = nx.MultiDiGraph() 328 G.add_weighted_edges_from([(1, 2, 1), (2, 3, 2)], weight="capacity") 329 flowCost, H = nx.network_simplex(G) 330 assert flowCost == 0 331 assert H == {1: {2: {0: 0}}, 2: {3: {0: 0}}, 3: {}} 332 333 334def test_negative_selfloops(): 335 """Negative selfloops should cause an exception if uncapacitated and 336 always be saturated otherwise. 337 """ 338 G = nx.DiGraph() 339 G.add_edge(1, 1, weight=-1) 340 pytest.raises(nx.NetworkXUnbounded, nx.network_simplex, G) 341 342 G[1][1]["capacity"] = 2 343 flowCost, H = nx.network_simplex(G) 344 assert flowCost == -2 345 assert H == {1: {1: 2}} 346 347 G = nx.MultiDiGraph() 348 G.add_edge(1, 1, "x", weight=-1) 349 G.add_edge(1, 1, "y", weight=1) 350 pytest.raises(nx.NetworkXUnbounded, nx.network_simplex, G) 351 352 G[1][1]["x"]["capacity"] = 2 353 flowCost, H = nx.network_simplex(G) 354 assert flowCost == -2 355 assert H == {1: {1: {"x": 2, "y": 0}}} 356 357 358def test_bone_shaped(): 359 # From #1283 360 G = nx.DiGraph() 361 G.add_node(0, demand=-4) 362 G.add_node(1, demand=2) 363 G.add_node(2, demand=2) 364 G.add_node(3, demand=4) 365 G.add_node(4, demand=-2) 366 G.add_node(5, demand=-2) 367 G.add_edge(0, 1, capacity=4) 368 G.add_edge(0, 2, capacity=4) 369 G.add_edge(4, 3, capacity=4) 370 G.add_edge(5, 3, capacity=4) 371 G.add_edge(0, 3, capacity=0) 372 flowCost, H = nx.network_simplex(G) 373 assert flowCost == 0 374 assert H == {0: {1: 2, 2: 2, 3: 0}, 1: {}, 2: {}, 3: {}, 4: {3: 2}, 5: {3: 2}} 375 376 377def test_graphs_type_exceptions(): 378 G = nx.Graph() 379 pytest.raises(nx.NetworkXNotImplemented, nx.network_simplex, G) 380 G = nx.MultiGraph() 381 pytest.raises(nx.NetworkXNotImplemented, nx.network_simplex, G) 382 G = nx.DiGraph() 383 pytest.raises(nx.NetworkXError, nx.network_simplex, G) 384