Source Code for Module datk.tests.helpers

 1  """ 
 2  Helper functions for tests in tests.py 
 3  """ 
 4  from datk.core.distalgs import Process 
 5   
 6 -def testLeaderElection( 
 7      network, 
 8      isLeader = lambda p: "status" in p.state and p.state["status"]=="leader", 
 9      isNonleader = lambda p: "status" in p.state and p.state["status"]=="non-leader" 
10      ): 
11      """Asserts that exactly one Process is Leader, and all other processes are Non-Leader""" 
12   
13      assert sum([isLeader(p) for p in network]) == 1 , "Leader Election Failed" 
14      assert sum([isNonleader(p) for p in network]) == len(network)-1, "Leader Election Failed" 
15   
16 -def testBroadcast(network, attr): 
17      """Asserts that p.state[attr] is identical for all processes p""" 
18      for p in network: 
19          assert attr in p.state 
20      assert len(set([p.state[attr] for p in network])) == 1, "Broadcasting " + attr + " failed." 
21   
22 -def testBFS(network): 
23      """Asserts that every Process, p, knows 'parent', and there  
24       exists exactly one Process where 'parent' is None""" 
25      found_root = False 
26      for p in network: 
27          assert 'parent' in p.state, "BFS Failed. state['parent'] not found." 
28          if p.state['parent'] is None: 
29              if found_root: 
30                  assert False, "BFS failed. No unique root node" 
31              else: 
32                  found_root = True 
33          else: 
34              assert isinstance(p.state['parent'], Process), "BFS FAILED" 
35   
36 -def testBFSWithChildren(network): 
37      """Asserts that every Process, p, knows 'parent' and 'children', and there 
38      exists exactly one Process where 'parent' is None""" 
39      found_root = False 
40      for p in network: 
41          assert 'parent' in p.state, "BFS Failed. state['parent'] not found." 
42          if p.state['parent'] is None: 
43              if found_root: 
44                  assert False, "BFS failed. No unique root node" 
45              else: 
46                  found_root = True 
47          else: 
48              assert isinstance(p.state['parent'], Process), "BFS FAILED" 
49              assert p in p.state['parent'].state['children'], "BFS FAILED" 
50   
51 -def testLubyMIS(network): 
52      """Asserts that every process knows a boolean value, 'MIS', and that the Processes 
53      where 'MIS' is True form a set that is both independent and maximal.""" 
54      for process in network: 
55          assert 'MIS' in process.state, "'MIS' not in Process state" 
56          assert isinstance(process.state['MIS'], bool) 
57          if process.state['MIS'] == True: 
58              assert not any([nbr.state['MIS'] for nbr in process.out_nbrs]), 'MIS not independent' 
59          if process.state['MIS'] == False: 
60              assert any([nbr.state['MIS'] for nbr in process.out_nbrs]), 'MIS not maximal' 
61