Merge "Drop the service name from ODU connection name"

[transportpce.git] / tests / transportpce_tests / 7.1 / test02_otn_renderer.py

diff --git a/tests/transportpce_tests/7.1/test02_otn_renderer.py b/tests/transportpce_tests/7.1/test02_otn_renderer.py
index 7d77949ce799b4cf2360774a5dd1d6cfb771b984..b7510bc42e4e8d89879dae409540568b808a2cd9 100644 (file)
--- a/tests/transportpce_tests/7.1/test02_otn_renderer.py
+++ b/tests/transportpce_tests/7.1/test02_otn_renderer.py
@@ -66,8 +66,9 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
     # this test has been removed, since it already exists in port-mapping
     # 1a) create a OTUC2 device renderer
     def test_02_service_path_create_otuc2(self):
-        response = test_utils_rfc8040.device_renderer_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'service-path',
+            {
                 'service-name': 'service_OTUC2',
                 'wave-number': '0',
                 'modulation-format': 'dp-qpsk',
@@ -79,13 +80,15 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
                 'max-freq': 196.125,
                 'lower-spectral-slot-number': 755,
                 'higher-spectral-slot-number': 768
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Interfaces created successfully for nodes: ', response['output']['result'])
         self.assertIn(
             {'node-id': 'XPDR-A2',
              'otu-interface-id': ['XPDR2-NETWORK1-OTUC2'],
-             'och-interface-id': ['XPDR2-NETWORK1-OTSIGROUP-200G']}, response['output']['node-interface'])
+             'och-interface-id': ['XPDR2-NETWORK1-OTSIGROUP-200G',
+                                  'XPDR2-NETWORK1-755:768']},
+            response['output']['node-interface'])
 
     def test_03_get_portmapping_network1(self):
         response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
@@ -163,14 +166,15 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
 
     # 1b) create a ODUC2 device renderer
     def test_07_otn_service_path_create_oduc2(self):
