fix some pylint issues
[transportpce.git] / tests / transportpce_tests / 2.2.1 / test_otn_end2end.py
index fdfcc4e92d3c9cf682025e93bcbde950fcb05662..d2208da6d8727f05201635885b552b3b65ce6f09 100644 (file)
@@ -9,6 +9,9 @@
 # http://www.apache.org/licenses/LICENSE-2.0
 ##############################################################################
 
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
 import unittest
 import time
 import requests
@@ -129,6 +132,7 @@ class TransportPCEtesting(unittest.TestCase):
 
     @classmethod
     def tearDownClass(cls):
+        # pylint: disable=not-an-iterable
         for process in cls.processes:
             test_utils.shutdown_process(process)
         print("all processes killed")
@@ -331,7 +335,7 @@ class TransportPCEtesting(unittest.TestCase):
 
     def test_18_check_no_interface_ODU4_spdra(self):
         response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU4")
-        self.assertEqual(response.status_code, requests.codes.not_found)
+        self.assertEqual(response.status_code, requests.codes.conflict)
         res = response.json()
         self.assertIn(
             {"error-type": "application", "error-tag": "data-missing",
@@ -494,12 +498,12 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertEqual(nb_links, 4)
         for link in res['network'][0]['ietf-network-topology:link']:
             linkId = link['link-id']
-            if (linkId == 'OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
-                    linkId == 'OTU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+            if (linkId in ('OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+                           'OTU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
                 self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 0)
                 self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 100000)
-            elif (linkId == 'ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
-                  linkId == 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+            elif (linkId in ('ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+                             'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
                 self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 100000)
                 self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 0)
                 self.assertEqual(link['transportpce-topology:otn-link-type'], 'ODTU4')
@@ -514,10 +518,10 @@ class TransportPCEtesting(unittest.TestCase):
         response = test_utils.get_otn_topo_request()
         res = response.json()
         for node in res['network'][0]['node']:
-            if (node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1'):
+            if node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
-                    if (tp['tp-id'] == 'XPDR1-NETWORK1'):
+                    if tp['tp-id'] == 'XPDR1-NETWORK1':
                         xpdrTpPortConAt = tp['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
                         self.assertEqual(len(xpdrTpPortConAt['ts-pool']), 80)
                         self.assertEqual(len(xpdrTpPortConAt['odtu-tpn-pool'][0]['tpn-pool']), 80)
@@ -745,8 +749,8 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertEqual(nb_links, 4)
         for link in res['network'][0]['ietf-network-topology:link']:
             linkId = link['link-id']
-            if (linkId == 'ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
-                    linkId == 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+            if (linkId in ('ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+                           'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
                 self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 90000)
                 self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 10000)
 
@@ -754,33 +758,19 @@ class TransportPCEtesting(unittest.TestCase):
         response = test_utils.get_otn_topo_request()
         res = response.json()
         for node in res['network'][0]['node']:
-            if (node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1'):
+            if node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
-                    if (tp['tp-id'] == 'XPDR1-NETWORK1'):
+                    if tp['tp-id'] == 'XPDR1-NETWORK1':
                         xpdrTpPortConAt = tp['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
                         self.assertEqual(len(xpdrTpPortConAt['ts-pool']), 72)
-                        tsPoolList = [i for i in range(1, 9)]
+                        tsPoolList = list(range(1, 9))
                         self.assertNotIn(tsPoolList, xpdrTpPortConAt['ts-pool'])
                         self.assertEqual(len(xpdrTpPortConAt['odtu-tpn-pool'][0]['tpn-pool']), 79)
                         self.assertNotIn(1, xpdrTpPortConAt['odtu-tpn-pool'][0]['tpn-pool'])
 
     def test_41_delete_10GE_service(self):
-        url = "{}/operations/org-openroadm-service:service-delete"
-        data = {"input": {
-                "sdnc-request-header": {
-                    "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
-                    "rpc-action": "service-delete",
-                    "request-system-id": "appname",
-                    "notification-url": "http://localhost:8585/NotificationServer/notify"
-                },
-                "service-delete-req-info": {
-                    "service-name": "service1-10GE",
-                    "tail-retention": "no"
-                }
-                }
-                }
-        response = test_utils.post_request(url, data)
+        response = test_utils.service_delete_request("service1-10GE")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Renderer service delete in progress',
@@ -803,15 +793,15 @@ class TransportPCEtesting(unittest.TestCase):
 
     def test_44_check_no_interface_ODU2E_NETWORK_spdra(self):
         response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU2e-service1")
-        self.assertEqual(response.status_code, requests.codes.not_found)
+        self.assertEqual(response.status_code, requests.codes.conflict)
 
     def test_45_check_no_interface_ODU2E_CLIENT_spdra(self):
         response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ODU2e-service1")
-        self.assertEqual(response.status_code, requests.codes.not_found)
+        self.assertEqual(response.status_code, requests.codes.conflict)
 
     def test_46_check_no_interface_10GE_CLIENT_spdra(self):
         response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ETHERNET10G")
-        self.assertEqual(response.status_code, requests.codes.not_found)
+        self.assertEqual(response.status_code, requests.codes.conflict)
 
     def test_47_check_otn_topo_links(self):
         response = test_utils.get_otn_topo_request()
@@ -821,8 +811,8 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertEqual(nb_links, 4)
         for link in res['network'][0]['ietf-network-topology:link']:
             linkId = link['link-id']
-            if (linkId == 'ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
-                    linkId == 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+            if (linkId in ('ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+                           'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
                 self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 100000)
                 self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 0)
 
@@ -833,27 +823,13 @@ class TransportPCEtesting(unittest.TestCase):
             if (node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1'):
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
-                    if (tp['tp-id'] == 'XPDR1-NETWORK1'):
+                    if tp['tp-id'] == 'XPDR1-NETWORK1':
                         xpdrTpPortConAt = tp['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
                         self.assertEqual(len(xpdrTpPortConAt['ts-pool']), 80)
                         self.assertEqual(len(xpdrTpPortConAt['odtu-tpn-pool'][0]['tpn-pool']), 80)
 
     def test_49_delete_ODU4_service(self):
-        url = "{}/operations/org-openroadm-service:service-delete"
-        data = {"input": {
-                "sdnc-request-header": {
-                    "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
-                    "rpc-action": "service-delete",
-                    "request-system-id": "appname",
-                    "notification-url": "http://localhost:8585/NotificationServer/notify"
-                },
-                "service-delete-req-info": {
-                    "service-name": "service1-ODU4",
-                    "tail-retention": "no"
-                }
-                }
-                }
-        response = test_utils.post_request(url, data)
+        response = test_utils.service_delete_request("service1-ODU4")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Renderer service delete in progress',
@@ -869,7 +845,7 @@ class TransportPCEtesting(unittest.TestCase):
 
     def test_51_check_no_interface_ODU4_spdra(self):
         response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU4")
-        self.assertEqual(response.status_code, requests.codes.not_found)
+        self.assertEqual(response.status_code, requests.codes.conflict)
 
     def test_52_check_otn_topo_links(self):
         self.test_22_check_otn_topo_otu4_links()
@@ -878,30 +854,16 @@ class TransportPCEtesting(unittest.TestCase):
         response = test_utils.get_otn_topo_request()
         res = response.json()
         for node in res['network'][0]['node']:
-            if (node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1'):
+            if node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
-                    if (tp['tp-id'] == 'XPDR1-NETWORK1'):
+                    if tp['tp-id'] == 'XPDR1-NETWORK1':
                         xpdrTpPortConAt = tp['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
                         self.assertNotIn('ts-pool', dict.keys(xpdrTpPortConAt))
                         self.assertNotIn('odtu-tpn-pool', dict.keys(xpdrTpPortConAt))
 
     def test_54_delete_OCH_OTU4_service(self):
-        url = "{}/operations/org-openroadm-service:service-delete"
-        data = {"input": {
-                "sdnc-request-header": {
-                    "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
-                    "rpc-action": "service-delete",
-                    "request-system-id": "appname",
-                    "notification-url": "http://localhost:8585/NotificationServer/notify"
-                },
-                "service-delete-req-info": {
-                    "service-name": "service1-OCH-OTU4",
-                    "tail-retention": "no"
-                }
-                }
-                }
-        response = test_utils.post_request(url, data)
+        response = test_utils.service_delete_request("service1-OCH-OTU4")
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Renderer service delete in progress',
@@ -910,7 +872,7 @@ class TransportPCEtesting(unittest.TestCase):
 
     def test_55_get_no_service(self):
         response = test_utils.get_service_list_request("")
-        self.assertEqual(response.status_code, requests.codes.not_found)
+        self.assertEqual(response.status_code, requests.codes.conflict)
         res = response.json()
         self.assertIn(
             {"error-type": "application", "error-tag": "data-missing",
@@ -920,11 +882,11 @@ class TransportPCEtesting(unittest.TestCase):
 
     def test_56_check_no_interface_OTU4_spdra(self):
         response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-OTU")
-        self.assertEqual(response.status_code, requests.codes.not_found)
+        self.assertEqual(response.status_code, requests.codes.conflict)
 
     def test_57_check_no_interface_OCH_spdra(self):
         response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-1")
-        self.assertEqual(response.status_code, requests.codes.not_found)
+        self.assertEqual(response.status_code, requests.codes.conflict)
 
     def test_58_getLinks_OtnTopology(self):
         response = test_utils.get_otn_topo_request()