Merge changes I018b69e8,Ia99c607c,Idc241d22
authorGuillaume Lambert <guillaume.lambert@orange.com>
Wed, 1 Dec 2021 10:44:57 +0000 (10:44 +0000)
committerGerrit Code Review <gerrit@opendaylight.org>
Wed, 1 Dec 2021 10:44:57 +0000 (10:44 +0000)
* changes:
  Fix more and more pylint warnings
  Fix more pylint warnings
  Fix some pylint warnings

29 files changed:
tests/transportpce_tests/1.2.1/test01_portmapping.py
tests/transportpce_tests/1.2.1/test02_topo_portmapping.py
tests/transportpce_tests/1.2.1/test03_topology.py
tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py
tests/transportpce_tests/1.2.1/test05_olm.py
tests/transportpce_tests/1.2.1/test06_end2end.py
tests/transportpce_tests/2.2.1/test01_portmapping.py
tests/transportpce_tests/2.2.1/test02_topo_portmapping.py
tests/transportpce_tests/2.2.1/test03_topology.py
tests/transportpce_tests/2.2.1/test04_otn_topology.py
tests/transportpce_tests/2.2.1/test05_flex_grid.py
tests/transportpce_tests/2.2.1/test06_renderer_service_path_nominal.py
tests/transportpce_tests/2.2.1/test07_otn_renderer.py
tests/transportpce_tests/2.2.1/test08_otn_sh_renderer.py
tests/transportpce_tests/2.2.1/test09_olm.py
tests/transportpce_tests/2.2.1/test10_tapi.py
tests/transportpce_tests/2.2.1/test11_otn_end2end.py
tests/transportpce_tests/2.2.1/test12_end2end.py
tests/transportpce_tests/2.2.1/test13_tapi_full_multi_layer.py
tests/transportpce_tests/2.2.1/test14_otn_switch_end2end.py
tests/transportpce_tests/7.1/test01_portmapping.py
tests/transportpce_tests/common/test_utils.py
tests/transportpce_tests/common/test_utils_rfc8040.py
tests/transportpce_tests/hybrid/test01_device_change_notifications.py
tests/transportpce_tests/hybrid/test02_B100G_end2end.py
tests/transportpce_tests/pce/test01_pce.py
tests/transportpce_tests/pce/test02_pce_400G.py
tests/transportpce_tests/with_docker/test01_gnpy.py
tests/transportpce_tests/with_docker/test02_nbinotifications.py

index 013004c379f155a84bee47b05d399976ca5d5759..9d28b0c4ded65f91c722deb28cc70e3b4ee21a0d 100644 (file)
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils_rfc8040  # nopep8
 
 
@@ -38,6 +41,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
         print("all processes killed")
 
     def setUp(self):
+        # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
         time.sleep(10)
 
index 093f9a8093c5392835f8d678bd674ddb332fea1c..a2f88b5f6ad8b4d6547f6d4ec542d444fa64ba83 100644 (file)
 import time
 import unittest
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils_rfc8040  # nopep8
 
 
index 34e7c3945613e2b555b0e170368249f24f5d47ea..d3050f16efc868478d04cddf1be04ce7be7ce0f3 100644 (file)
 import time
 import unittest
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -158,8 +161,8 @@ class TransportPCETopologyTesting(unittest.TestCase):
                 (check_list[linkType]).remove(linkId)
             else:
                 self.assertFalse(True)
-        for link_type in check_list:
-            self.assertEqual(len(check_list[link_type]), 0)
+        for val in check_list.values():
+            self.assertEqual(len(val), 0)
 
     def test_05_getNodes_OpenRoadmTopology(self):
         # pylint: disable=redundant-unittest-assert
@@ -320,8 +323,8 @@ class TransportPCETopologyTesting(unittest.TestCase):
                 (check_list[linkType]).remove(linkId)
             else:
                 self.assertFalse(True)
-        for link_type in check_list:
-            self.assertEqual(len(check_list[link_type]), 0)
+        for val in check_list.values():
+            self.assertEqual(len(val), 0)
 
     def test_13_connect_ROADMC(self):
         response = test_utils.mount_device("ROADMC01", ('roadmc', self.NODE_VERSION))
@@ -434,8 +437,8 @@ class TransportPCETopologyTesting(unittest.TestCase):
                 (check_list[linkType]).remove(linkId)
             else:
                 self.assertFalse(True)
-        for link_type in check_list:
-            self.assertEqual(len(check_list[link_type]), 0)
+        for val in check_list.values():
+            self.assertEqual(len(val), 0)
 
     def test_19_getLinkOmsAttributesOpenRoadmTopology(self):
         response = test_utils.get_ordm_topo_request("")
@@ -843,8 +846,8 @@ class TransportPCETopologyTesting(unittest.TestCase):
                 (check_list[linkType]).remove(linkId)
             else:
                 roadmtoroadmLink += 1
-        for link_type in check_list:
-            self.assertEqual(len(check_list[link_type]), 0)
+        for val in check_list.values():
+            self.assertEqual(len(val), 0)
         self.assertEqual(roadmtoroadmLink, 6)
         for i in range(0, nbLink):
             self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]
