Update portmapping funct tests for otn-mpdr conf

- add new tests in portmapping functional tests for device 2.2.1 to
  check the OTN device
- add specific tests in portmapping func tests and test_utils to check
  configuration of the odu-switching-pool part
- speed up test execution

JIRA: TRNSPRTPCE-612
Signed-off-by: Gilles Thouenon <gilles.thouenon@orange.com>
Change-Id: Ie171beee181de052884fe553b7b0739a215cd04d

diff --git a/tests/transportpce_tests/2.2.1/test01_portmapping.py b/tests/transportpce_tests/2.2.1/test01_portmapping.py
index 8c01d71ac6617e5167827d16a7670fdb243f2433..a96956a42a44f77c686337829d1bc29c943f2834 100644 (file)
--- a/tests/transportpce_tests/2.2.1/test01_portmapping.py
+++ b/tests/transportpce_tests/2.2.1/test01_portmapping.py
@@ -31,7 +31,8 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.processes = test_utils_rfc8040.start_tpce()
-        cls.processes = test_utils_rfc8040.start_sims([('xpdra', cls.NODE_VERSION), ('roadma', cls.NODE_VERSION)])
+        cls.processes = test_utils_rfc8040.start_sims([('xpdra', cls.NODE_VERSION), ('roadma', cls.NODE_VERSION),
+                                                       ('spdra', cls.NODE_VERSION)])
 
     @classmethod
     def tearDownClass(cls):
@@ -43,7 +44,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
     def setUp(self):
         # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
-        time.sleep(10)
+        time.sleep(1)
 
     def test_01_rdm_device_connection(self):
         response = test_utils_rfc8040.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
@@ -53,7 +54,6 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
         response = test_utils_rfc8040.check_device_connection("ROADM-A1")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertEqual(response['connection-status'], 'connected')
-        time.sleep(10)
 
     def test_03_rdm_portmapping_info(self):
         response = test_utils_rfc8040.get_portmapping_node_info("ROADM-A1")
@@ -66,7 +66,6 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
              'node-vendor': 'vendorA',
              'node-model': 'model2'},
             response['node-info'])
-        time.sleep(3)
 
     def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
         response = test_utils_rfc8040.portmapping_request("ROADM-A1", "DEG1-TTP-TXRX")
@@ -114,7 +113,6 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
         response = test_utils_rfc8040.check_device_connection("XPDR-A1")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertEqual(response['connection-status'], 'connected')
-        time.sleep(10)
 
     def test_10_xpdr_portmapping_info(self):
         response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A1")
@@ -127,7 +125,6 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
              'node-vendor': 'vendorA',
              'node-model': 'model2'},
             response['node-info'])
-        time.sleep(3)
 
     def test_11_xpdr_portmapping_NETWORK1(self):
         response = test_utils_rfc8040.portmapping_request("XPDR-A1", "XPDR1-NETWORK1")
@@ -179,11 +176,191 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
             response['mapping'])
 
