blob: 179e691105c9dca73f1cc9ab32e403d4cb74e294 [file] [log] [blame]
sgovindacc736782017-05-02 20:06:37 +05301from unittest import TestCase, main
2from connection_mgr import ConnectionManager
3
4class TestConection_mgr(TestCase):
5
6 def gen_endpoints(self):
7 consul_endpoint = "localhost:8500"
8 voltha_endpoint= "localhost:8880"
9 controller_endpoints = ["localhost:6633","localhost:6644","localhost:6655"]
10 return (consul_endpoint,voltha_endpoint,controller_endpoints)
11
12 def gen_devices(self):
13 device =lambda: None
14 device.id = "1"
15 device.datapath_id = 1
16 device.desc = '{mfr_desc: "cord porject" hw_desc: "simualted pon" sw_desc: "simualted pon"\
17 serial_num: "2f150d56afa2405eba3ba24e33ce8df9" dp_desc: "n/a"}'
18 device.switch_features = '{ n_buffers: 256 n_tables: 2 capabilities: 15 }'
19 device.root_device_id = "a245bd8bb8b8"
20 devices = [device]
21 return devices,device
22
23 def gen_packet_in(self):
24 packet_in = 1
25 return packet_in
26
27 def test_connection_mgr_init(self):
28 consul_endpoint,voltha_endpoint,controller_endpoints = self.gen_endpoints()
29 test_connection_init = ConnectionManager(consul_endpoint, voltha_endpoint, controller_endpoints)
30 self.assertEqual(test_connection_init.consul_endpoint,consul_endpoint)
sathishg23ac35c2017-10-14 03:21:43 +053031 self.assertEqual(test_connection_init.vcore_endpoint, voltha_endpoint)
sgovindacc736782017-05-02 20:06:37 +053032 self.assertEqual(test_connection_init.controller_endpoints, controller_endpoints)
33
34 def test_resolve_endpoint(self):
35 consul_endpoint, voltha_endpoint, controller_endpoints = self.gen_endpoints()
36 test_connection_init = ConnectionManager(consul_endpoint, voltha_endpoint, controller_endpoints)
37 host,port = test_connection_init.resolve_endpoint(endpoint=consul_endpoint)
38 assert isinstance(port, int)
39 assert isinstance(host, basestring)
40
41 def test_refresh_agent_connections(self):
42 consul_endpoint, voltha_endpoint, controller_endpoints = self.gen_endpoints()
43 devices,device = self.gen_devices()
44 test_connection_init = ConnectionManager(consul_endpoint, voltha_endpoint, controller_endpoints)
45 test_connection_init.refresh_agent_connections(devices)
46
47 def test_create_agent(self):
48 consul_endpoint, voltha_endpoint, controller_endpoints = self.gen_endpoints()
49 devices,device = self.gen_devices()
50 test_connection_init = ConnectionManager(consul_endpoint, voltha_endpoint, controller_endpoints)
51 test_connection_init.create_agent(device)
52
53 def test_delete_agent(self):
54 consul_endpoint, voltha_endpoint, controller_endpoints = self.gen_endpoints()
55 devices,device = self.gen_devices()
56 test_connection_init = ConnectionManager(consul_endpoint, voltha_endpoint, controller_endpoints)
57 test_connection_init.create_agent(device)
58 with self.assertRaises(Exception) as context:
59 test_connection_init.delete_agent(device.datapath_id)
60 print context.exception
sathishg23ac35c2017-10-14 03:21:43 +053061 self.assertTrue('\'NoneType\' object has no attribute \'disconnect\'' in str(context.exception))
sgovindacc736782017-05-02 20:06:37 +053062
63 def test_forward_packet_in(self):
64 consul_endpoint, voltha_endpoint, controller_endpoints = self.gen_endpoints()
65 devices,device = self.gen_devices()
66 packet_in = self.gen_packet_in()
67 test_connection_init = ConnectionManager(consul_endpoint, voltha_endpoint, controller_endpoints)
68 test_connection_init.create_agent(device)
69 test_connection_init.forward_packet_in(device.id, packet_in)
70
71if __name__ == '__main__':
sathishg23ac35c2017-10-14 03:21:43 +053072 main()