index 37261ad560020ce05af83458031931eb7d549600..485ce41695be5e1410b91994228bed8db12d34f7 100644 (file)
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -38,6 +41,7 @@ class TransportPCERendererTesting(unittest.TestCase):
         print("all processes killed")
 
     def setUp(self):
+        # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
         time.sleep(10)
 
index e6f43e7fa7f0c4fa00117902921f2ec3355cd00f..050609975f3d51e44e2691a54ca4aaf34a348ec1 100644 (file)
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -41,6 +44,7 @@ class TransportOlmTesting(unittest.TestCase):
         print("all processes killed")
 
     def setUp(self):
+        # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
         time.sleep(1)
 
index 9d9fe8c3a52c4df512315abf976c0b92a0c68e2b..c8f4eac61a76ce552fb7e2d1fd6259e0cfea38c5 100644 (file)
@@ -15,8 +15,11 @@ import base64
 import time
 import unittest
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -138,6 +141,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
@@ -813,6 +817,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")
@@ -830,6 +835,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")
index a48709dbb37075708f18010efb8ca76715775a68..fa650b74259640117b8370db19c8a04c799e9355 100644 (file)
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils_rfc8040  # nopep8
 
 
@@ -38,6 +41,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase):
         print("all processes killed")
 
     def setUp(self):
+        # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
         time.sleep(10)
 
index a2b8bf189ea7bae7f4a22180d09603cf77912046..06e83473b9efbfd6adb04b182477a3768747454f 100644 (file)
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils_rfc8040  # nopep8
 
 
@@ -54,6 +57,7 @@ class TransportPCEtesting(unittest.TestCase):
         nbMappings = 0
         for i in range(0, len(firstEntry)):
             nodeId = firstEntry[i]['node-id']
+            # pylint: disable=consider-using-f-string
             print("nodeId={}".format(nodeId))
             nodeMapId = nodeId.split("-")[0] + "-" + nodeId.split("-")[1]
             print("nodeMapId={}".format(nodeMapId))
index 3ba43d79cdcf7ba9604e35d33984ef58bc0aaad9..43bdbd6731bcf60b2625ebb3a49a7b9a86f3630c 100644 (file)
@@ -17,8 +17,11 @@ import unittest
 import time
 import logging
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -160,8 +163,8 @@ class TransportPCEtesting(unittest.TestCase):
                 (check_list[linkType]).remove(linkId)
             else:
                 self.assertFalse(True)
-        for link_type in check_list:
-            self.assertEqual(len(check_list[link_type]), 0)
+        for val in check_list.values():
+            self.assertEqual(len(val), 0)
 
     def test_05_getNodes_OpenRoadmTopology(self):
         # pylint: disable=redundant-unittest-assert
@@ -320,8 +323,8 @@ class TransportPCEtesting(unittest.TestCase):
                 (check_list[linkType]).remove(linkId)
             else:
                 self.assertFalse(True)
-        for link_type in check_list:
-            self.assertEqual(len(check_list[link_type]), 0)
+        for val in check_list.values():
+            self.assertEqual(len(val), 0)
 
     def test_13_connect_ROADMC(self):
         response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION))
@@ -435,8 +438,8 @@ class TransportPCEtesting(unittest.TestCase):
                 (check_list[linkType]).remove(linkId)
             else:
                 self.assertFalse(True)
-        for link_type in check_list:
-            self.assertEqual(len(check_list[link_type]), 0)
+        for val in check_list.values():
+            self.assertEqual(len(val), 0)
 
     def test_19_getLinkOmsAttributesOpenRoadmTopology(self):
         response = test_utils.get_ordm_topo_request("")
@@ -847,8 +850,8 @@ class TransportPCEtesting(unittest.TestCase):
                 (check_list[linkType]).remove(linkId)
             else:
                 roadmtoroadmLink += 1
-        for link_type in check_list:
-            self.assertEqual(len(check_list[link_type]), 0)
+        for val in check_list.values():
+            self.assertEqual(len(val), 0)
         self.assertEqual(roadmtoroadmLink, 6)
         for i in range(0, nbLink):
             self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]
index 018a92f6172992a5986af3000d8271921783222f..41b1f65b045ee05e4a54b79ab4afd9e7e87b6358 100644 (file)
@@ -16,8 +16,11 @@ import unittest
 import time
 import logging
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -182,6 +185,7 @@ class TransportPCEtesting(unittest.TestCase):
                     tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
                     if tpType == 'XPONDER-CLIENT':
                         client += 1
+                        # pylint: disable=consider-using-f-string
                         print("tpId = {}".format(tpId))
                         print("tp= {}".format(res['network'][0]['node'][i]
                               ['ietf-network-topology:termination-point'][j]))
index 2719a91f3c2288ff887d43d58244cadeaf03d40c..73411e7952c95203ac6d45acd9b90d2da240a2c4 100644 (file)
@@ -9,11 +9,16 @@
 # http://www.apache.org/licenses/LICENSE-2.0
 ##############################################################################
 
