Source Code for Module datk.tests.tests

  1  """ 
  2  Algorithm Test Suite 
  3   
  4  Tests algorithms defined in algs.py 
  5  """ 
  6   
  7  try: 
  8      from datk.core.distalgs import * 
  9      from datk.core.networks import * 
 10      from datk.core.algs import * 
 11      from datk.core.tester import * 
 12  except ImportError: 
 13      raise ImportError( 
 14  """ Imports failed\n 
 15  To run tests, execute the following commands: 
 16  $ cd ../.. 
 17  $ python -m datk.tests.tests """) 
 18  from helpers import * 
 19   
 20 -def configure_ipython(): 
 21    """ 
 22    Convenient helper function to determine if environment is IPython. 
 23   
 24    Sets matplotlib inline, if indeed in IPython 
 25    Note that drawing is only safe in IPython qtconsole with matplotlib inline 
 26     
 27    @return: True iff environment is IPython 
 28    """ 
 29    try: 
 30      __IPYTHON__ 
 31      ip = get_ipython() 
 32      ip.magic("%matplotlib inline")  
 33    except NameError: 
 34      return False 
 35    else: 
 36      return True 
 37   
 38  in_ipython = configure_ipython() 
 39   
 40  Algorithm.DEFAULT_PARAMS = {"draw":False, "verbosity" : Algorithm.QUIET} 
 41   
 42  @test 
 43 -def LCR_UNI_RING(): 
 44      r = Unidirectional_Ring(6) 
 45      LCR(r) 
 46      testLeaderElection(r) 
 47   
 48  @test 
 49 -def LCR_BI_RING(): 
 50      r = Bidirectional_Ring(6) 
 51      LCR(r) 
 52      testLeaderElection(r) 
 53   
 54  @test 
 55 -def ASYNC_LCR_UNI_RING(): 
 56      r = Unidirectional_Ring(6) 
 57      AsyncLCR(r) 
 58      testLeaderElection(r) 
 59   
 60  @test 
 61 -def ASYNC_LCR_BI_RING(): 
 62      r = Bidirectional_Ring(6) 
 63      AsyncLCR(r) 
 64      testLeaderElection(r) 
 65   
 66  @test 
 67 -def FLOODMAX_UNI_RING(): 
 68      r = Unidirectional_Ring(4) 
 69      FloodMax(r) 
 70      testLeaderElection(r) 
 71   
 72  @test 
 73 -def FLOODMAX_BI_RING(): 
 74      r = Bidirectional_Ring(4) 
 75      FloodMax(r) 
 76      testLeaderElection(r) 
 77   
 78  @test 
 79 -def FLOODMAX_BI_LINE(): 
 80      l = Bidirectional_Line(4) 
 81      FloodMax(l) 
 82      testLeaderElection(l) 
 83   
 84  @test 
 85 -def FLOODMAX_COMPLETE_GRAPH(): 
 86      g = Complete_Graph(10) 
 87      FloodMax(g) 
 88      testLeaderElection(g) 
 89   
 90  @test 
 91 -def FLOODMAX_RANDOM_GRAPH(): 
 92      g = Random_Line_Network(16) 
 93      FloodMax(g) 
 94      testLeaderElection(g) 
 95   
 96  @test 
 97 -def SYNCH_BFS(): 
 98      x = Random_Line_Network(10) 
 99      FloodMax(x) 
