X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F2.2.1%2Ftest_otn_renderer.py;h=69eee8579fc0cafca609152608df71ec4e7fc0c9;hb=f14d37954ffcf069743b1c1b8f52811606667ed8;hp=cfd90383010295a8e6a6e0eef401baa1e4be68d3;hpb=79249b3cbb0c888563fa049737121137541adc81;p=transportpce.git diff --git a/tests/transportpce_tests/2.2.1/test_otn_renderer.py b/tests/transportpce_tests/2.2.1/test_otn_renderer.py index cfd903830..69eee8579 100644 --- a/tests/transportpce_tests/2.2.1/test_otn_renderer.py +++ b/tests/transportpce_tests/2.2.1/test_otn_renderer.py @@ -22,22 +22,20 @@ import logging import test_utils +def extract_a_from_b(a, b): + return dict([(i, b[i]) for i in b.keys() if i in a.keys()]) + + class TransportPCEtesting(unittest.TestCase): - honeynode_process1 = None + sim_process1 = None odl_process = None restconf_baseurl = "http://localhost:8181/restconf" @classmethod def setUpClass(cls): - print("starting honeynode1...") - cls.honeynode_process1 = test_utils.start_spdra_honeynode() - time.sleep(30) - - print("starting opendaylight...") cls.odl_process = test_utils.start_tpce() - time.sleep(60) - print("opendaylight started") + cls.sim_process1 = test_utils.start_sim('spdrav2') @classmethod def tearDownClass(cls): @@ -46,11 +44,11 @@ class TransportPCEtesting(unittest.TestCase): child.wait() cls.odl_process.send_signal(signal.SIGINT) cls.odl_process.wait() - for child in psutil.Process(cls.honeynode_process1.pid).children(): + for child in psutil.Process(cls.sim_process1.pid).children(): child.send_signal(signal.SIGINT) child.wait() - cls.honeynode_process1.send_signal(signal.SIGINT) - cls.honeynode_process1.wait() + cls.sim_process1.send_signal(signal.SIGINT) + cls.sim_process1.wait() def setUp(self): time.sleep(5) @@ -64,7 +62,7 @@ class TransportPCEtesting(unittest.TestCase): "netconf-node-topology:username": "admin", "netconf-node-topology:password": "admin", "netconf-node-topology:host": "127.0.0.1", - "netconf-node-topology:port": "17845", + "netconf-node-topology:port": test_utils.sims['spdrav2']['port'], "netconf-node-topology:tcp-only": "false", "netconf-node-topology:pass-through": {}}]} headers = {'content-type': 'application/json'} @@ -100,8 +98,10 @@ class TransportPCEtesting(unittest.TestCase): 'org-openroadm-port-types:if-10GE'], 'supporting-port': 'CP1-SFP4-P1', 'supporting-circuit-pack-name': 'CP1-SFP4', - 'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional', - 'port-qual': 'xpdr-client'}, + 'logical-connection-point': 'XPDR1-CLIENT1', + 'port-direction': 'bidirectional', + 'port-qual': 'xpdr-client', + 'lcp-hash-val': '8b3efff522736722500b5e68fb6e696e'}, res['mapping']) def test_03_get_portmapping_NETWORK1(self): @@ -122,7 +122,8 @@ class TransportPCEtesting(unittest.TestCase): "port-direction": "bidirectional", "port-qual": "xpdr-network", "supporting-circuit-pack-name": "CP1-CFP0", - "xponder-type": "mpdr"}, + "xponder-type": "mpdr", + 'lcp-hash-val': '1021db8d2affe7386705c438c67ea21f'}, res['mapping']) def test_04_service_path_create_OCH_OTU4(self): @@ -167,7 +168,8 @@ class TransportPCEtesting(unittest.TestCase): "port-direction": "bidirectional", "port-qual": "xpdr-network", "supporting-circuit-pack-name": "CP1-CFP0", - "xponder-type": "mpdr"}, + "xponder-type": "mpdr", + "lcp-hash-val": "1021db8d2affe7386705c438c67ea21f"}, res['mapping']) def test_06_check_interface_och(self): @@ -180,10 +182,24 @@ class TransportPCEtesting(unittest.TestCase): "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() + + input_dict = {'name': 'XPDR1-NETWORK1-1', + 'administrative-state': 'inService', + 'supporting-circuit-pack-name': 'CP1-CFP0', + 'type': 'org-openroadm-interfaces:opticalChannel', + 'supporting-port': 'CP1-CFP0-P1' + } + # assertDictContainsSubset is deprecated + ''' self.assertDictContainsSubset({'name': 'XPDR1-NETWORK1-1', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', 'type': 'org-openroadm-interfaces:opticalChannel', 'supporting-port': 'CP1-CFP0-P1'}, res['interface'][0]) + ''' + self.assertDictEqual(input_dict, + extract_a_from_b(input_dict, + res['interface'][0]) + ) self.assertDictEqual( {u'frequency': 196.1, u'rate': u'org-openroadm-common-types:R100G', u'transmit-power': -5}, @@ -199,10 +215,24 @@ class TransportPCEtesting(unittest.TestCase): "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() + input_dict = {'name': 'XPDR1-NETWORK1-OTU', + 'administrative-state': 'inService', + 'supporting-circuit-pack-name': 'CP1-CFP0', + 'supporting-interface': 'XPDR1-NETWORK1-1', + 'type': 'org-openroadm-interfaces:otnOtu', + 'supporting-port': 'CP1-CFP0-P1'} + + # assertDictContainsSubset is deprecated + ''' self.assertDictContainsSubset({'name': 'XPDR1-NETWORK1-OTU', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', 'supporting-interface': 'XPDR1-NETWORK1-1', 'type': 'org-openroadm-interfaces:otnOtu', 'supporting-port': 'CP1-CFP0-P1'}, res['interface'][0]) + ''' + self.assertDictEqual(input_dict, extract_a_from_b(input_dict, + res['interface'][0]) + ) + self.assertDictEqual( {u'rate': u'org-openroadm-otn-common-types:OTU4', u'fec': u'scfec'}, @@ -242,15 +272,17 @@ class TransportPCEtesting(unittest.TestCase): res = response.json() self.assertIn( {"logical-connection-point": "XPDR1-NETWORK1", - "supporting-port": "CP1-CFP0-P1", - "supported-interface-capability": [ - "org-openroadm-port-types:if-OCH-OTU4-ODU4" - ], - "port-direction": "bidirectional", - "port-qual": "xpdr-network", - "supporting-circuit-pack-name": "CP1-CFP0", - "xponder-type": "mpdr", - "supporting-odu4": "XPDR1-NETWORK1-ODU4"}, + "supporting-port": "CP1-CFP0-P1", + "supported-interface-capability": [ + "org-openroadm-port-types:if-OCH-OTU4-ODU4" + ], + "port-direction": "bidirectional", + "port-qual": "xpdr-network", + "supporting-circuit-pack-name": "CP1-CFP0", + "xponder-type": "mpdr", + "supporting-odu4": "XPDR1-NETWORK1-ODU4", + "lcp-hash-val": "1021db8d2affe7386705c438c67ea21f" + }, res['mapping']) def test_10_check_interface_ODU4(self): @@ -263,6 +295,23 @@ class TransportPCEtesting(unittest.TestCase): "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() + input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU4', 'administrative-state': 'inService', + 'supporting-circuit-pack-name': 'CP1-CFP0', 'supporting-interface': 'XPDR1-NETWORK1-OTU', + 'type': 'org-openroadm-interfaces:otnOdu', + 'supporting-port': 'CP1-CFP0-P1'} + input_dict_2 = {'odu-function': 'org-openroadm-otn-common-types:ODU-TTP', + 'rate': 'org-openroadm-otn-common-types:ODU4'} + + self.assertDictEqual(input_dict_1, extract_a_from_b(input_dict_1, + res['interface'][0]) + ) + self.assertDictEqual(input_dict_2, + extract_a_from_b(input_dict_2, + res['interface'][0][ + 'org-openroadm-otn-odu-interfaces:odu']) + + ) + ''' self.assertDictContainsSubset({'name': 'XPDR1-NETWORK1-ODU4', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', 'supporting-interface': 'XPDR1-NETWORK1-OTU', 'type': 'org-openroadm-interfaces:otnOdu', @@ -271,6 +320,7 @@ class TransportPCEtesting(unittest.TestCase): {'odu-function': 'org-openroadm-otn-common-types:ODU-TTP', 'rate': 'org-openroadm-otn-common-types:ODU4'}, res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) + ''' self.assertDictEqual( {u'payload-type': u'21', u'exp-payload-type': u'21'}, res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) @@ -314,10 +364,22 @@ class TransportPCEtesting(unittest.TestCase): "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() + input_dict = {'name': 'XPDR1-CLIENT1-ETHERNET10G', + 'administrative-state': 'inService', + 'supporting-circuit-pack-name': 'CP1-SFP4', + 'type': 'org-openroadm-interfaces:ethernetCsmacd', + 'supporting-port': 'CP1-SFP4-P1' + } + + ''' self.assertDictContainsSubset({'name': 'XPDR1-CLIENT1-ETHERNET10G', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-SFP4', 'type': 'org-openroadm-interfaces:ethernetCsmacd', 'supporting-port': 'CP1-SFP4-P1'}, res['interface'][0]) + ''' + self.assertDictEqual(input_dict, extract_a_from_b(input_dict, + res['interface'][0]) + ) self.assertDictEqual( {u'speed': 10000}, res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']) @@ -332,6 +394,27 @@ class TransportPCEtesting(unittest.TestCase): "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() + + input_dict_1 = {'name': 'XPDR1-CLIENT1-ODU2e-service1', + 'administrative-state': 'inService', + 'supporting-circuit-pack-name': 'CP1-SFP4', + 'supporting-interface': 'XPDR1-CLIENT1-ETHERNET10G', + 'type': 'org-openroadm-interfaces:otnOdu', + 'supporting-port': 'CP1-SFP4-P1'} + input_dict_2 = { + 'odu-function': 'org-openroadm-otn-common-types:ODU-TTP-CTP', + 'rate': 'org-openroadm-otn-common-types:ODU2e', + 'monitoring-mode': 'terminated'} + + self.assertDictEqual(input_dict_1, extract_a_from_b(input_dict_1, + res['interface'][0]) + ) + self.assertDictEqual(input_dict_2, + extract_a_from_b(input_dict_2, res['interface'][0][ + 'org-openroadm-otn-odu-interfaces:odu']) + ) + + ''' self.assertDictContainsSubset({'name': 'XPDR1-CLIENT1-ODU2e-service1', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-SFP4', 'supporting-interface': 'XPDR1-CLIENT1-ETHERNET10G', @@ -341,6 +424,7 @@ class TransportPCEtesting(unittest.TestCase): 'odu-function': 'org-openroadm-otn-common-types:ODU-TTP-CTP', 'rate': 'org-openroadm-otn-common-types:ODU2e', 'monitoring-mode': 'terminated'}, res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']) + ''' self.assertDictEqual( {u'payload-type': u'03', u'exp-payload-type': u'03'}, res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu']) @@ -355,6 +439,35 @@ class TransportPCEtesting(unittest.TestCase): "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() + input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU2e-service1', 'administrative-state': 'inService', + 'supporting-circuit-pack-name': 'CP1-CFP0', + 'supporting-interface': 'XPDR1-NETWORK1-ODU4', + 'type': 'org-openroadm-interfaces:otnOdu', + 'supporting-port': 'CP1-CFP0-P1'} + input_dict_2 = { + 'odu-function': 'org-openroadm-otn-common-types:ODU-CTP', + 'rate': 'org-openroadm-otn-common-types:ODU2e', + 'monitoring-mode': 'monitored'} + + input_dict_3 = {'trib-port-number': 1} + + self.assertDictEqual(input_dict_1, extract_a_from_b(input_dict_1, + res['interface'][0]) + ) + + self.assertDictEqual(input_dict_2, + extract_a_from_b(input_dict_2, + res['interface'][0][ + 'org-openroadm-otn-odu-interfaces:odu'] + )) + + self.assertDictEqual(input_dict_3, + extract_a_from_b(input_dict_3, + res['interface'][0][ + 'org-openroadm-otn-odu-interfaces:odu'][ + 'parent-odu-allocation'])) + + ''' self.assertDictContainsSubset({'name': 'XPDR1-NETWORK1-ODU2e-service1', 'administrative-state': 'inService', 'supporting-circuit-pack-name': 'CP1-CFP0', 'supporting-interface': 'XPDR1-NETWORK1-ODU4', @@ -367,8 +480,11 @@ class TransportPCEtesting(unittest.TestCase): self.assertDictContainsSubset( {'trib-port-number': 1}, res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation']) + ''' self.assertIn(1, - res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['parent-odu-allocation']['trib-slots']) + res['interface'][0][ + 'org-openroadm-otn-odu-interfaces:odu'][ + 'parent-odu-allocation']['trib-slots']) def test_15_check_ODU2E_connection(self): url = ("{}/config/network-topology:network-topology/topology/topology-netconf/" @@ -380,10 +496,21 @@ class TransportPCEtesting(unittest.TestCase): "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() + input_dict_1 = { + 'connection-name': + 'XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1', + 'direction': 'bidirectional' + } + + self.assertDictEqual(input_dict_1, + extract_a_from_b(input_dict_1, + res['odu-connection'][0])) + ''' self.assertDictContainsSubset({ 'connection-name': 'XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1', 'direction': 'bidirectional'}, res['odu-connection'][0]) + ''' self.assertDictEqual({u'dst-if': u'XPDR1-NETWORK1-ODU2e-service1'}, res['odu-connection'][0]['destination']) self.assertDictEqual({u'src-if': u'XPDR1-CLIENT1-ODU2e-service1'},