+# pylint: disable=no-member
+
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -29,11 +34,13 @@ class TransportPCEPortMappingTesting(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")
 
     def setUp(self):
+        # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
         time.sleep(10)
 
index 853daed46858cd528e22c90ba61a091fc4d5b6cf..1abc1e31d496099f0a996b8bded4a3fedd3f1adc 100644 (file)
 import unittest
 # from unittest.result import failfast
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
index 3139986a00299c545fe8b4d016aa6a188276915c..3089c7fc6c22788d049497846fd81009c3d69c07 100644 (file)
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
index dbfbec5f631fd802a0b68967132602f82b2b3dba..18c4516287c2d39d57d8a44cd3f4f99b0a15aa96 100644 (file)
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
index 8672fcbb3d9ef99c1fa163a590580c4be818a925..63ced0f742a591c969d35798ae5ebf0662e873c5 100644 (file)
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -41,6 +44,7 @@ class TransportOlmTesting(unittest.TestCase):
         print("all processes killed")
 
     def setUp(self):
+        # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
         time.sleep(1)
 
index 9251c2acb697283779f69758c414e2bd70ce8043..6b5ef4b41b3898b666619b5a08a51a8438deb709 100644 (file)
 # 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/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -173,6 +176,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):
@@ -771,8 +775,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
 
index 385887c47d2e302ed3ae588682d047785b988cd0..9d6bcd84b6efe67fdd52275c87977655d6246d57 100644 (file)
 
 # pylint: disable=no-member
 # pylint: disable=too-many-public-methods
+# pylint: disable=too-many-lines
 
 import base64
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -562,7 +566,7 @@ 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 node['node-id'] == 'SPDR-SC1-XPDR1':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
                     if tp['tp-id'] == 'XPDR1-NETWORK1':
@@ -812,7 +816,7 @@ 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 node['node-id'] == 'SPDR-SC1-XPDR1':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
                     if tp['tp-id'] == 'XPDR1-NETWORK1':
@@ -882,7 +886,7 @@ 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 node['node-id'] == 'SPDR-SC1-XPDR1':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
                     if tp['tp-id'] == 'XPDR1-NETWORK1':
@@ -918,7 +922,7 @@ 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 node['node-id'] == 'SPDR-SC1-XPDR1':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
                     if tp['tp-id'] == 'XPDR1-NETWORK1':
@@ -1047,6 +1051,7 @@ class TransportPCEtesting(unittest.TestCase):
         time.sleep(2)
 
     def test_66_create_OCH_OTU4_service_2(self):
+        # pylint: disable=line-too-long
         self.cr_serv_sample_data["input"]["service-name"] = "service2-OCH-OTU4"
         self.cr_serv_sample_data["input"]["connection-type"] = "infrastructure"
         self.cr_serv_sample_data["input"]["service-a-end"]["service-rate"] = "100"
@@ -1086,6 +1091,7 @@ class TransportPCEtesting(unittest.TestCase):
         time.sleep(2)
 
     def test_68_create_ODU4_service_2(self):
+        # pylint: disable=line-too-long
         self.cr_serv_sample_data["input"]["service-name"] = "service2-ODU4"
         self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "ODU"
         self.cr_serv_sample_data["input"]["service-a-end"]["tx-direction"]["port"]["port-device-name"] = "SPDR-SA1-XPDR3"
@@ -1125,6 +1131,7 @@ class TransportPCEtesting(unittest.TestCase):
         time.sleep(2)
 
     def test_70_create_1GE_service(self):
+        # pylint: disable=line-too-long
         self.cr_serv_sample_data["input"]["service-name"] = "service1-1GE"
         self.cr_serv_sample_data["input"]["connection-type"] = "service"
         self.cr_serv_sample_data["input"]["service-a-end"]["service-rate"] = "1"
@@ -1365,7 +1372,7 @@ 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-XPDR3' or 'SPDR-SC1-XPDR3':
+            if node['node-id'] == 'SPDR-SA1-XPDR3' or node['node-id'] == 'SPDR-SC1-XPDR3':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
                     if tp['tp-id'] == 'XPDR3-NETWORK1':
@@ -1435,7 +1442,7 @@ 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-XPDR3' or 'SPDR-SC1-XPDR3'):
+            if node['node-id'] == 'SPDR-SA1-XPDR3' or node['node-id'] == 'SPDR-SC1-XPDR3':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
                     if tp['tp-id'] == 'XPDR3-NETWORK1':
index 97ee94d1b7753c1c3efda9316c4d058856e76a2b..9472e3892057e9b3e9591257975cb98ce5cc666e 100644 (file)
@@ -15,8 +15,11 @@ import base64
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -131,6 +134,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]))
 
     def test_01_connect_xpdrA(self):
@@ -763,6 +767,7 @@ class TransportPCEFulltesting(unittest.TestCase):
 
     def test_49_loop_create_eth_service(self):
         for i in range(1, 6):
+            # pylint: disable=consider-using-f-string
             print("iteration number {}".format(i))
             print("eth service creation")
             self.test_11_create_eth_service1()
@@ -780,6 +785,7 @@ class TransportPCEFulltesting(unittest.TestCase):
             time.sleep(5)
 
         for i in range(1, 6):
+            # pylint: disable=consider-using-f-string
             print("iteration number {}".format(i))
             print("oc service creation")
             self.test_36_create_oc_service1()