-    def test_15_xpdr_device_disconnection(self):
+    def test_15_spdr_device_connection(self):
+        response = test_utils_rfc8040.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
+        self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201)
+
+    def test_16_spdr_device_connected(self):
+        response = test_utils_rfc8040.check_device_connection("SPDR-SA1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual(response['connection-status'], 'connected')
+
+    def test_17_spdr_portmapping_info(self):
+        response = test_utils_rfc8040.get_portmapping_node_info("SPDR-SA1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual(
+            {'node-type': 'xpdr',
+             'node-ip-address': '1.2.3.4',
+             'node-clli': 'NodeSA',
+             'openroadm-version': '2.2.1',
+             'node-vendor': 'vendorA',
+             'node-model': 'universal-switchponder'},
+            response['node-info'])
+
+    def test_18_spdr_switching_pool_1(self):
+        response = test_utils_rfc8040.portmapping_switching_pool_request("SPDR-SA1", "1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual("blocking",
+                         response['switching_pool'][0]['switching-pool-type'])
+        self.assertEqual(4,
+                         len(response['switching_pool'][0]['non-blocking-list']))
+        self.assertIn(
+            {'nbl-number': 11,
+             'interconnect-bandwidth': 0,
+             'interconnect-bandwidth-unit': 1000000000,
+             'lcp-list': ['XPDR1-NETWORK1', 'XPDR1-CLIENT1']},
+            response['switching_pool'][0]['non-blocking-list'])
+
+    def test_19_spdr_switching_pool_2(self):
+        response = test_utils_rfc8040.portmapping_switching_pool_request("SPDR-SA1", "2")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual("non-blocking",
+                         response['switching_pool'][0]['switching-pool-type'])
+        self.assertEqual(1,
+                         len(response['switching_pool'][0]['non-blocking-list']))
+        self.assertIn(
+            {'nbl-number': 2,
+             'interconnect-bandwidth': 0,
+             'interconnect-bandwidth-unit': 1000000000,
+             'lcp-list': ["XPDR2-CLIENT3", "XPDR2-CLIENT4", "XPDR2-NETWORK4",
+                          "XPDR2-NETWORK1", "XPDR2-NETWORK3", "XPDR2-NETWORK2",
+                          "XPDR2-CLIENT1", "XPDR2-CLIENT2"]},
+            response['switching_pool'][0]['non-blocking-list'])
+
+    def test_20_spdr_switching_pool_3(self):
+        response = test_utils_rfc8040.portmapping_switching_pool_request("SPDR-SA1", "3")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual("blocking",
+                         response['switching_pool'][0]['switching-pool-type'])
+        self.assertEqual(4,
+                         len(response['switching_pool'][0]['non-blocking-list']))
+        self.assertIn(
+            {'nbl-number': 83,
+             'interconnect-bandwidth': 0,
+             'interconnect-bandwidth-unit': 1000000000,
+             'lcp-list': ['XPDR3-NETWORK1', 'XPDR3-CLIENT3']},
+            response['switching_pool'][0]['non-blocking-list'])
+
+    def test_21_spdr_portmapping_mappings(self):
+        response = test_utils_rfc8040.get_portmapping("SPDR-SA1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual(18, len(response['nodes'][0]['mapping']))
+
+    def test_22_spdr_portmapping_XPDR1_CLIENT1(self):
+        response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR1-CLIENT1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertIn(
+            {"logical-connection-point": "XPDR1-CLIENT1",
+             "supporting-circuit-pack-name": "CP1-SFP1",
+             "supported-interface-capability": ["org-openroadm-port-types:if-10GE-ODU2e",
+                                                "org-openroadm-port-types:if-10GE-ODU2"],
+             "supporting-port": "CP1-SFP1-P1",
+             "lcp-hash-val": "FqlcrxV7p30=",
+             "port-direction": "bidirectional",
+             "port-admin-state": "InService",
+             "xponder-type": "mpdr",
+             "port-qual": "xpdr-client",
+             "port-oper-state": "InService"},
+            response['mapping'])
+
+    def test_23_spdr_portmapping_XPDR1_NETWORK1(self):
+        response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR1-NETWORK1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertIn(
+            {"logical-connection-point": "XPDR1-NETWORK1",
+             "supporting-circuit-pack-name": "CP1-CFP0",
+             "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
+             "supporting-port": "CP1-CFP0-P1",
+             "lcp-hash-val": "Swfw02qXGyI=",
+             "port-direction": "bidirectional",
+             "port-admin-state": "InService",
+             "xponder-type": "mpdr",
+             "port-qual": "xpdr-network",
+             "port-oper-state": "InService"},
+            response['mapping'])
+
+    def test_24_spdr_portmapping_XPDR2_CLIENT2(self):
+        response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR2-CLIENT2")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertIn(
+            {"logical-connection-point": "XPDR2-CLIENT2",
+             "supporting-circuit-pack-name": "CP2-QSFP2",
+             "supported-interface-capability": ["org-openroadm-port-types:if-100GE-ODU4",
+                                                "org-openroadm-port-types:if-100GE"],
+             "supporting-port": "CP2-QSFP2-P1",
+             "lcp-hash-val": "AN/WSSRXne3t",
+             "port-direction": "bidirectional",
+             "port-admin-state": "InService",
+             "xponder-type": "switch",
+             "port-qual": "switch-client",
+             "port-oper-state": "InService"},
+            response['mapping'])
+
+    def test_25_spdr_portmapping_XPDR2_NETWORK2(self):
+        response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR2-NETWORK2")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertIn(
+            {"logical-connection-point": "XPDR2-NETWORK2",
+             "supporting-circuit-pack-name": "CP6-CFP",
+             "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
+             "supporting-port": "CP6-CFP-P1",
+             "lcp-hash-val": "exT821pFtOQ=",
+             "port-direction": "bidirectional",
+             "port-admin-state": "InService",
+             "xponder-type": "switch",
+             "port-qual": "switch-network",
+             "port-oper-state": "InService"},
+            response['mapping'])
+
+    def test_26_spdr_portmapping_XPDR3_CLIENT3(self):
+        response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR3-CLIENT3")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertIn(
+            {"logical-connection-point": "XPDR3-CLIENT3",
+             "supporting-circuit-pack-name": "CP3-SFP3",
+             "supported-interface-capability": ["org-openroadm-port-types:if-1GE-ODU0"],
+             "supporting-port": "CP3-SFP3-P1",
+             "lcp-hash-val": "AKsQ/HRQdtdN",
+             "port-direction": "bidirectional",
+             "port-admin-state": "InService",
+             "xponder-type": "mpdr",
+             "port-qual": "xpdr-client",
+             "port-oper-state": "InService"},
+            response['mapping'])
+
+    def test_27_spdr_portmapping_XPDR3_NETWORK1(self):
+        response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR3-NETWORK1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertIn(
+            {"logical-connection-point": "XPDR3-NETWORK1",
+             "supporting-circuit-pack-name": "CP3-CFP0",
+             "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
+             "supporting-port": "CP3-CFP0-P1",
+             "lcp-hash-val": "ANnxoi7K8q30",
+             "port-direction": "bidirectional",
+             "port-admin-state": "InService",
+             "xponder-type": "mpdr",
+             "port-qual": "xpdr-network",
+             "port-oper-state": "InService"},
+            response['mapping'])
+
+    def test_28_spdr_device_disconnection(self):
+        response = test_utils_rfc8040.unmount_device("SPDR-SA1")
+        self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
+
+    def test_29_xpdr_device_disconnected(self):
+        response = test_utils_rfc8040.check_device_connection("SPDR-SA1")
+        self.assertEqual(response['status_code'], requests.codes.conflict)
+        self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
+        self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
+        self.assertEqual(response['connection-status']['error-message'],
+                         'Request could not be completed because the relevant data model content does not exist')
+
+    def test_30_xpdr_device_disconnection(self):
         response = test_utils_rfc8040.unmount_device("XPDR-A1")
         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 
