Code Review / transportpce.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | review | tree
raw | inline | side by side
Merge "Drop the service name from ODU connection name"
[transportpce.git] / tests / transportpce_tests / hybrid / test01_device_change_notifications.py
diff --git a/tests/transportpce_tests/hybrid/test01_device_change_notifications.py b/tests/transportpce_tests/hybrid/test01_device_change_notifications.py
index c57316b20ecd35c60fa3c08c1c5c14a8f604b944..9f735f86dfe57df8868486ab0a0359ba230a4a64 100644 (file)
--- a/tests/transportpce_tests/hybrid/test01_device_change_notifications.py
+++ b/tests/transportpce_tests/hybrid/test01_device_change_notifications.py
@@ -14,9 +14,12 @@ import json
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
-import test_utils
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
+import test_utils  # nopep8
 
 
 class TransportPCEFulltesting(unittest.TestCase):
@@ -37,11 +40,9 @@ class TransportPCEFulltesting(unittest.TestCase):
             "node-id": "XPDRA01",
             "service-format": "Ethernet",
             "clli": "SNJSCAMCJP8",
-                    "tx-direction": {
+                    "tx-direction": [{
                         "port": {
-                            "port-device-name": "1/0/C1",
                             "port-type": "fixed",
-                            "port-name": "1",
                             "port-rack": "000000.00",
                             "port-shelf": "Chassis#1"
                         },
@@ -50,23 +51,23 @@ class TransportPCEFulltesting(unittest.TestCase):
                             "lgx-port-name": "Some lgx-port-name",
                             "lgx-port-rack": "000000.00",
                             "lgx-port-shelf": "00"
-                        }
-                    },
-            "rx-direction": {
-                        "port": {
-                            "port-device-name": "1/0/C1",
-                            "port-type": "fixed",
-                            "port-name": "1",
-                            "port-rack": "000000.00",
-                            "port-shelf": "Chassis#1"
                         },
-                        "lgx": {
-                            "lgx-device-name": "Some lgx-device-name",
-                            "lgx-port-name": "Some lgx-port-name",
-                            "lgx-port-rack": "000000.00",
-                            "lgx-port-shelf": "00"
-                        }
-                    },
+                        "index": 0
+                    }],
+            "rx-direction": [{
+                "port": {
+                    "port-type": "fixed",
+                    "port-rack": "000000.00",
+                    "port-shelf": "Chassis#1"
+                },
+                "lgx": {
+                    "lgx-device-name": "Some lgx-device-name",
+                    "lgx-port-name": "Some lgx-port-name",
+                    "lgx-port-rack": "000000.00",
+                    "lgx-port-shelf": "00"
+                },
+                "index": 0
+            }],
             "optic-type": "gray"
         },
         "service-z-end": {
@@ -74,11 +75,9 @@ class TransportPCEFulltesting(unittest.TestCase):
             "node-id": "XPDR-C1",
             "service-format": "Ethernet",
             "clli": "SNJSCAMCJT4",
-                    "tx-direction": {
+                    "tx-direction": [{
                         "port": {
-                            "port-device-name": "1/0/C1",
                             "port-type": "fixed",
-                            "port-name": "1",
                             "port-rack": "000000.00",
                             "port-shelf": "Chassis#1"
                         },
@@ -87,23 +86,23 @@ class TransportPCEFulltesting(unittest.TestCase):
                             "lgx-port-name": "Some lgx-port-name",
                             "lgx-port-rack": "000000.00",
                             "lgx-port-shelf": "00"
-                        }
-                    },
-            "rx-direction": {
-                        "port": {
-                            "port-device-name": "1/0/C1",
-                            "port-type": "fixed",
-                            "port-name": "1",
-                            "port-rack": "000000.00",
-                            "port-shelf": "Chassis#1"
                         },
-                        "lgx": {
-                            "lgx-device-name": "Some lgx-device-name",
-                            "lgx-port-name": "Some lgx-port-name",
-                            "lgx-port-rack": "000000.00",
-                            "lgx-port-shelf": "00"
-                        }
-                    },
+                        "index": 0
+                    }],
+            "rx-direction": [{
+                "port": {
+                    "port-type": "fixed",
+                    "port-rack": "000000.00",
+                    "port-shelf": "Chassis#1"
+                },
+                "lgx": {
+                    "lgx-device-name": "Some lgx-device-name",
+                    "lgx-port-name": "Some lgx-port-name",
+                    "lgx-port-rack": "000000.00",
+                    "lgx-port-shelf": "00"
+                },
+                "index": 0
+            }],
             "optic-type": "gray"
         },
         "due-date": "2016-11-28T00:00:01Z",
@@ -133,6 +132,7 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(10)
 
     def setUp(self):  # instruction executed before each test method
+        # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
 
     def test_01_connect_xpdrA(self):
