Upgrade openroadm service models to 10.1

[transportpce.git] / tests / transportpce_tests / 2.2.1 / test10_tapi.py

diff --git a/tests/transportpce_tests/2.2.1/test10_tapi.py b/tests/transportpce_tests/2.2.1/test10_tapi.py
index 25a09c2b29b2176749a0595a1ea3439b1cbf5f04..f3960b1abc3e518a1b3ab610184a44441678ce35 100644 (file)
--- a/tests/transportpce_tests/2.2.1/test10_tapi.py
+++ b/tests/transportpce_tests/2.2.1/test10_tapi.py
@@ -15,12 +15,15 @@
 # pylint: disable=unsupported-assignment-operation
 
 import os
+# pylint: disable=wrong-import-order
 import sys
 import time
 import unittest
 import requests
 sys.path.append('transportpce_tests/common/')
-import test_utils
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
+import test_utils  # nopep8
 
 
 CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
@@ -46,13 +49,7 @@ class TransportTapitesting(unittest.TestCase):
             "service-format": "OTU",
             "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
             "clli": "NodeSA",
-            "subrate-eth-sla": {
-                    "subrate-eth-sla": {
-                        "committed-info-rate": "100000",
-                        "committed-burst-size": "64"
-                    }
-            },
-            "tx-direction": {
+            "tx-direction": [{
                 "port": {
                     "port-device-name": "SPDR-SA1-XPDR1",
                     "port-type": "fixed",
@@ -65,9 +62,10 @@ class TransportTapitesting(unittest.TestCase):
                     "lgx-port-name": "Some lgx-port-name",
                     "lgx-port-rack": "000000.00",
                     "lgx-port-shelf": "00"
-                }
-            },
-            "rx-direction": {
+                },
+                "index": 0
+            }],
+            "rx-direction": [{
                 "port": {
                     "port-device-name": "SPDR-SA1-XPDR1",
                     "port-type": "fixed",
@@ -80,8 +78,9 @@ class TransportTapitesting(unittest.TestCase):
                     "lgx-port-name": "Some lgx-port-name",
                     "lgx-port-rack": "000000.00",
                     "lgx-port-shelf": "00"
-                }
-            },
+                },
+                "index": 0
+            }],
             "optic-type": "gray"
         },
         "service-z-end": {
@@ -90,13 +89,7 @@ class TransportTapitesting(unittest.TestCase):
             "service-format": "OTU",
             "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
             "clli": "NodeSC",
-            "subrate-eth-sla": {
-                    "subrate-eth-sla": {
-                        "committed-info-rate": "100000",
-                        "committed-burst-size": "64"
-                    }
-            },
-            "tx-direction": {
+            "tx-direction": [{
                 "port": {
                     "port-device-name": "SPDR-SC1-XPDR1",
                     "port-type": "fixed",
@@ -109,9 +102,10 @@ class TransportTapitesting(unittest.TestCase):
                     "lgx-port-name": "Some lgx-port-name",
                     "lgx-port-rack": "000000.00",
                     "lgx-port-shelf": "00"
-                }
-            },
-            "rx-direction": {
+                },
+                "index": 0
+            }],
+            "rx-direction": [{
                 "port": {
                     "port-device-name": "SPDR-SC1-XPDR1",
                     "port-type": "fixed",
@@ -124,8 +118,9 @@ class TransportTapitesting(unittest.TestCase):
                     "lgx-port-name": "Some lgx-port-name",
                     "lgx-port-rack": "000000.00",
                     "lgx-port-shelf": "00"
-                }
-            },
+                },
+                "index": 0
+            }],
             "optic-type": "gray"
         },
         "due-date": "2018-06-15T00:00:01Z",
@@ -173,6 +168,7 @@ class TransportTapitesting(unittest.TestCase):
     def setUp(self):  # instruction executed before each test method
         if self.init_failed:
             self.fail('Feature installation failed')
+        # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
 
     def test_01_get_tapi_topology_T100G(self):
@@ -236,7 +232,7 @@ class TransportTapitesting(unittest.TestCase):
         response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
         res = response.json()
-        self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
+        self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'Topology should contain 1 node')
         self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 
     def test_05_disconnect_roadmb(self):
@@ -250,7 +246,17 @@ class TransportTapitesting(unittest.TestCase):
         time.sleep(10)
 
     def test_07_check_tapi_topos(self):
-        self.test_04_check_tapi_topos()
+        url = "{}/operations/tapi-topology:get-topology-details"
+        data = {
+            "tapi-topology:input": {
+                "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
+            }
+        }
+        response = test_utils.post_request(url, data)
+        self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
+        res = response.json()
+        self.assertNotIn("node", res["output"]["topology"], 'Topology should contain no node')
+        self.assertNotIn("link", res["output"]["topology"], 'Topology should contain no link')
 
     def test_08_connect_rdma(self):
         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
@@ -313,9 +319,9 @@ class TransportTapitesting(unittest.TestCase):
         res = response.json()
         self.assertEqual(1, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
                          'Node should contain 1 owned-node-edge-points')
-        self.assertEqual("XPDR1-CLIENT1",
+        self.assertEqual("XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1",
                          res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
-                         'name of owned-node-edge-points should be XPDR1-CLIENT1')
+                         'name of owned-node-edge-points should be XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1')
 
     def test_14_check_tapi_topology_T0(self):
         url = "{}/operations/tapi-topology:get-topology-details"
@@ -375,12 +381,12 @@ class TransportTapitesting(unittest.TestCase):
         res = response.json()
         self.assertEqual(2, len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]),
                          'Node should contain 2 owned-node-edge-points')
-        self.assertEqual("XPDR1-CLIENT1",
+        self.assertEqual("XPDR-C1-XPDR1+DSR+XPDR1-CLIENT1",
                          res["output"]["topology"]["node"][0]["owned-node-edge-point"][0]["name"][0]["value"],
-                         'name of owned-node-edge-points should be XPDR1-CLIENT1')
-        self.assertEqual("XPDR1-CLIENT1",
+                         'name of owned-node-edge-points should be XPDR-C1-XPDR1+DSR+XPDR1-CLIENT1')
+        self.assertEqual("XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1",
                          res["output"]["topology"]["node"][0]["owned-node-edge-point"][1]["name"][0]["value"],
-                         'name of owned-node-edge-points should be XPDR1-CLIENT1')
+                         'name of owned-node-edge-points should be XPDR-A1-XPDR1+DSR+XPDR1-CLIENT1')
 
     def test_19_check_tapi_topology_T0(self):
         url = "{}/operations/tapi-topology:get-topology-details"
@@ -681,8 +687,7 @@ class TransportTapitesting(unittest.TestCase):
         res = response.json()
         links = res['network'][0]['ietf-network-topology:link']
         for link in links:
-            if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
-                    link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
+            if link["org-openroadm-common-network:link-type"] in ('XPONDER-OUTPUT', 'XPONDER-INPUT'):
                 link_name = link["link-id"]
                 response = test_utils.delete_request(url+link_name)
                 self.assertEqual(response.status_code, requests.codes.ok)
@@ -761,8 +766,8 @@ def find_object_with_key(list_dicts, key, value):
 
 def count_object_with_double_key(list_dicts, key1, key2, value):
     nb = 0
-    for dict in list_dicts:
-        if dict[key1][0][key2] == value:
+    for dictio in list_dicts:
+        if dictio[key1][0][key2] == value:
             nb += 1
     return nb