-    def test_16_xpdr_device_disconnected(self):
+    def test_31_xpdr_device_disconnected(self):
         response = test_utils_rfc8040.check_device_connection("XPDR-A1")
         self.assertEqual(response['status_code'], requests.codes.conflict)
         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
@@ -191,7 +368,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
         self.assertEqual(response['connection-status']['error-message'],
                          'Request could not be completed because the relevant data model content does not exist')
 
-    def test_17_xpdr_device_not_connected(self):
+    def test_32_xpdr_device_not_connected(self):
         response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A1")
         self.assertEqual(response['status_code'], requests.codes.conflict)
         self.assertIn(response['node-info']['error-type'], ('protocol', 'application'))
@@ -199,11 +376,11 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
         self.assertEqual(response['node-info']['error-message'],
                          'Request could not be completed because the relevant data model content does not exist')
 
-    def test_18_rdm_device_disconnection(self):
+    def test_33_rdm_device_disconnection(self):
         response = test_utils_rfc8040.unmount_device("ROADM-A1")
         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 
-    def test_19_rdm_device_disconnected(self):
+    def test_34_rdm_device_disconnected(self):
         response = test_utils_rfc8040.check_device_connection("ROADM-A1")
         self.assertEqual(response['status_code'], requests.codes.conflict)
         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
@@ -211,7 +388,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
         self.assertEqual(response['connection-status']['error-message'],
                          'Request could not be completed because the relevant data model content does not exist')
 
-    def test_20_rdm_device_not_connected(self):
+    def test_35_rdm_device_not_connected(self):
         response = test_utils_rfc8040.get_portmapping_node_info("ROADM-A1")
         self.assertEqual(response['status_code'], requests.codes.conflict)
         self.assertIn(response['node-info']['error-type'], ('protocol', 'application'))