@@ -240,12 +240,12 @@ class TransportPCEFulltesting(unittest.TestCase):
     def test_13_change_status_line_port_xpdra(self):
         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
         body = {"ports": [{
-                    "port-name": "1",
-                    "logical-connection-point": "XPDR1-NETWORK1",
-                    "port-type": "CFP2",
-                    "circuit-id": "XPDRA-NETWORK",
-                    "administrative-state": "outOfService",
-                    "port-qual": "xpdr-network"}]}
+            "port-name": "1",
+            "logical-connection-point": "XPDR1-NETWORK1",
+            "port-type": "CFP2",
+            "circuit-id": "XPDRA-NETWORK",
+            "administrative-state": "outOfService",
+            "port-qual": "xpdr-network"}]}
         response = requests.request("PUT", url.format("http://127.0.0.1:8130/restconf"),
                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
@@ -317,12 +317,12 @@ class TransportPCEFulltesting(unittest.TestCase):
     def test_17_restore_status_line_port_xpdra(self):
         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
         body = {"ports": [{
-                    "port-name": "1",
-                    "logical-connection-point": "XPDR1-NETWORK1",
-                    "port-type": "CFP2",
-                    "circuit-id": "XPDRA-NETWORK",
-                    "administrative-state": "inService",
-                    "port-qual": "xpdr-network"}]}
+            "port-name": "1",
+            "logical-connection-point": "XPDR1-NETWORK1",
+            "port-type": "CFP2",
+            "circuit-id": "XPDRA-NETWORK",
+            "administrative-state": "inService",
+            "port-qual": "xpdr-network"}]}
         response = requests.request("PUT", url.format("http://127.0.0.1:8130/restconf"),
                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
@@ -367,12 +367,12 @@ class TransportPCEFulltesting(unittest.TestCase):
     def test_21_change_status_port_roadma_srg(self):
         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
         body = {"ports": [{
-                    "port-name": "C1",
-                    "logical-connection-point": "SRG1-PP1",
-                    "port-type": "client",
-                    "circuit-id": "SRG1",
-                    "administrative-state": "outOfService",
-                    "port-qual": "roadm-external"}]}
+            "port-name": "C1",
+            "logical-connection-point": "SRG1-PP1",
+            "port-type": "client",
+            "circuit-id": "SRG1",
+            "administrative-state": "outOfService",
+            "port-qual": "roadm-external"}]}
         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
@@ -436,12 +436,12 @@ class TransportPCEFulltesting(unittest.TestCase):
     def test_24_restore_status_port_roadma_srg(self):
         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
         body = {"ports": [{
-                    "port-name": "C1",
-                    "logical-connection-point": "SRG1-PP1",
-                    "port-type": "client",
-                    "circuit-id": "SRG1",
-                    "administrative-state": "inService",
-                    "port-qual": "roadm-external"}]}
+            "port-name": "C1",
+            "logical-connection-point": "SRG1-PP1",
+            "port-type": "client",
+            "circuit-id": "SRG1",
+            "administrative-state": "inService",
+            "port-qual": "roadm-external"}]}
         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
@@ -460,12 +460,12 @@ class TransportPCEFulltesting(unittest.TestCase):
     def test_28_change_status_line_port_roadma_deg(self):
         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
         body = {"ports": [{
-                    "port-name": "L1",
-                    "logical-connection-point": "DEG2-TTP-TXRX",
-                    "port-type": "LINE",
-                    "circuit-id": "1",
-                    "administrative-state": "outOfService",
-                    "port-qual": "roadm-external"}]}
+            "port-name": "L1",
+            "logical-connection-point": "DEG2-TTP-TXRX",
+            "port-type": "LINE",
+            "circuit-id": "1",
+            "administrative-state": "outOfService",
+            "port-qual": "roadm-external"}]}
         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
@@ -529,12 +529,12 @@ class TransportPCEFulltesting(unittest.TestCase):
     def test_31_restore_status_line_port_roadma_srg(self):
         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
         body = {"ports": [{
-                    "port-name": "L1",
-                    "logical-connection-point": "DEG2-TTP-TXRX",
-                    "port-type": "LINE",
-                    "circuit-id": "1",
-                    "administrative-state": "inService",
-                    "port-qual": "roadm-external"}]}
+            "port-name": "L1",
+            "logical-connection-point": "DEG2-TTP-TXRX",
+            "port-type": "LINE",
+            "circuit-id": "1",
+            "administrative-state": "inService",
+            "port-qual": "roadm-external"}]}
         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
@@ -553,10 +553,10 @@ class TransportPCEFulltesting(unittest.TestCase):
     def test_35_change_status_line_port_xpdrc(self):
         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
         body = {"ports": [{
-                    "port-name": "1",
-                    "port-type": "CFP2",
-                    "administrative-state": "outOfService",
-                    "port-qual": "xpdr-network"}]}
+            "port-name": "1",
+            "port-type": "CFP2",
+            "administrative-state": "outOfService",
+            "port-qual": "xpdr-network"}]}
         response = requests.request("PUT", url.format("http://127.0.0.1:8154/restconf"),
                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
@@ -620,10 +620,10 @@ class TransportPCEFulltesting(unittest.TestCase):
     def test_38_restore_status_line_port_xpdrc(self):
         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
         body = {"ports": [{
-                    "port-name": "1",
-                    "port-type": "CFP2",
-                    "administrative-state": "inService",
-                    "port-qual": "xpdr-network"}]}
+            "port-name": "1",
+            "port-type": "CFP2",
+            "administrative-state": "inService",
+            "port-qual": "xpdr-network"}]}
         response = requests.request("PUT", url.format("http://127.0.0.1:8154/restconf"),
                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
@@ -642,12 +642,12 @@ class TransportPCEFulltesting(unittest.TestCase):
     def test_42_change_status_port_roadma_srg(self):
         url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C2"
         body = {"ports": [{
-                    "port-name": "C2",
-                    "logical-connection-point": "SRG1-PP2",
-                    "port-type": "client",
-                    "circuit-id": "SRG1",
-                    "administrative-state": "outOfService",
-                    "port-qual": "roadm-external"}]}
+            "port-name": "C2",
+            "logical-connection-point": "SRG1-PP2",
+            "port-type": "client",
+            "circuit-id": "SRG1",
+            "administrative-state": "outOfService",
+            "port-qual": "roadm-external"}]}
         response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
                                     data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
                                     auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
@@ -719,8 +719,7 @@ class TransportPCEFulltesting(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)
Repository for Transport PCE project.