1 # -*- coding: utf-8 -*- 2 """Max flow algorithm test suite. 3 4 Run with nose: nosetests -v test_max_flow.py 5 """ 6 7 __author__ = """Loc Sguin-C. <loicseguin (at] gmail.com>""" 8 # Copyright (C) 2010 Loc Sguin-C. <loicseguin (at] gmail.com> 9 # All rights reserved. 10 # BSD license. 11 12 13 import networkx as nx 14 from nose.tools import * 15 16 def compare_flows(G, s, t, solnFlows, solnValue): 17 flowValue, flowDict = nx.ford_fulkerson(G, s, t) 18 assert_equal(flowValue, solnValue) 19 assert_equal(flowDict, solnFlows) 20 assert_equal(nx.min_cut(G, s, t), solnValue) 21 assert_equal(nx.max_flow(G, s, t), solnValue) 22 assert_equal(nx.ford_fulkerson_flow(G, s, t), solnFlows) 23 24 25 class TestMaxflow: 26 def test_graph1(self): 27 # Trivial undirected graph 28 G = nx.Graph() 29 G.add_edge(1,2, capacity = 1.0) 30 31 solnFlows = {1: {2: 1.0}, 32 2: {1: 1.0}} 33 34 compare_flows(G, 1, 2, solnFlows, 1.0) 35 36 def test_graph2(self): 37 # A more complex undirected graph 38 # adapted from www.topcoder.com/tc?module=Statc&d1=tutorials&d2=maxFlow 39 G = nx.Graph() 40 G.add_edge('x','a', capacity = 3.0) 41 G.add_edge('x','b', capacity = 1.0) 42 G.add_edge('a','c', capacity = 3.0) 43 G.add_edge('b','c', capacity = 5.0) 44 G.add_edge('b','d', capacity = 4.0) 45 G.add_edge('d','e', capacity = 2.0) 46 G.add_edge('c','y', capacity = 2.0) 47 G.add_edge('e','y', capacity = 3.0) 48 49 H = {'x': {'a': 3, 'b': 1}, 50 'a': {'c': 3, 'x': 3}, 51 'b': {'c': 1, 'd': 2, 'x': 1}, 52 'c': {'a': 3, 'b': 1, 'y': 2}, 53 'd': {'b': 2, 'e': 2}, 54 'e': {'d': 2, 'y': 2}, 55 'y': {'c': 2, 'e': 2}} 56 57 compare_flows(G, 'x', 'y', H, 4.0) 58 59 def test_digraph1(self): 60 # The classic directed graph example 61 G = nx.DiGraph() 62 G.add_edge('a','b', capacity = 1000.0) 63 G.add_edge('a','c', capacity = 1000.0) 64 G.add_edge('b','c', capacity = 1.0) 65 G.add_edge('b','d', capacity = 1000.0) 66 G.add_edge('c','d', capacity = 1000.0) 67 68 H = {'a': {'b': 1000.0, 'c': 1000.0}, 69 'b': {'c': 0, 'd': 1000.0}, 70 'c': {'d': 1000.0}, 71 'd': {}} 72 73 compare_flows(G, 'a', 'd', H, 2000.0) 74 75 # An example in which some edges end up with zero flow. 76 G = nx.DiGraph() 77 G.add_edge('s', 'b', capacity = 2) 78 G.add_edge('s', 'c', capacity = 1) 79 G.add_edge('c', 'd', capacity = 1) 80 G.add_edge('d', 'a', capacity = 1) 81 G.add_edge('b', 'a', capacity = 2) 82 G.add_edge('a', 't', capacity = 2) 83 84 H = {'s': {'b': 2, 'c': 0}, 85 'c': {'d': 0}, 86 'd': {'a': 0}, 87 'b': {'a': 2}, 88 'a': {'t': 2}, 89 't': {}} 90 91 compare_flows(G, 's', 't', H, 2) 92 93 def test_digraph2(self): 94 # A directed graph example from Cormen et al. 95 G = nx.DiGraph() 96 G.add_edge('s','v1', capacity = 16.0) 97 G.add_edge('s','v2', capacity = 13.0) 98 G.add_edge('v1','v2', capacity = 10.0) 99 G.add_edge('v2','v1', capacity = 4.0) 100 G.add_edge('v1','v3', capacity = 12.0) 101 G.add_edge('v3','v2', capacity = 9.0) 102 G.add_edge('v2','v4', capacity = 14.0) 103 G.add_edge('v4','v3', capacity = 7.0) 104 G.add_edge('v3','t', capacity = 20.0) 105 G.add_edge('v4','t', capacity = 4.0) 106 107 H = {'s': {'v1': 12.0, 'v2': 11.0}, 108 'v2': {'v1': 0, 'v4': 11.0}, 109 'v1': {'v2': 0, 'v3': 12.0}, 110 'v3': {'v2': 0, 't': 19.0}, 111 'v4': {'v3': 7.0, 't': 4.0}, 112 't': {}} 113 114 compare_flows(G, 's', 't', H, 23.0) 115 116 def test_digraph3(self): 117 # A more complex directed graph 118 # from www.topcoder.com/tc?module=Statc&d1=tutorials&d2=maxFlow 119 G = nx.DiGraph() 120 G.add_edge('x','a', capacity = 3.0) 121 G.add_edge('x','b', capacity = 1.0) 122 G.add_edge('a','c', capacity = 3.0) 123 G.add_edge('b','c', capacity = 5.0) 124 G.add_edge('b','d', capacity = 4.0) 125 G.add_edge('d','e', capacity = 2.0) 126 G.add_edge('c','y', capacity = 2.0) 127 G.add_edge('e','y', capacity = 3.0) 128 129 H = {'x': {'a': 2.0, 'b': 1.0}, 130 'a': {'c': 2.0}, 131 'b': {'c': 0, 'd': 1.0}, 132 'c': {'y': 2.0}, 133 'd': {'e': 1.0}, 134 'e': {'y': 1.0}, 135 'y': {}} 136 137 compare_flows(G, 'x', 'y', H, 3.0) 138 139 def test_optional_capacity(self): 140 # Test optional capacity parameter. 141 G = nx.DiGraph() 142 G.add_edge('x','a', spam = 3.0) 143 G.add_edge('x','b', spam = 1.0) 144 G.add_edge('a','c', spam = 3.0) 145 G.add_edge('b','c', spam = 5.0) 146 G.add_edge('b','d', spam = 4.0) 147 G.add_edge('d','e', spam = 2.0) 148 G.add_edge('c','y', spam = 2.0) 149 G.add_edge('e','y', spam = 3.0) 150 151 solnFlows = {'x': {'a': 2.0, 'b': 1.0}, 152 'a': {'c': 2.0}, 153 'b': {'c': 0, 'd': 1.0}, 154 'c': {'y': 2.0}, 155 'd': {'e': 1.0}, 156 'e': {'y': 1.0}, 157 'y': {}} 158 solnValue = 3.0 159 s = 'x' 160 t = 'y' 161 162 flowValue, flowDict = nx.ford_fulkerson(G, s, t, capacity = 'spam') 163 assert_equal(flowValue, solnValue) 164 assert_equal(flowDict, solnFlows) 165 assert_equal(nx.min_cut(G, s, t, capacity = 'spam'), solnValue) 166 assert_equal(nx.max_flow(G, s, t, capacity = 'spam'), solnValue) 167 assert_equal(nx.ford_fulkerson_flow(G, s, t, capacity = 'spam'), 168 solnFlows) 169 170 def test_digraph_infcap_edges(self): 171 # DiGraph with infinite capacity edges 172 G = nx.DiGraph() 173 G.add_edge('s', 'a') 174 G.add_edge('s', 'b', capacity = 30) 175 G.add_edge('a', 'c', capacity = 25) 176 G.add_edge('b', 'c', capacity = 12) 177 G.add_edge('a', 't', capacity = 60) 178 G.add_edge('c', 't') 179 180 H = {'s': {'a': 85, 'b': 12}, 181 'a': {'c': 25, 't': 60}, 182 'b': {'c': 12}, 183 'c': {'t': 37}, 184 't': {}} 185 186 compare_flows(G, 's', 't', H, 97) 187 188 # DiGraph with infinite capacity digon 189 G = nx.DiGraph() 190 G.add_edge('s', 'a', capacity = 85) 191 G.add_edge('s', 'b', capacity = 30) 192 G.add_edge('a', 'c') 193 G.add_edge('c', 'a') 194 G.add_edge('b', 'c', capacity = 12) 195 G.add_edge('a', 't', capacity = 60) 196 G.add_edge('c', 't', capacity = 37) 197 198 H = {'s': {'a': 85, 'b': 12}, 199 'a': {'c': 25, 't': 60}, 200 'c': {'a': 0, 't': 37}, 201 'b': {'c': 12}, 202 't': {}} 203 204 compare_flows(G, 's', 't', H, 97) 205 206 207 def test_digraph_infcap_path(self): 208 # Graph with infinite capacity (s, t)-path 209 G = nx.DiGraph() 210 G.add_edge('s', 'a') 211 G.add_edge('s', 'b', capacity = 30) 212 G.add_edge('a', 'c') 213 G.add_edge('b', 'c', capacity = 12) 214 G.add_edge('a', 't', capacity = 60) 215 G.add_edge('c', 't') 216 217 assert_raises(nx.NetworkXUnbounded, 218 nx.ford_fulkerson, G, 's', 't') 219 assert_raises(nx.NetworkXUnbounded, 220 nx.max_flow, G, 's', 't') 221 assert_raises(nx.NetworkXUnbounded, 222 nx.ford_fulkerson_flow, G, 's', 't') 223 assert_raises(nx.NetworkXUnbounded, 224 nx.min_cut, G, 's', 't') 225 226 def test_graph_infcap_edges(self): 227 # Undirected graph with infinite capacity edges 228 G = nx.Graph() 229 G.add_edge('s', 'a') 230 G.add_edge('s', 'b', capacity = 30) 231 G.add_edge('a', 'c', capacity = 25) 232 G.add_edge('b', 'c', capacity = 12) 233 G.add_edge('a', 't', capacity = 60) 234 G.add_edge('c', 't') 235 236 H = {'s': {'a': 85, 'b': 12}, 237 'a': {'c': 25, 's': 85, 't': 60}, 238 'b': {'c': 12, 's': 12}, 239 'c': {'a': 25, 'b': 12, 't': 37}, 240 't': {'a': 60, 'c': 37}} 241 242 compare_flows(G, 's', 't', H, 97) 243 244 def test_digraph4(self): 245 # From ticket #429 by mfrasca. 246 G = nx.DiGraph() 247 G.add_edge('s', 'a', capacity = 2) 248 G.add_edge('s', 'b', capacity = 2) 249 G.add_edge('a', 'b', capacity = 5) 250 G.add_edge('a', 't', capacity = 1) 251 G.add_edge('b', 'a', capacity = 1) 252 G.add_edge('b', 't', capacity = 3) 253 flowSoln = {'a': {'b': 1, 't': 1}, 254 'b': {'a': 0, 't': 3}, 255 's': {'a': 2, 'b': 2}, 256 't': {}} 257 compare_flows(G, 's', 't', flowSoln, 4) 258 259 260 def test_disconnected(self): 261 G = nx.Graph() 262 G.add_weighted_edges_from([(0,1,1),(1,2,1),(2,3,1)],weight='capacity') 263 G.remove_node(1) 264 assert_equal(nx.max_flow(G,0,3),0) 265 266 def test_source_target_not_in_graph(self): 267 G = nx.Graph() 268 G.add_weighted_edges_from([(0,1,1),(1,2,1),(2,3,1)],weight='capacity') 269 G.remove_node(0) 270 assert_raises(nx.NetworkXError,nx.max_flow,G,0,3) 271 G.add_weighted_edges_from([(0,1,1),(1,2,1),(2,3,1)],weight='capacity') 272 G.remove_node(3) 273 assert_raises(nx.NetworkXError,nx.max_flow,G,0,3) 274
