Code Review / transportpce.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | review | tree
raw | inline | side by side
Update PCE for OTUC4, ODUC4 and 100GE MXPDR
[transportpce.git] / tests / transportpce_tests / 7.1 / test_pce_400G.py
diff --git a/tests/transportpce_tests/7.1/test_pce_400G.py b/tests/transportpce_tests/7.1/test_pce_400G.py
index 189e9245bfc6362af9ddefa5b9b26439e20552ac..0a8ed732056f88dff2aa63c083691c0f9f524488 100644 (file)
--- a/tests/transportpce_tests/7.1/test_pce_400G.py
+++ b/tests/transportpce_tests/7.1/test_pce_400G.py
@@ -33,7 +33,22 @@ class TransportPCE400Gtesting(unittest.TestCase):
             TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                             "..", "..", "sample_configs", "honeynode-topo400G.json")
             with open(TOPO_BI_DIR_FILE, 'r') as topo_bi_dir:
-                cls.simple_topo_bi_dir_data = topo_bi_dir.read()
+                cls.topo_bi_dir_data = topo_bi_dir.read()
+
+            OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
+                                            "..", "..", "sample_configs", "honeynode-otntopo400G.json")
+            with open(OTN_TOPO_BI_DIR_FILE, 'r') as otn_topo_bi_dir:
+                cls.otn_topo_bi_dir_data = otn_topo_bi_dir.read()
+
+            OTUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
+                                            "..", "..", "sample_configs", "honeynode-otntopo400GwithOTUC4.json")
+            with open(OTUC4_OTN_TOPO_BI_DIR_FILE, 'r') as otuc4_otn_topo_bi_dir:
+                cls.otuc4_otn_topo_bi_dir_data = otuc4_otn_topo_bi_dir.read()
+
+            ODUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
+                                            "..", "..", "sample_configs", "honeynode-otntopo400GwithODUC4.json")
+            with open(ODUC4_OTN_TOPO_BI_DIR_FILE, 'r') as oduc4_otn_topo_bi_dir:
+                cls.oduc4_otn_topo_bi_dir_data = oduc4_otn_topo_bi_dir.read()
 
             PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                                      "..", "..", "sample_configs", "pce_portmapping_71.json")
@@ -71,14 +86,14 @@ class TransportPCE400Gtesting(unittest.TestCase):
         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.created))
         time.sleep(2)
 
-    # Load simple bidirectional topology
-    def test_02_load_simple_topology_bi(self):
-        response = test_utils.put_jsonrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_bi_dir_data)
+    # Load openroadm topology
+    def test_02_load_openroadm_topology_bi(self):
+        response = test_utils.put_jsonrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.topo_bi_dir_data)
         self.assertEqual(response.status_code, requests.codes.ok)
         time.sleep(2)
 
     # Path Computation success
-    def test_03_path_computation_xpdr_bi(self):
+    def test_03_path_computation_400G_xpdr_bi(self):
         response = test_utils.path_computation_request("request-1", "service-1",
                                                        {"node-id": "XPDR-A2", "service-rate": "400",
                                                            "service-format": "Ethernet", "clli": "nodeA"},
@@ -112,18 +127,125 @@ class TransportPCE400Gtesting(unittest.TestCase):
                          ['zToA-direction']['modulation-format'])
         time.sleep(5)
 