index 4901b377c6c4bf88e290d65b9026b0dee9fe5148..75225611032bc4e5c1457cc2b87e1ab4d22c0088 100644 (file)
@@ -16,8 +16,11 @@ import os
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -82,6 +85,7 @@ class TransportPCEtesting(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
+        # pylint: disable=unsubscriptable-object
         cls.init_failed = False
         os.environ['JAVA_MIN_MEM'] = '1024M'
         os.environ['JAVA_MAX_MEM'] = '4096M'
@@ -252,8 +256,9 @@ class TransportPCEtesting(unittest.TestCase):
         time.sleep(self.WAITING)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
-        global service_pm_uuid
-        service_pm_uuid = res['output']['service']['uuid']
+        global SERVICE_PM_UUID
+        SERVICE_PM_UUID = res['output']['service']['uuid']
+        # pylint: disable=consider-using-f-string
         print("photonic media service uuid : {}".format(res['output']['service']['uuid']))
 
         input_dict_1 = {'administrative-state': 'LOCKED',
@@ -278,13 +283,13 @@ class TransportPCEtesting(unittest.TestCase):
 
     def test_16_get_service_PhotonicMedia(self):
         response = test_utils.get_service_list_request(
-            "services/" + str(service_pm_uuid))
+            "services/" + str(SERVICE_PM_UUID))
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertEqual(
             res['services'][0]['administrative-state'], 'inService')
         self.assertEqual(
-            res['services'][0]['service-name'], service_pm_uuid)
+            res['services'][0]['service-name'], SERVICE_PM_UUID)
         self.assertEqual(
             res['services'][0]['connection-type'], 'infrastructure')
         self.assertEqual(
@@ -293,6 +298,7 @@ class TransportPCEtesting(unittest.TestCase):
 
 # test create connectivity service from spdrA to spdrC for odu
     def test_17_create_connectivity_service_ODU(self):
+        # pylint: disable=line-too-long
         self.cr_serv_sample_data["input"]["end-point"][0]["layer-protocol-name"] = "ODU"
         self.cr_serv_sample_data["input"]["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = "eecbfa6e-57ab-3651-9606-c22c8ce73f18"
         self.cr_serv_sample_data["input"]["end-point"][1]["layer-protocol-name"] = "ODU"
@@ -303,8 +309,9 @@ class TransportPCEtesting(unittest.TestCase):
         time.sleep(self.WAITING)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
-        global service_odu_uuid
-        service_odu_uuid = res['output']['service']['uuid']
+        global SERVICE_ODU_UUID
+        SERVICE_ODU_UUID = res['output']['service']['uuid']
+        # pylint: disable=consider-using-f-string
         print("odu service uuid : {}".format(res['output']['service']['uuid']))
 
         input_dict_1 = {'administrative-state': 'LOCKED',
@@ -329,13 +336,13 @@ class TransportPCEtesting(unittest.TestCase):
 
     def test_18_get_service_ODU(self):
         response = test_utils.get_service_list_request(
-            "services/" + str(service_odu_uuid))
+            "services/" + str(SERVICE_ODU_UUID))
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertEqual(
             res['services'][0]['administrative-state'], 'inService')
         self.assertEqual(
-            res['services'][0]['service-name'], service_odu_uuid)
+            res['services'][0]['service-name'], SERVICE_ODU_UUID)
         self.assertEqual(
             res['services'][0]['connection-type'], 'infrastructure')
         self.assertEqual(
@@ -344,6 +351,7 @@ class TransportPCEtesting(unittest.TestCase):
 
 # test create connectivity service from spdrA to spdrC for dsr
     def test_19_create_connectivity_service_DSR(self):
+        # pylint: disable=line-too-long
         self.cr_serv_sample_data["input"]["end-point"][0]["layer-protocol-name"] = "DSR"
         self.cr_serv_sample_data["input"]["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = "c14797a0-adcc-3875-a1fe-df8949d1a2d7"
         self.cr_serv_sample_data["input"]["end-point"][1]["layer-protocol-name"] = "DSR"
@@ -355,8 +363,9 @@ class TransportPCEtesting(unittest.TestCase):
         time.sleep(self.WAITING)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
-        global service_dsr_uuid
-        service_dsr_uuid = res['output']['service']['uuid']
+        global SERVICE_DSR_UUID
+        SERVICE_DSR_UUID = res['output']['service']['uuid']
+        # pylint: disable=consider-using-f-string
         print("dsr service uuid : {}".format(res['output']['service']['uuid']))
 
         input_dict_1 = {'administrative-state': 'LOCKED',
@@ -384,13 +393,13 @@ class TransportPCEtesting(unittest.TestCase):
 
     def test_20_get_service_DSR(self):
         response = test_utils.get_service_list_request(
-            "services/" + str(service_dsr_uuid))
+            "services/" + str(SERVICE_DSR_UUID))
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertEqual(
             res['services'][0]['administrative-state'], 'inService')
         self.assertEqual(
-            res['services'][0]['service-name'], service_dsr_uuid)
+            res['services'][0]['service-name'], SERVICE_DSR_UUID)
         self.assertEqual(
             res['services'][0]['connection-type'], 'service')
         self.assertEqual(
@@ -403,17 +412,17 @@ class TransportPCEtesting(unittest.TestCase):
         res = response.json()
         liste_service = res['output']['service']
         for ele in liste_service:
-            if ele['uuid'] == service_pm_uuid:
+            if ele['uuid'] == SERVICE_PM_UUID:
                 self.assertEqual(ele['operational-state'], 'ENABLED')
                 self.assertEqual(ele['service-layer'], 'PHOTONIC_MEDIA')
                 nbconnection = len(ele['connection'])
                 self.assertEqual(nbconnection, 9, 'There should be 9 connections')
-            elif ele['uuid'] == service_odu_uuid:
+            elif ele['uuid'] == SERVICE_ODU_UUID:
                 self.assertEqual(ele['operational-state'], 'ENABLED')
                 self.assertEqual(ele['service-layer'], 'ODU')
                 nbconnection = len(ele['connection'])
                 self.assertEqual(nbconnection, 3, 'There should be 3 connections')
-            elif ele['uuid'] == service_dsr_uuid:
+            elif ele['uuid'] == SERVICE_DSR_UUID:
                 self.assertEqual(ele['operational-state'], 'ENABLED')
                 self.assertEqual(ele['service-layer'], 'DSR')
                 nbconnection = len(ele['connection'])
@@ -423,17 +432,17 @@ class TransportPCEtesting(unittest.TestCase):
         time.sleep(2)
 
     def test_22_delete_connectivity_service_DSR(self):
-        response = test_utils.tapi_delete_connectivity_request(service_dsr_uuid)
+        response = test_utils.tapi_delete_connectivity_request(SERVICE_DSR_UUID)
         self.assertEqual(response.status_code, requests.codes.no_content)
         time.sleep(self.WAITING)
 
     def test_23_delete_connectivity_service_ODU(self):
-        response = test_utils.tapi_delete_connectivity_request(service_odu_uuid)
+        response = test_utils.tapi_delete_connectivity_request(SERVICE_ODU_UUID)
         self.assertEqual(response.status_code, requests.codes.no_content)
         time.sleep(self.WAITING)
 
     def test_24_delete_connectivity_service_PhotonicMedia(self):
-        response = test_utils.tapi_delete_connectivity_request(service_pm_uuid)
+        response = test_utils.tapi_delete_connectivity_request(SERVICE_PM_UUID)
         self.assertEqual(response.status_code, requests.codes.no_content)
         time.sleep(self.WAITING)
 
index 2c4f46f43779cf3f4272a88d3131e9df811c5dfb..ab5a56225a5a395930cabdc546bc1d52c5dcb95a 100644 (file)
 
 # pylint: disable=no-member
 # pylint: disable=too-many-public-methods
+# pylint: disable=too-many-lines
 
 import base64
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -593,6 +597,7 @@ class TransportPCEtesting(unittest.TestCase):
 
 # test service-create for OCH-OTU4 service from spdrB to spdrC
     def test_35_create_OCH_OTU4_service_BC(self):
+        # pylint: disable=line-too-long
         self.cr_serv_sample_data["input"]["service-name"] = "service-OCH-OTU4-BC"
         self.cr_serv_sample_data["input"]["service-a-end"]["node-id"] = "SPDR-SB1"
         self.cr_serv_sample_data["input"]["service-a-end"]["clli"] = "NodeSB"
@@ -732,6 +737,7 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         liste_tp = res['node'][0]['ietf-network-topology:termination-point']
+        # pylint: disable=consider-using-f-string
         for ele in liste_tp:
             if ele['tp-id'] == 'XPDR2-NETWORK1':
                 self.assertEqual({'frequency': 196.1,
@@ -823,6 +829,7 @@ class TransportPCEtesting(unittest.TestCase):
 
 # test service-create for 100GE service from spdrA to spdrC via spdrB
     def test_46_create_100GE_service_ABC(self):
+        # pylint: disable=line-too-long
         self.cr_serv_sample_data["input"]["service-name"] = "service-100GE-ABC"
         self.cr_serv_sample_data["input"]["connection-type"] = "service"
         self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "Ethernet"
@@ -1344,6 +1351,7 @@ class TransportPCEtesting(unittest.TestCase):
         self.assertEqual(response.status_code, requests.codes.created)
 
     def test_83_create_OCH_OTU4_service_AC(self):
+        # pylint: disable=line-too-long
         self.cr_serv_sample_data["input"]["service-name"] = "service-OCH-OTU4-AC"
         self.cr_serv_sample_data["input"]["connection-type"] = "infrastructure"
         self.cr_serv_sample_data["input"]["service-a-end"]["service-rate"] = "100"
@@ -1384,6 +1392,7 @@ class TransportPCEtesting(unittest.TestCase):
 
 # test service-create for 100GE service from spdrA to spdrC via spdrB
     def test_85_create_100GE_service_AC(self):
+        # pylint: disable=line-too-long
         self.cr_serv_sample_data["input"]["service-name"] = "service-100GE-AC"
         self.cr_serv_sample_data["input"]["connection-type"] = "service"
         self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "Ethernet"
index 7178eaec922a38d9617da8eb44c6558fce4e3bc7..b5317f550fcdd73d4c968a2dc0e8b4959c2abe2c 100644 (file)
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils_rfc8040  # nopep8
 
 
@@ -37,6 +40,7 @@ class TransportPCE400GPortMappingTesting(unittest.TestCase):
         print("all processes killed")
 
     def setUp(self):
+        # pylint: disable=consider-using-f-string
         print("execution of {}".format(self.id().split(".")[-1]))
         time.sleep(10)
 
index d5118149b0816a022b83040eeea715af8aa07859..c18a4d1f5ee1288834881c70e5f9835ae3864ef9 100644 (file)
@@ -13,6 +13,7 @@
 
 import json
 import os
+# pylint: disable=wrong-import-order
 import sys
 import re
 import signal
@@ -22,6 +23,7 @@ import time
 import psutil
 import requests
 
+# pylint: disable=import-error
 import simulators
 
 SIMS = simulators.SIMS
@@ -131,7 +133,7 @@ def start_karaf():
     executable = os.path.join(
         os.path.dirname(os.path.realpath(__file__)),
         "..", "..", "..", KARAF_INSTALLDIR, "target", "assembly", "bin", "karaf")
-    with open('odl.log', 'w') as outfile:
+    with open('odl.log', 'w', encoding='utf-8') as outfile:
         return subprocess.Popen(
             ["sh", executable, "server"], stdout=outfile, stderr=outfile, stdin=None)
 
@@ -142,7 +144,7 @@ def start_lighty():
         os.path.dirname(os.path.realpath(__file__)),
         "..", "..", "..", "lighty", "target", "tpce",
         "clean-start-controller.sh")
-    with open(TPCE_LOG, 'w') as outfile:
+    with open(TPCE_LOG, 'w', encoding='utf-8') as outfile:
         return subprocess.Popen(
             ["sh", executable], stdout=outfile, stderr=outfile, stdin=None)
 
@@ -496,7 +498,7 @@ def start_honeynode(log_file: str, sim):
     sample_directory = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                     "..", "..", "sample_configs", "openroadm", sim[1])
     if os.path.isfile(executable):
-        with open(log_file, 'w') as outfile:
+        with open(log_file, 'w', encoding='utf-8') as outfile:
             return subprocess.Popen(
                 [executable, SIMS[sim]['port'], os.path.join(sample_directory, SIMS[sim]['configfile'])],
                 stdout=outfile, stderr=outfile)
@@ -512,7 +514,7 @@ def wait_until_log_contains(log_file, regexp, time_to_wait=60):
         with TimeOut(seconds=time_to_wait):
             while not os.path.exists(log_file):
                 time.sleep(0.2)
-            filelogs = open(log_file, 'r')
+            filelogs = open(log_file, 'r', encoding='utf-8')
             filelogs.seek(0, 2)
             filefound = True
             print("Searching for pattern '" + regexp + "' in " + os.path.basename(log_file), end='... ', flush=True)
index 5093ebcf5429f72c94a76faf58196805d452f630..e1323f2172f539e4816c409ad2dc15212d81c0c3 100644 (file)
@@ -13,6 +13,7 @@
 
 import json
 import os
+# pylint: disable=wrong-import-order
 import sys
 import re
 import signal
@@ -22,6 +23,7 @@ import time
 import psutil
 import requests
 
+# pylint: disable=import-error
 import simulators
 
 SIMS = simulators.SIMS
@@ -141,7 +143,7 @@ def start_karaf():
     executable = os.path.join(
         os.path.dirname(os.path.realpath(__file__)),
         "..", "..", "..", KARAF_INSTALLDIR, "target", "assembly", "bin", "karaf")
-    with open('odl.log', 'w') as outfile:
+    with open('odl.log', 'w', encoding='utf-8') as outfile:
         return subprocess.Popen(
             ["sh", executable, "server"], stdout=outfile, stderr=outfile, stdin=None)
 
@@ -152,7 +154,7 @@ def start_lighty():
         os.path.dirname(os.path.realpath(__file__)),
         "..", "..", "..", "lighty", "target", "tpce",
         "clean-start-controller.sh")
-    with open(TPCE_LOG, 'w') as outfile:
+    with open(TPCE_LOG, 'w', encoding='utf-8') as outfile:
         return subprocess.Popen(
             ["sh", executable], stdout=outfile, stderr=outfile, stdin=None)
 
@@ -182,7 +184,7 @@ def start_honeynode(log_file: str, sim):
     sample_directory = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                     "..", "..", "sample_configs", "openroadm", sim[1])
     if os.path.isfile(executable):
-        with open(log_file, 'w') as outfile:
+        with open(log_file, 'w', encoding='utf-8') as outfile:
             return subprocess.Popen(
                 [executable, SIMS[sim]['port'], os.path.join(sample_directory, SIMS[sim]['configfile'])],
                 stdout=outfile, stderr=outfile)
@@ -198,7 +200,7 @@ def wait_until_log_contains(log_file, regexp, time_to_wait=60):
         with TimeOut(seconds=time_to_wait):
             while not os.path.exists(log_file):
                 time.sleep(0.2)
-            filelogs = open(log_file, 'r')
+            filelogs = open(log_file, 'r', encoding='utf-8')
             filelogs.seek(0, 2)
             filefound = True
             print("Searching for pattern '" + regexp + "' in " + os.path.basename(log_file), end='... ', flush=True)
@@ -306,7 +308,7 @@ def get_portmapping_node_info(node: str):
     url = "{}/data/transportpce-portmapping:network/nodes={}/node-info"
     response = get_request(url.format('{}', node))
     res = response.json()
-    key = u'transportpce-portmapping:node-info'
+    key = 'transportpce-portmapping:node-info'
     if key in res.keys():
         node_info = res[key]
     else:
index d2e86838bf3d2896d5ecd6dd7ce4600ea6d455bd..9ab4ae2d5efedd27f441a8efed9ad2e78100575c 100644 (file)
@@ -14,8 +14,11 @@ import json
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -133,6 +136,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):
index db0a67c1e61e2b768ea02005b679683860fcbe79..988d4240802477832f749df31e76dadb2c2f42ce 100644 (file)
@@ -9,15 +9,20 @@
 # http://www.apache.org/licenses/LICENSE-2.0
 ##############################################################################
 