-        response = test_utils_rfc8040.device_renderer_otn_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'otn-service-path',
+            {
                 'service-name': 'service_ODUC2',
                 'operation': 'create',
                 'service-rate': '200',
                 'service-format': 'ODU',
                 'nodes': [{'node-id': 'XPDR-A2', 'network-tp': 'XPDR2-NETWORK1'}]
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Otn Service path was set up successfully for node :XPDR-A2', response['output']['result'])
         self.assertIn(
@@ -217,8 +221,9 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
 
     # 1c) create Ethernet device renderer
     def test_10_otn_service_path_create_100ge(self):
-        response = test_utils_rfc8040.device_renderer_otn_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'otn-service-path',
+            {
                 'service-name': 'service_Ethernet',
                 'operation': 'create',
                 'service-rate': '100',
@@ -226,16 +231,16 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
                 'nodes': [{'node-id': 'XPDR-A2', 'client-tp': 'XPDR2-CLIENT1', 'network-tp': 'XPDR2-NETWORK1'}],
                 'ethernet-encoding': 'eth encode',
                 'opucn-trib-slots': ['1.1', '1.20']
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Otn Service path was set up successfully for node :XPDR-A2', response['output']['result'])
         self.assertEqual('XPDR-A2', response['output']['node-interface'][0]['node-id'])
-        self.assertIn('XPDR2-CLIENT1-ODU4-service_Ethernet-x-XPDR2-NETWORK1-ODU4-service_Ethernet',
+        self.assertIn('XPDR2-CLIENT1-ODU4-x-XPDR2-NETWORK1-ODU4',
                       response['output']['node-interface'][0]['connection-id'])
         self.assertIn('XPDR2-CLIENT1-ETHERNET-100G', response['output']['node-interface'][0]['eth-interface-id'])
-        self.assertIn('XPDR2-NETWORK1-ODU4-service_Ethernet',
+        self.assertIn('XPDR2-NETWORK1-ODU4',
                       response['output']['node-interface'][0]['odu-interface-id'])
-        self.assertIn('XPDR2-CLIENT1-ODU4-service_Ethernet',
+        self.assertIn('XPDR2-CLIENT1-ODU4',
                       response['output']['node-interface'][0]['odu-interface-id'])
 
     def test_11_check_interface_100ge_client(self):
@@ -257,9 +262,9 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
 
     def test_12_check_interface_odu4_client(self):
         response = test_utils_rfc8040.check_node_attribute_request(
-            "XPDR-A2", "interface", "XPDR2-CLIENT1-ODU4-service_Ethernet")
+            "XPDR-A2", "interface", "XPDR2-CLIENT1-ODU4")
         self.assertEqual(response['status_code'], requests.codes.ok)
-        input_dict_1 = {'name': 'XPDR2-CLIENT1-ODU4-service_Ethernet',
+        input_dict_1 = {'name': 'XPDR2-CLIENT1-ODU4',
                         'administrative-state': 'inService',
                         'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT',
                         'supporting-interface-list': 'XPDR2-CLIENT1-ETHERNET-100G',
@@ -281,9 +286,9 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
 
     def test_13_check_interface_odu4_network(self):
         response = test_utils_rfc8040.check_node_attribute_request(
-            "XPDR-A2", "interface", "XPDR2-NETWORK1-ODU4-service_Ethernet")
+            "XPDR-A2", "interface", "XPDR2-NETWORK1-ODU4")
         self.assertEqual(response['status_code'], requests.codes.ok)
-        input_dict_1 = {'name': 'XPDR2-NETWORK1-ODU4-service_Ethernet',
+        input_dict_1 = {'name': 'XPDR2-NETWORK1-ODU4',
                         'administrative-state': 'inService',
                         'supporting-circuit-pack-name': '1/2/2-PLUG-NET',
                         'supporting-interface-list': 'XPDR2-NETWORK1-ODUC2',
@@ -312,25 +317,26 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
     def test_14_check_odu_connection_xpdra2(self):
         response = test_utils_rfc8040.check_node_attribute_request(
             "XPDR-A2",
-            "odu-connection", "XPDR2-CLIENT1-ODU4-service_Ethernet-x-XPDR2-NETWORK1-ODU4-service_Ethernet")
+            "odu-connection", "XPDR2-CLIENT1-ODU4-x-XPDR2-NETWORK1-ODU4")
         self.assertEqual(response['status_code'], requests.codes.ok)
         input_dict_1 = {
             'connection-name':
-            'XPDR2-CLIENT1-ODU4-service_Ethernet-x-XPDR2-NETWORK1-ODU4-service_Ethernet',
+            'XPDR2-CLIENT1-ODU4-x-XPDR2-NETWORK1-ODU4',
             'direction': 'bidirectional'
         }
 
         self.assertDictEqual(dict(input_dict_1, **response['odu-connection'][0]),
                              response['odu-connection'][0])
-        self.assertDictEqual({'dst-if': 'XPDR2-NETWORK1-ODU4-service_Ethernet'},
+        self.assertDictEqual({'dst-if': 'XPDR2-NETWORK1-ODU4'},
                              response['odu-connection'][0]['destination'])
-        self.assertDictEqual({'src-if': 'XPDR2-CLIENT1-ODU4-service_Ethernet'},
+        self.assertDictEqual({'src-if': 'XPDR2-CLIENT1-ODU4'},
                              response['odu-connection'][0]['source'])
 
     # 1d) Delete Ethernet device interfaces
     def test_15_otn_service_path_delete_100ge(self):
-        response = test_utils_rfc8040.device_renderer_otn_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'otn-service-path',
+            {
                 'service-name': 'service_Ethernet',
                 'operation': 'delete',
                 'service-rate': '100',
@@ -339,24 +345,24 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
                 'ethernet-encoding': 'eth encode',
                 'trib-slot': ['1'],
                 'trib-port-number': '1'
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Request processed', response['output']['result'])
 
     def test_16_check_no_odu_connection(self):
         response = test_utils_rfc8040.check_node_attribute_request(
             "XPDR-A2",
-            "odu-connection", "XPDR2-CLIENT1-ODU4-service_Ethernet-x-XPDR2-NETWORK1-ODU4-service_Ethernet")
+            "odu-connection", "XPDR2-CLIENT1-ODU4-x-XPDR2-NETWORK1-ODU4")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_17_check_no_interface_odu_network(self):
         response = test_utils_rfc8040.check_node_attribute_request(
-            "XPDR-A2", "interface", "XPDR2-NETWORK1-ODU4-service_Ethernet")
+            "XPDR-A2", "interface", "XPDR2-NETWORK1-ODU4")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_18_check_no_interface_odu_client(self):
         response = test_utils_rfc8040.check_node_attribute_request(
-            "XPDR-A2", "interface", "XPDR2-CLIENT1-ODU4-service_Ethernet")
+            "XPDR-A2", "interface", "XPDR2-CLIENT1-ODU4")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
     def test_19_check_no_interface_100ge_client(self):
@@ -366,25 +372,35 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
 
     # 1e) Delete ODUC2 device interfaces
     def test_20_otn_service_path_delete_oduc2(self):
-        response = test_utils_rfc8040.device_renderer_otn_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'otn-service-path',
+            {
                 'service-name': 'service_ODUC2',
                 'operation': 'delete',
                 'service-rate': '200',
                 'service-format': 'ODU',
                 'nodes': [{'node-id': 'XPDR-A2', 'network-tp': 'XPDR2-NETWORK1'}]
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Request processed', response['output']['result'])
+        # Here you have remove the added oducn supporting port interface
+        del self.NETWORK2_CHECK_DICT["supporting-oducn"]
 
     def test_21_check_no_interface_oduc2(self):
         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A2", "interface", "XPDR2-NETWORK1-ODUC2")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
+    # Check if port-mapping data is updated, where the supporting-oducn is deleted
+    def test_21a_check_no_oduc2(self):
+        response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
+        self.assertRaises(KeyError, lambda: response["supporting-oducn"])
+
     # 1f) Delete OTUC2 device interfaces
+
     def test_22_service_path_delete_otuc2(self):
-        response = test_utils_rfc8040.device_renderer_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'service-path',
+            {
                 'service-name': 'service_OTUC2',
                 'wave-number': '0',
                 'modulation-format': 'dp-qpsk',
@@ -396,9 +412,10 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
                 'max-freq': 196.125,
                 'lower-spectral-slot-number': 755,
                 'higher-spectral-slot-number': 768
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Request processed', response['output']['result'])
+        del self.NETWORK2_CHECK_DICT["supporting-otucn"]
 
     def test_23_check_no_interface_otuc2(self):
         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR2-NETWORK1-OTUC2")
@@ -413,10 +430,15 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR2-NETWORK1-755:768")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
+    def test_25a_check_no_otuc2(self):
+        response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
+        self.assertRaises(KeyError, lambda: response["supporting-otucn"])
+
     # 2a) create a OTUC3 device renderer
     def test_26_service_path_create_otuc3(self):
-        response = test_utils_rfc8040.device_renderer_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'service-path',
+            {
                 'service-name': 'service_OTUC3',
                 'wave-number': '0',
                 'modulation-format': 'dp-qam8',
@@ -428,13 +450,15 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
                 'max-freq': 196.125,
                 'lower-spectral-slot-number': 755,
                 'higher-spectral-slot-number': 768
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Interfaces created successfully for nodes: ', response['output']['result'])
         self.assertIn(
             {'node-id': 'XPDR-A2',
              'otu-interface-id': ['XPDR2-NETWORK1-OTUC3'],
-             'och-interface-id': ['XPDR2-NETWORK1-OTSIGROUP-300G']}, response['output']['node-interface'])
+             'och-interface-id': ['XPDR2-NETWORK1-755:768',
+                                  'XPDR2-NETWORK1-OTSIGROUP-300G']},
+            response['output']['node-interface'])
 
     def test_27_get_portmapping_network1(self):
         response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
@@ -513,14 +537,15 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
 
     # 2b) create a ODUC3 device renderer
     def test_31_otn_service_path_create_oduc3(self):
-        response = test_utils_rfc8040.device_renderer_otn_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'otn-service-path',
+            {
                 'service-name': 'service_ODUC3',
                 'operation': 'create',
                 'service-rate': '300',
                 'service-format': 'ODU',
                 'nodes': [{'node-id': 'XPDR-A2', 'network-tp': 'XPDR2-NETWORK1'}]
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Otn Service path was set up successfully for node :XPDR-A2', response['output']['result'])
         self.assertIn(
@@ -572,25 +597,32 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
 
     # 2e) Delete ODUC3 device interfaces
     def test_34_otn_service_path_delete_oduc3(self):
-        response = test_utils_rfc8040.device_renderer_otn_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'otn-service-path',
+            {
                 'service-name': 'service_ODUC3',
                 'operation': 'delete',
                 'service-rate': '300',
                 'service-format': 'ODU',
                 'nodes': [{'node-id': 'XPDR-A2', 'network-tp': 'XPDR2-NETWORK1'}]
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Request processed', response['output']['result'])
+        del self.NETWORK2_CHECK_DICT["supporting-oducn"]
 
     def test_35_check_no_interface_oduc3(self):
         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A2", "interface", "XPDR2-NETWORK1-ODUC3")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
+    def test_35a_check_no_oduc3(self):
+        response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
+        self.assertRaises(KeyError, lambda: response["supporting-oducn"])
+
     # 2f) Delete OTUC3 device interfaces
     def test_36_service_path_delete_otuc3(self):
-        response = test_utils_rfc8040.device_renderer_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'service-path',
+            {
                 'service-name': 'service_OTUC3',
                 'wave-number': '0',
                 'modulation-format': 'dp-qam8',
@@ -602,9 +634,10 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
                 'max-freq': 196.125,
                 'lower-spectral-slot-number': 755,
                 'higher-spectral-slot-number': 768
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Request processed', response['output']['result'])
+        del self.NETWORK2_CHECK_DICT["supporting-otucn"]
 
     def test_37_check_no_interface_otuc3(self):
         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR2-NETWORK1-OTUC3")
@@ -619,10 +652,15 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR2-NETWORK1-755:768")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
+    def test_39a_check_no_otuc3(self):
+        response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
+        self.assertRaises(KeyError, lambda: response["supporting-otucn"])
+
     # 3a) create a OTUC4 device renderer
-    def test_40_service_path_create_otuc3(self):
-        response = test_utils_rfc8040.device_renderer_service_path_request(
-            {'input': {
+    def test_40_service_path_create_otuc4(self):
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'service-path',
+            {
                 'service-name': 'service_OTUC4',
                 'wave-number': '0',
                 'modulation-format': 'dp-qam16',
@@ -634,13 +672,15 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
                 'max-freq': 196.125,
                 'lower-spectral-slot-number': 755,
                 'higher-spectral-slot-number': 768
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Interfaces created successfully for nodes: ', response['output']['result'])
         self.assertIn(
             {'node-id': 'XPDR-A2',
              'otu-interface-id': ['XPDR2-NETWORK1-OTUC4'],
-             'och-interface-id': ['XPDR2-NETWORK1-OTSIGROUP-400G']}, response['output']['node-interface'])
+             'och-interface-id': ['XPDR2-NETWORK1-755:768',
+                                  'XPDR2-NETWORK1-OTSIGROUP-400G']},
+            response['output']['node-interface'])
 
     def test_41_get_portmapping_network1(self):
         response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
@@ -719,14 +759,15 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
 
     # 3b) create a ODUC4 device renderer
     def test_45_otn_service_path_create_oduc3(self):
-        response = test_utils_rfc8040.device_renderer_otn_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'otn-service-path',
+            {
                 'service-name': 'service_ODUC4',
                 'operation': 'create',
                 'service-rate': '400',
                 'service-format': 'ODU',
                 'nodes': [{'node-id': 'XPDR-A2', 'network-tp': 'XPDR2-NETWORK1'}]
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Otn Service path was set up successfully for node :XPDR-A2', response['output']['result'])
         self.assertIn(
@@ -778,25 +819,32 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
 
     # 3e) Delete ODUC4 device interfaces
     def test_48_otn_service_path_delete_oduc4(self):
-        response = test_utils_rfc8040.device_renderer_otn_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'otn-service-path',
+            {
                 'service-name': 'service_ODUC4',
                 'operation': 'delete',
                 'service-rate': '400',
                 'service-format': 'ODU',
                 'nodes': [{'node-id': 'XPDR-A2', 'network-tp': 'XPDR2-NETWORK1'}]
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Request processed', response['output']['result'])
+        del self.NETWORK2_CHECK_DICT["supporting-oducn"]
 
     def test_49_check_no_interface_oduc4(self):
         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A2", "interface", "XPDR2-NETWORK1-ODUC4")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
+    def test_49a_check_no_oduc4(self):
+        response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
+        self.assertRaises(KeyError, lambda: response["supporting-oducn"])
+
     # 3f) Delete OTUC4 device interfaces
     def test_50_service_path_delete_otuc4(self):
-        response = test_utils_rfc8040.device_renderer_service_path_request(
-            {'input': {
+        response = test_utils_rfc8040.transportpce_api_rpc_request(
+            'transportpce-device-renderer', 'service-path',
+            {
                 'service-name': 'service_OTUC4',
                 'wave-number': '0',
                 'modulation-format': 'dp-qam16',
@@ -808,9 +856,10 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
                 'max-freq': 196.125,
                 'lower-spectral-slot-number': 755,
                 'higher-spectral-slot-number': 768
-            }})
+            })
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertIn('Request processed', response['output']['result'])
+        del self.NETWORK2_CHECK_DICT["supporting-otucn"]
 
     def test_51_check_no_interface_otuc4(self):
         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR2-NETWORK1-OTUC4")
@@ -825,6 +874,10 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR2-NETWORK1-755:768")
         self.assertEqual(response['status_code'], requests.codes.conflict)
 
+    def test_53a_check_no_otuc4(self):
+        response = test_utils_rfc8040.portmapping_request("XPDR-A2", "XPDR2-NETWORK1")
+        self.assertRaises(KeyError, lambda: response["supporting-otucn"])
+
     # Disconnect the XPDR
     def test_54_xpdr_device_disconnection(self):
         response = test_utils_rfc8040.unmount_device("XPDR-A2")