Manage 100GE on XPDR with a list of xponder
[transportpce.git] / tests / transportpce_tests / 1.2.1 / test06_end2end.py
index 3f7ca3b52aff45306c1664bf6eba47d95da1c572..7bfc70c27a3c25bbb2b49067bceafaf0858cdd15 100644 (file)
@@ -15,9 +15,13 @@ import base64
 import time
 import unittest
 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):
     cr_serv_sample_data = {"input": {
@@ -36,40 +40,8 @@ class TransportPCEFulltesting(unittest.TestCase):
             "node-id": "XPDRA01",
             "service-format": "Ethernet",
             "clli": "SNJSCAMCJP8",
-            "tx-direction": {
-                "port": {
-                    "port-device-name":
-                        "ROUTER_SNJSCAMCJP8_000000.00_00",
-                    "port-type": "router",
-                    "port-name": "Gigabit Ethernet_Tx.ge-5/0/0.0",
-                    "port-rack": "000000.00",
-                    "port-shelf": "00"
-                },
-                "lgx": {
-                    "lgx-device-name":
-                        "LGX Panel_SNJSCAMCJP8_000000.00_00",
-                    "lgx-port-name": "LGX Back.3",
-                    "lgx-port-rack": "000000.00",
-                    "lgx-port-shelf": "00"
-                }
-            },
-            "rx-direction": {
-                "port": {
-                    "port-device-name":
-                        "ROUTER_SNJSCAMCJP8_000000.00_00",
-                    "port-type": "router",
-                    "port-name": "Gigabit Ethernet_Rx.ge-5/0/0.0",
-                    "port-rack": "000000.00",
-                    "port-shelf": "00"
-                },
-                "lgx": {
-                    "lgx-device-name":
-                        "LGX Panel_SNJSCAMCJP8_000000.00_00",
-                    "lgx-port-name": "LGX Back.4",
-                    "lgx-port-rack": "000000.00",
-                    "lgx-port-shelf": "00"
-                }
-            },
+            "tx-direction": {},
+            "rx-direction": {},
             "optic-type": "gray"
         },
         "service-z-end": {
@@ -77,40 +49,8 @@ class TransportPCEFulltesting(unittest.TestCase):
             "node-id": "XPDRC01",
             "service-format": "Ethernet",
             "clli": "SNJSCAMCJT4",
-            "tx-direction": {
-                "port": {
-                    "port-device-name":
-                        "ROUTER_SNJSCAMCJT4_000000.00_00",
-                    "port-type": "router",
-                    "port-name": "Gigabit Ethernet_Tx.ge-1/0/0.0",
-                    "port-rack": "000000.00",
-                    "port-shelf": "00"
-                },
-                "lgx": {
-                    "lgx-device-name":
-                        "LGX Panel_SNJSCAMCJT4_000000.00_00",
-                    "lgx-port-name": "LGX Back.29",
-                    "lgx-port-rack": "000000.00",
-                    "lgx-port-shelf": "00"
-                }
-            },
-            "rx-direction": {
-                "port": {
-                    "port-device-name":
-                        "ROUTER_SNJSCAMCJT4_000000.00_00",
-                    "port-type": "router",
-                    "port-name": "Gigabit Ethernet_Rx.ge-1/0/0.0",
-                    "port-rack": "000000.00",
-                    "port-shelf": "00"
-                },
-                "lgx": {
-                    "lgx-device-name":
-                        "LGX Panel_SNJSCAMCJT4_000000.00_00",
-                    "lgx-port-name": "LGX Back.30",
-                    "lgx-port-rack": "000000.00",
-                    "lgx-port-shelf": "00"
-                }
-            },
+            "tx-direction": {},
+            "rx-direction": {},
             "optic-type": "gray"
         },
         "due-date": "2016-11-28T00:00:01Z",
@@ -137,6 +77,7 @@ class TransportPCEFulltesting(unittest.TestCase):
         print("all processes killed")
 
     def setUp(self):  # instruction executed before each test method
+        # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
 
     #  connect netconf devices
@@ -249,7 +190,8 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(2)
 
     def test_13_check_xc1_ROADMA(self):
-        response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
+        response = test_utils.check_netconf_node_request(
+            "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
@@ -271,7 +213,8 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(5)
 
     def test_14_check_xc1_ROADMC(self):
-        response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
+        response = test_utils.check_netconf_node_request(
+            "ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
@@ -299,12 +242,12 @@ class TransportPCEFulltesting(unittest.TestCase):
         liste_tp = res['node'][0]['ietf-network-topology:termination-point']
         for ele in liste_tp:
             if ele['tp-id'] == 'XPDR1-NETWORK1':
-                self.assertEqual({u'frequency': 196.1,
-                                  u'width': 40},
+                self.assertEqual({'frequency': 196.1,
+                                  'width': 40},
                                  ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
-            if ele['tp-id'] == 'XPDR1-CLIENT2' or ele['tp-id'] == 'XPDR1-CLIENT1':
+            elif ele['tp-id'] in ('XPDR1-CLIENT2', 'XPDR1-CLIENT1'):
                 self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
-            if ele['tp-id'] == 'XPDR1-NETWORK2':
+            elif ele['tp-id'] == 'XPDR1-NETWORK2':
                 self.assertNotIn('org-openroadm-network-topology:xpdr-network-attributes', dict.keys(ele))
         time.sleep(3)
 
@@ -323,7 +266,7 @@ class TransportPCEFulltesting(unittest.TestCase):
                     ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
                 freq_map_array = [int(x) for x in freq_map]
                 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
-            if ele['tp-id'] == 'SRG1-PP2-TXRX':
+            elif ele['tp-id'] == 'SRG1-PP2-TXRX':
                 self.assertNotIn('avail-freq-maps', dict.keys(ele))
         time.sleep(3)
 
@@ -342,11 +285,11 @@ class TransportPCEFulltesting(unittest.TestCase):
                     ele['org-openroadm-network-topology:ctp-attributes']['avail-freq-maps'][0]['freq-map'])
                 freq_map_array = [int(x) for x in freq_map]
                 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
-            if ele['tp-id'] == 'DEG2-TTP-TXRX':
-                    freq_map = base64.b64decode(
-                        ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
-                    freq_map_array = [int(x) for x in freq_map]
-                    self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
+            elif ele['tp-id'] == 'DEG2-TTP-TXRX':
+                freq_map = base64.b64decode(
+                    ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
+                freq_map_array = [int(x) for x in freq_map]
+                self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
         time.sleep(3)
 
     def test_18_connect_xprdA_N2_to_roadmA_PP2(self):
@@ -411,7 +354,8 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(1)
 
     def test_24_check_xc2_ROADMA(self):
-        response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP2-TXRX-753:760")
+        response = test_utils.check_netconf_node_request(
+            "ROADMA01", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP2-TXRX-753:760")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
@@ -437,14 +381,14 @@ class TransportPCEFulltesting(unittest.TestCase):
         liste_tp = res['node'][0]['ietf-network-topology:termination-point']
         for ele in liste_tp:
             if ele['tp-id'] == 'XPDR1-NETWORK1':
-                self.assertEqual({u'frequency': 196.1,
-                                  u'width': 40},
+                self.assertEqual({'frequency': 196.1,
+                                  'width': 40},
                                  ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
-            if ele['tp-id'] == 'XPDR1-NETWORK2':
-                self.assertEqual({u'frequency': 196.05,
-                                  u'width': 40},
+            elif ele['tp-id'] == 'XPDR1-NETWORK2':
+                self.assertEqual({'frequency': 196.05,
+                                  'width': 40},
                                  ele['org-openroadm-network-topology:xpdr-network-attributes']['wavelength'])
-            if ele['tp-id'] == 'XPDR1-CLIENT1' or ele['tp-id'] == 'XPDR1-CLIENT2':
+            elif ele['tp-id'] in ('XPDR1-CLIENT1', 'XPDR1-CLIENT2'):
                 self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
         time.sleep(10)
 
@@ -465,13 +409,13 @@ class TransportPCEFulltesting(unittest.TestCase):
                 freq_map_array = [int(x) for x in freq_map]
                 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
                 self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
-            if ele['tp-id'] == 'SRG1-PP2-TXRX':
+            elif ele['tp-id'] == 'SRG1-PP2-TXRX':
                 freq_map = base64.b64decode(
                     ele['org-openroadm-network-topology:pp-attributes']['avail-freq-maps'][0]['freq-map'])
                 freq_map_array = [int(x) for x in freq_map]
                 self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
                 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
-            if ele['tp-id'] == 'SRG1-PP3-TXRX':
+            elif ele['tp-id'] == 'SRG1-PP3-TXRX':
                 self.assertNotIn('org-openroadm-network-topology:pp-attributes', dict.keys(ele))
         time.sleep(10)
 
@@ -492,7 +436,7 @@ class TransportPCEFulltesting(unittest.TestCase):
                 freq_map_array = [int(x) for x in freq_map]
                 self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
                 self.assertEqual(freq_map_array[94], 0, "Lambda 2 should not be available")
-            if ele['tp-id'] == 'DEG2-TTP-TXRX':
+            elif ele['tp-id'] == 'DEG2-TTP-TXRX':
                 freq_map = base64.b64decode(
                     ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
                 freq_map_array = [int(x) for x in freq_map]
@@ -559,21 +503,18 @@ class TransportPCEFulltesting(unittest.TestCase):
         res = response.json()
         liste_tp = res['node'][0]['ietf-network-topology:termination-point']
         for ele in liste_tp:
-            if ((ele[u'org-openroadm-common-network:tp-type'] ==
-                 'XPONDER-CLIENT')
-                    and (ele['tp-id'] == 'XPDR1-CLIENT1' or ele[
-                        'tp-id'] == 'XPDR1-CLIENT3')):
+            if (ele['org-openroadm-common-network:tp-type'] == 'XPONDER-CLIENT'
+                    and ele['tp-id'] in ('XPDR1-CLIENT1', 'XPDR1-CLIENT3')):
                 self.assertNotIn(
                     'org-openroadm-network-topology:xpdr-client-attributes',
                     dict.keys(ele))
-            elif (ele[u'org-openroadm-common-network:tp-type'] ==
-                  'XPONDER-NETWORK'):
-                self.assertIn(u'tail-equipment-id', dict.keys(
-                    ele[u'org-openroadm-network-topology:'
-                        u'xpdr-network-attributes']))
+            elif ele['org-openroadm-common-network:tp-type'] == 'XPONDER-NETWORK':
+                self.assertIn('tail-equipment-id', dict.keys(
+                    ele['org-openroadm-network-topology:'
+                        'xpdr-network-attributes']))
                 self.assertNotIn('wavelength', dict.keys(
-                    ele[u'org-openroadm-network-topology:'
-                        u'xpdr-network-attributes']))
+                    ele['org-openroadm-network-topology:'
+                        'xpdr-network-attributes']))
         time.sleep(10)
 
     def test_34_check_topo_ROADMA_SRG1(self):
@@ -620,7 +561,7 @@ class TransportPCEFulltesting(unittest.TestCase):
                 freq_map_array = [int(x) for x in freq_map]
                 self.assertEqual(freq_map_array[95], 255, "Lambda 1 should be available")
                 self.assertEqual(freq_map_array[94], 255, "Lambda 2 should be available")
-            if ele['tp-id'] == 'DEG2-TTP-TXRX':
+            elif ele['tp-id'] == 'DEG2-TTP-TXRX':
                 freq_map = base64.b64decode(
                     ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
                 freq_map_array = [int(x) for x in freq_map]
@@ -660,7 +601,8 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(1)
 
     def test_38_check_xc1_ROADMA(self):
-        response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
+        response = test_utils.check_netconf_node_request(
+            "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
@@ -682,7 +624,8 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(7)
 
     def test_39_check_xc1_ROADMC(self):
-        response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
+        response = test_utils.check_netconf_node_request(
+            "ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
@@ -734,7 +677,8 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(2)
 
     def test_42_check_xc2_ROADMA(self):
-        response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP2-TXRX-DEG1-TTP-TXRX-753:760")
+        response = test_utils.check_netconf_node_request(
+            "ROADMA01", "roadm-connections/SRG1-PP2-TXRX-DEG1-TTP-TXRX-753:760")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
@@ -806,6 +750,7 @@ class TransportPCEFulltesting(unittest.TestCase):
         self.test_35_check_topo_ROADMA_DEG1()
 
     def test_49_loop_create_eth_service(self):
+        # pylint: disable=consider-using-f-string
         for i in range(1, 6):
             print("iteration number {}".format(i))
             print("eth service creation")
@@ -823,6 +768,7 @@ class TransportPCEFulltesting(unittest.TestCase):
             response = test_utils.service_delete_request("service1")
             time.sleep(5)
 
+        # pylint: disable=consider-using-f-string
         for i in range(1, 6):
             print("iteration number {}".format(i))
             print("oc service creation")