+# pylint: disable=invalid-name
 # pylint: disable=no-member
 # pylint: disable=too-many-public-methods
+# pylint: disable=too-many-lines
 
 import base64
 import unittest
 import time
 import requests
+# pylint: disable=wrong-import-order
 import sys
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -584,7 +589,7 @@ 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'] == 'XPDR-A2-XPDR2' or 'XPDR-C2-XPDR2':
+            if node['node-id'] == 'XPDR-A2-XPDR2' or node['node-id'] == 'XPDR-C2-XPDR2':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
                     if tp['tp-id'] == 'XPDR2-NETWORK1':
@@ -840,7 +845,7 @@ 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'] == 'XPDR-A2-XPDR2' or 'XPDR-C2-XPDR2':
+            if node['node-id'] == 'XPDR-A2-XPDR2' or node['node-id'] == 'XPDR-C2-XPDR2':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
                     if tp['tp-id'] == 'XPDR2-NETWORK1':
@@ -914,7 +919,7 @@ 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'] == 'XPDR-A2-XPDR2' or 'XPDR-C2-XPDR2':
+            if node['node-id'] == 'XPDR-A2-XPDR2' or node['node-id'] == 'XPDR-C2-XPDR2':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
                     if tp['tp-id'] == 'XPDR2-NETWORK1':
