| sgovinda | cc73678 | 2017-05-02 20:06:37 +0530 | [diff] [blame] | 1 | from unittest import TestCase, main |
| 2 | from connection_mgr import ConnectionManager |
| 3 | |
| 4 | class 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) |
| 31 | self.assertEqual(test_connection_init.voltha_endpoint, voltha_endpoint) |
| 32 | 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 |
| 61 | self.assertTrue('\'NoneType\' object has no attribute \'disconnect\'' in context.exception) |
| 62 | |
| 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 | |
| 71 | if __name__ == '__main__': |
| 72 | main() |