diff --git a/tests/transportpce_tests/7.1/test01_portmapping.py b/tests/transportpce_tests/7.1/test01_portmapping.py
index 6aed5f30971968dc6c4f8eeed699ad397b9211cf..a7bf2b9490e898c02227e17223366f6123c4bf49 100644 (file)
--- a/tests/transportpce_tests/7.1/test01_portmapping.py
+++ b/tests/transportpce_tests/7.1/test01_portmapping.py
@@ -42,7 +42,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
     def setUp(self):
         # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
-        time.sleep(10)
+        time.sleep(1)
 
     def test_01_xpdr_device_connection(self):
         response = test_utils_rfc8040.mount_device("XPDR-A2",
@@ -55,7 +55,6 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
         response = test_utils_rfc8040.check_device_connection("XPDR-A2")
         self.assertEqual(response['status_code'], requests.codes.ok)
         self.assertEqual(response['connection-status'], 'connected')
-        time.sleep(10)
 
     # Check node info in the port-mappings
     def test_03_xpdr_portmapping_info(self):
@@ -69,7 +68,6 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
              'node-vendor': 'vendorA',
              'node-model': 'model'},
             response['node-info'])
-        time.sleep(3)
 
     # Check the if-capabilities and the other details for network
     def test_04_tpdr_portmapping_NETWORK1(self):
@@ -158,11 +156,24 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
         self.assertEqual(str(res['mc-capabilities'][0]['center-freq-granularity']), '3.125')
         self.assertEqual(str(res['mc-capabilities'][0]['slot-width-granularity']), '6.25')
 
-    def test_09_xpdr_device_disconnection(self):
+    def test_09_mpdr_switching_pool(self):
+        response = test_utils_rfc8040.portmapping_switching_pool_request("XPDR-A2", "1")
+        self.assertEqual(response['status_code'], requests.codes.ok)
+        self.assertEqual("blocking",
+                         response['switching_pool'][0]['switching-pool-type'])
+        self.assertEqual(2,
+                         len(response['switching_pool'][0]['non-blocking-list']))
+        self.assertIn(
+            {'nbl-number': 2,
+             'interconnect-bandwidth': 0,
+             'lcp-list': ['XPDR2-NETWORK1', 'XPDR2-CLIENT2']},
+            response['switching_pool'][0]['non-blocking-list'])
+
+    def test_10_xpdr_device_disconnection(self):
         response = test_utils_rfc8040.unmount_device("XPDR-A2")
         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 
-    def test_10_xpdr_device_disconnected(self):
+    def test_11_xpdr_device_disconnected(self):
         response = test_utils_rfc8040.check_device_connection("XPDR-A2")
         self.assertEqual(response['status_code'], requests.codes.conflict)
         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
@@ -170,7 +181,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
         self.assertEqual(response['connection-status']['error-message'],
                          'Request could not be completed because the relevant data model content does not exist')
 
-    def test_11_xpdr_device_not_connected(self):
+    def test_12_xpdr_device_not_connected(self):
         response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A2")
         self.assertEqual(response['status_code'], requests.codes.conflict)
         self.assertIn(response['node-info']['error-type'], ('protocol', 'application'))

diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py
index a5c901fb9de9823f49aa2cb2ff9c8b97293d7d09..263d3b6f4a761a780060d0b4f618fb3b5af98728 100644 (file)
--- a/tests/transportpce_tests/common/test_utils_rfc8040.py
+++ b/tests/transportpce_tests/common/test_utils_rfc8040.py
@@ -385,6 +385,18 @@ def portmapping_request(node: str, mapping: str):
             'mapping': mapping}
 
 
+def portmapping_switching_pool_request(node: str, switching_pool: str):
+    url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}/switching-pool-lcp={}',
+           'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}/switching-pool-lcp/{}'}
+    response = get_request(url[RESTCONF_VERSION].format('{}', node, switching_pool))
+    res = response.json()
+    return_key = {'rfc8040': 'transportpce-portmapping:switching-pool-lcp',
+                  'draft-bierman02': 'switching-pool-lcp'}
+    switching_pool = res[return_key[RESTCONF_VERSION]]
+    return {'status_code': response.status_code,
+            'switching_pool': switching_pool}
+
+
 def portmapping_mc_capa_request(node: str, mc_capa: str):
     url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}/mc-capabilities={}',
            'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}/mc-capabilities/{}'}