@@ -992,7 +997,7 @@ 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'] == 'XPDR-A2-XPDR2' or 'XPDR-C2-XPDR2'):
+            if node['node-id'] == 'XPDR-A2-XPDR2' or node['node-id'] == 'XPDR-C2-XPDR2':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
                     if tp['tp-id'] == 'XPDR2-NETWORK1':
@@ -1028,7 +1033,7 @@ 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'] == 'XPDR-A2-XPDR2' or 'XPDR-C2-XPDR2':
+            if node['node-id'] == 'XPDR-A2-XPDR2' or node['node-id'] == 'XPDR-C2-XPDR2':
                 tpList = node['ietf-network-topology:termination-point']
                 for tp in tpList:
                     if tp['tp-id'] == 'XPDR2-NETWORK1':
index 18f552460606c8cfb409f031425fedf1f0f7e6ed..e20a04a1b4a5af5dffe18b6c4b8fdc127400c768 100644 (file)
 
 import unittest
 import os
+# pylint: disable=wrong-import-order
 import sys
 import time
 import requests
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -35,22 +38,22 @@ class TransportPCEtesting(unittest.TestCase):
             sample_files_parsed = False
             TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                             "..", "..", "sample_configs", "honeynode-topo.xml")
-            with open(TOPO_BI_DIR_FILE, 'r') as topo_bi_dir:
+            with open(TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as topo_bi_dir:
                 cls.simple_topo_bi_dir_data = topo_bi_dir.read()
 
             TOPO_UNI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                              "..", "..", "sample_configs", "NW-simple-topology.xml")
 