100      testLeaderElection(x) 
101   
102      SynchBFS(x) 
103      testBFS(x) 
104   
105  @test 
106 -def SYNCH_BFS_ACK(): 
107      x = Bidirectional_Line(6, lambda t:t) 
108   
109      FloodMax(x) 
110      testLeaderElection(x) 
111   
112      SynchBFSAck(x) 
113      testBFSWithChildren(x) 
114   
115  @test 
116 -def SYNCH_CONVERGE_HEIGHT(): 
117      x = Random_Line_Network(10) 
118   
119      FloodMax(x) 
120      testLeaderElection(x) 
121   
122      SynchBFS(x) 
123      testBFS(x) 
124   
125      SynchConvergeHeight(x) 
126   
127  @test 
128 -def SYNCH_BROADCAST_HEIGHT(): 
129      x = Random_Line_Network(10) 
130   
131      FloodMax(x) 
132      testLeaderElection(x) 
133   
134      SynchBFSAck(x) 
135      testBFSWithChildren(x) 
136   
137      SynchConvergeHeight(x) 
138   
139      SynchBroadcast(x, {"attr":"height", "draw":in_ipython, "verbosity" : Algorithm.QUIET}) 
140      testBroadcast(x, 'height') 
141   
142  @test 
143 -def ASYNCH_BROADCAST_HEIGHT(): 
144      x = Random_Line_Network(10) 
145   
146      FloodMax(x) 
147      testLeaderElection(x) 
148   
149      SynchBFSAck(x) 
150      testBFSWithChildren(x) 
151   
152      AsynchConvergeHeight(x) 
153   
154      SynchBroadcast(x, {"attr":"height", "draw":in_ipython, "verbosity" : Algorithm.QUIET}) 
155      testBroadcast(x, 'height') 
156   
157  @test 
158 -def send_receive_msgs(): 
159      A = LCR() 
160      a1 = Message(A) 
161      a2 = Message(A) 
162      a3 = Message(A) 
163   
164      B = LCR() 
165      b1 = Message(B) 
166      b2 = Message(B) 
167   
168      x = Bidirectional_Ring(4, lambda p:p) 
169      assert x[0].get_msgs(A) == [] 
170      x[0].send_msg(a1) 
171      x[0].send_msg(a2) 
172      assert a1.author == x[0] and a2.author == x[0] 
173      x[2].send_msg(a3) 
174      assert a3.author == x[2] 
175      x[0].send_msg(b1) 
176      assert b1.author == x[0] 
177      x[2].send_msg(b2) 
178      assert b2.author == x[2] 
179   
180      assert x[1].get_msgs(B) == [b1,b2] 
181      assert x[1].get_msgs(A, x[0]) == [a1, a2] 
182      assert x[1].get_msgs(A, x[0]) == [] 
183      assert x[1].get_msgs(A) == [a3] 
184      assert x[1].get_msgs(A) == [] 
185   
186  @test 
187 -def SYNCH_DO_NOTHING(): 
188      x = Random_Line_Network(5) 
189      state = x.state() 
190      assert Do_Nothing(x).message_count == 0 
191      assert state == x.state() 
192   
193  @test 
194 -def COMPOSE_SYNCH_LCR_AND_DO_NOTHING(): 
195      x = Unidirectional_Ring(5) 
196      x1 = x.clone() 
197   
198      A = LCR() 
199      A(x) 
200      testLeaderElection(x) 
201   
202      C = Compose(LCR(), Do_Nothing()) 
203      C(x1) 
204      testLeaderElection(x1) 
205   
206      assert C.message_count == A.message_count, "Wrong message count" 
207   
208  @test 
209 -def COMPOSE_SYNCH_LCR(): 
210      x = Unidirectional_Ring(10) 
211      x1 = x.clone() 
212      x2 = x.clone() 
213   
214      A = LCR() 
215      A(x) 
216      testLeaderElection(x) 
217   
218      B = Compose(LCR(name="B1"), LCR(name="B2")) 
219      B(x1) 
220      testLeaderElection(x1) 
221   
222      C = Compose(Compose(LCR(), LCR()), LCR()) 
223      C(x2) 
224      testLeaderElection(x2) 
225   
226      assert B.message_count == 2*A.message_count, "Compose LCR LCR wrong message count" 
227      assert C.message_count == 3*A.message_count, "Compose LCR LCR LCR wrong message count" 
228   
229  @test 
230 -def CHAIN_BROADCAST_HEIGHT(): 
231      fm = FloodMax() 
232      bfs = SynchBFSAck() 
233      converge = SynchConvergeHeight() 
234      broadcast = SynchBroadcast(params ={"attr":"height"}) 
235   
236      A = Chain(Chain(fm, bfs), Chain(converge, broadcast)) 
237      x = Random_Line_Network(10) 
238      A(x) 
239      testLeaderElection(x) 
240      testBFSWithChildren(x) 
241      testBroadcast(x, 'height') 
242   
243  @test 
244 -def SYNCH_LUBY_MIS_BI_RING(): 
245      x = Bidirectional_Ring(10, lambda t:t) 
246      SynchLubyMIS(x) 
247      testLubyMIS(x) 
248   
249  @test 
250 -def SYNCH_LUBY_MIS(): 
251      x = Random_Line_Network(10) 
252      SynchLubyMIS(x) 
253      testLubyMIS(x) 
254   
255  summarize() 
256