-    # Test deleted complex topology
-    def test_04_test_topology_complex_deleted(self):
-        response = test_utils.get_ordm_topo_request("node/XPONDER-3-2")
-        self.assertEqual(response.status_code, requests.codes.conflict)
-        time.sleep(1)
+    # Load otn topology
+    def test_04_load_otn_topology_bi(self):
+        response = test_utils.put_jsonrequest(test_utils.URL_CONFIG_OTN_TOPO, self.otn_topo_bi_dir_data)
+        self.assertEqual(response.status_code, requests.codes.ok)
+        time.sleep(2)
+
+    # Path Computation success
+    def test_05_path_computation_OTUC4_xpdr_bi(self):
+        response = test_utils.path_computation_request("request-1", "service-OTUC4",
+                                                       {"service-rate": "400", "clli": "NodeA",
+                                                           "service-format": "OTU", "node-id": "XPDR-A2",
+                                                           "rx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2"}}
+                                                       },
+                                                       {"service-rate": "400", "clli": "NodeC",
+                                                           "service-format": "OTU", "node-id": "XPDR-C2",
+                                                           "rx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2"}}
+                                                       })
+        self.assertEqual(response.status_code, requests.codes.ok)
+        res = response.json()
+        self.assertIn('Path is calculated',
+                      res['output']['configuration-response-common']['response-message'])
+
+        self.assertEqual(1, res['output']['response-parameters']['path-description']
+                         ['aToZ-direction']['aToZ-wavelength-number'])
+        self.assertEqual(400, res['output']['response-parameters']['path-description']
+                         ['aToZ-direction']['rate'])
+        self.assertEqual(196.0375, res['output']['response-parameters']['path-description']
+                         ['aToZ-direction']['aToZ-min-frequency'])
+        self.assertEqual(196.12500, res['output']['response-parameters']['path-description']
+                         ['aToZ-direction']['aToZ-max-frequency'])
+        self.assertEqual('dp-qam16', res['output']['response-parameters']['path-description']
+                         ['aToZ-direction']['modulation-format'])
+
+        self.assertEqual(1, res['output']['response-parameters']['path-description']
+                         ['zToA-direction']['zToA-wavelength-number'])
+        self.assertEqual(400, res['output']['response-parameters']['path-description']
+                         ['zToA-direction']['rate'])
+        self.assertEqual(196.0375, res['output']['response-parameters']['path-description']
+                         ['zToA-direction']['zToA-min-frequency'])
+        self.assertEqual(196.12500, res['output']['response-parameters']['path-description']
+                         ['zToA-direction']['zToA-max-frequency'])
+        self.assertEqual('dp-qam16', res['output']['response-parameters']['path-description']
+                         ['zToA-direction']['modulation-format'])
+        time.sleep(5)
+
+    # Load otn topology with OTUC4 links
+    def test_06_load_otuc4_otn_topology_bi(self):
+        response = test_utils.put_jsonrequest(test_utils.URL_CONFIG_OTN_TOPO, self.otuc4_otn_topo_bi_dir_data)
+        self.assertEqual(response.status_code, requests.codes.ok)
+        time.sleep(2)
+
+    # Path Computation success
+    def test_07_path_computation_ODUC4_xpdr_bi(self):
+        response = test_utils.path_computation_request("request-1", "service-ODUC4",
+                                                       {"service-rate": "400", "clli": "NodeA", "service-format": "ODU",
+                                                           "node-id": "XPDR-A2-XPDR2",
+                                                           "tx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2"}}
+                                                       },
+                                                       {"service-rate": "400", "clli": "NodeC", "service-format": "ODU",
+                                                           "node-id": "XPDR-C2-XPDR2",
+                                                           "tx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2"}}
+                                                       })
+        self.assertEqual(response.status_code, requests.codes.ok)
+        res = response.json()
+        self.assertIn('Path is calculated',
+                      res['output']['configuration-response-common']['response-message'])
+
+        self.assertEqual(400, res['output']['response-parameters']['path-description']
+                         ['aToZ-direction']['rate'])
+        self.assertEqual('dp-qam16', res['output']['response-parameters']['path-description']
+                         ['aToZ-direction']['modulation-format'])
+
+        self.assertEqual(400, res['output']['response-parameters']['path-description']
+                         ['zToA-direction']['rate'])
+        self.assertEqual('dp-qam16', res['output']['response-parameters']['path-description']
+                         ['zToA-direction']['modulation-format'])
+        time.sleep(5)
 
-    # Delete portmapping
-    def test_05_delete_port_mapping(self):
-        response = test_utils.delete_request(test_utils.URL_FULL_PORTMAPPING)
+    # Load otn topology with OTUC4 links
+    def test_08_load_oduc4_otn_topology_bi(self):
+        response = test_utils.put_jsonrequest(test_utils.URL_CONFIG_OTN_TOPO, self.oduc4_otn_topo_bi_dir_data)
         self.assertEqual(response.status_code, requests.codes.ok)
         time.sleep(2)
 
+    # Path Computation success
+    def test_09_path_computation_100G_xpdr_bi(self):
+        response = test_utils.path_computation_request("request-1", "service-100GE",
+                                                       {"service-rate": "100", "clli": "NodeA", "service-format": "Ethernet",
+                                                        "node-id": "XPDR-A2",
+                                                        "tx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2",
+                                                           "port-name": "XPDR2-CLIENT1"}}},
+                                                       {"service-rate": "100", "clli": "NodeC", "service-format": "Ethernet",
+                                                        "node-id": "XPDR-C2",
+                                                        "tx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2",
+                                                           "port-name": "XPDR2-CLIENT1"}}})
+
+        self.assertEqual(response.status_code, requests.codes.ok)
+        res = response.json()
+        self.assertIn('Path is calculated',
+                      res['output']['configuration-response-common']['response-message'])
+
+        self.assertEqual(100, res['output']['response-parameters']['path-description']
+                         ['aToZ-direction']['rate'])
+        self.assertEqual(1, res['output']['response-parameters']['path-description']
+                         ['aToZ-direction']['trib-port-number'])
+        self.assertEqual(1, res['output']['response-parameters']['path-description']
+                         ['aToZ-direction']['trib-slot-number'])
+        self.assertEqual('dp-qpsk', res['output']['response-parameters']['path-description']
+                         ['aToZ-direction']['modulation-format'])
+
+        self.assertEqual(100, res['output']['response-parameters']['path-description']
+                         ['zToA-direction']['rate'])
+        self.assertEqual(1, res['output']['response-parameters']['path-description']
+                         ['zToA-direction']['trib-port-number'])
+        self.assertEqual(1, res['output']['response-parameters']['path-description']
+                         ['zToA-direction']['trib-slot-number'])
+        self.assertEqual('dp-qpsk', res['output']['response-parameters']['path-description']
+                         ['zToA-direction']['modulation-format'])
+        time.sleep(5)
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)
Repository for Transport PCE project.