-            with open(TOPO_UNI_DIR_FILE, 'r') as topo_uni_dir:
+            with open(TOPO_UNI_DIR_FILE, 'r', encoding='utf-8') as topo_uni_dir:
                 cls.simple_topo_uni_dir_data = topo_uni_dir.read()
 
             TOPO_UNI_DIR_COMPLEX_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                                      "..", "..", "sample_configs", "NW-for-test-5-4.xml")
-            with open(TOPO_UNI_DIR_COMPLEX_FILE, 'r') as topo_uni_dir_complex:
+            with open(TOPO_UNI_DIR_COMPLEX_FILE, 'r', encoding='utf-8') as topo_uni_dir_complex:
                 cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read()
             PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                              "..", "..", "sample_configs", "pce_portmapping_121.json")
-            with open(PORT_MAPPING_FILE, 'r') as port_mapping:
+            with open(PORT_MAPPING_FILE, 'r', encoding='utf-8') as port_mapping:
                 cls.port_mapping_data = port_mapping.read()
             sample_files_parsed = True
         except PermissionError as err:
index 7be2c48c684864c412e12f278d1fc7bbcf9dc948..afe3b9a157d59da771791b95f8d3ad387118f5e7 100644 (file)
@@ -8,15 +8,19 @@
 # http://www.apache.org/licenses/LICENSE-2.0
 ##############################################################################
 
+# pylint: disable=invalid-name
 # pylint: disable=no-member
 # pylint: disable=too-many-public-methods
 
 import unittest
 import os
+# pylint: disable=wrong-import-order
 import sys
 import time
 import requests
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -28,31 +32,37 @@ class TransportPCE400Gtesting(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
+        # pylint: disable=bare-except
         try:
             sample_files_parsed = False
             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:
+                                            "..", "..", "sample_configs",
+                                            "honeynode-topo400G.json")
+            with open(TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as topo_bi_dir:
                 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:
+                                                "..", "..", "sample_configs",
+                                                "honeynode-otntopo400G.json")
+            with open(OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') 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:
+                                                      "..", "..", "sample_configs",
+                                                      "honeynode-otntopo400GwithOTUC4.json")
+            with open(OTUC4_OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') 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:
+                                                      "..", "..", "sample_configs",
+                                                      "honeynode-otntopo400GwithODUC4.json")
+            with open(ODUC4_OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') 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")
-            with open(PORT_MAPPING_FILE, 'r') as port_mapping:
+                                             "..", "..", "sample_configs",
+                                             "pce_portmapping_71.json")
+            with open(PORT_MAPPING_FILE, 'r', encoding='utf-8') as port_mapping:
                 cls.port_mapping_data = port_mapping.read()
             sample_files_parsed = True
         except PermissionError as err:
@@ -77,6 +87,7 @@ class TransportPCE400Gtesting(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]))
         time.sleep(1)
 
@@ -136,13 +147,17 @@ class TransportPCE400Gtesting(unittest.TestCase):
     # 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": "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"}}
+                                                       {"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()
@@ -181,13 +196,17 @@ class TransportPCE400Gtesting(unittest.TestCase):
     # 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",
-                                                           "tx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2"}}
+                                                       {"service-rate": "400",
+                                                        "clli": "NodeA",
+                                                        "service-format": "ODU",
+                                                        "node-id": "XPDR-A2",
+                                                        "tx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2"}}
                                                         },
-                                                       {"service-rate": "400", "clli": "NodeC", "service-format": "ODU",
-                                                           "node-id": "XPDR-C2",
-                                                           "tx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2"}}
+                                                       {"service-rate": "400",
+                                                        "clli": "NodeC",
+                                                        "service-format": "ODU",
+                                                        "node-id": "XPDR-C2",
+                                                        "tx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2"}}
                                                         })
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
@@ -214,11 +233,15 @@ class TransportPCE400Gtesting(unittest.TestCase):
     # 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",
+                                                       {"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",
+                                                       {"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"}}})
index 0fae049f63d712ab585b9b3fffaf19211b955488..99e717f59870cbb7bafe93a8de8075104ee7e597 100644 (file)
 
 import unittest
 import os
+# pylint: disable=wrong-import-order
 import sys
 import time
 import requests
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -36,21 +39,21 @@ class TransportGNPYtesting(unittest.TestCase):
             sample_files_parsed = False
             TOPO_CLLINET_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                              "..", "..", "sample_configs", "gnpy", "clliNetwork.json")
-            with open(TOPO_CLLINET_FILE, 'r') as topo_cllinet:
+            with open(TOPO_CLLINET_FILE, 'r', encoding='utf-8') as topo_cllinet:
                 cls.topo_cllinet_data = topo_cllinet.read()
 
             TOPO_ORDNET_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                             "..", "..", "sample_configs", "gnpy", "openroadmNetwork.json")
-            with open(TOPO_ORDNET_FILE, 'r') as topo_ordnet:
+            with open(TOPO_ORDNET_FILE, 'r', encoding='utf-8') as topo_ordnet:
                 cls.topo_ordnet_data = topo_ordnet.read()
 
             TOPO_ORDTOPO_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                              "..", "..", "sample_configs", "gnpy", "openroadmTopology.json")
-            with open(TOPO_ORDTOPO_FILE, 'r') as topo_ordtopo:
+            with open(TOPO_ORDTOPO_FILE, 'r', encoding='utf-8') as topo_ordtopo:
                 cls.topo_ordtopo_data = topo_ordtopo.read()
             PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                              "..", "..", "sample_configs", "gnpy", "gnpy_portmapping_121.json")
-            with open(PORT_MAPPING_FILE, 'r') as port_mapping:
+            with open(PORT_MAPPING_FILE, 'r', encoding='utf-8') as port_mapping:
                 cls.port_mapping_data = port_mapping.read()
             sample_files_parsed = True
         except PermissionError as err:
index 2c01853bff135a325d2b47ffd93d108df3a0dac6..235f75ef742936ce2e5e8bab49a9dc23bbcb4c72 100644 (file)
 
 import os
 import json
+# pylint: disable=wrong-import-order
 import sys
 import unittest
 import time
 import requests
 sys.path.append('transportpce_tests/common/')
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
 import test_utils  # nopep8
 
 
@@ -117,6 +120,7 @@ class TransportNbiNotificationstesting(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
+        # pylint: disable=unsubscriptable-object
         # TODO: for lighty manage the activation of NBI notification feature
         cls.init_failed = False
         cls.processes = test_utils.start_tpce()
@@ -149,6 +153,7 @@ class TransportNbiNotificationstesting(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]))
 
     def test_01_connect_xpdrA(self):