Code Review / transportpce.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | review | tree
raw | inline | side by side
Fix new pylint issues in functional tests
[transportpce.git] / tests / transportpce_tests / with_docker / test02_nbinotifications.py
diff --git a/tests/transportpce_tests/with_docker/test02_nbinotifications.py b/tests/transportpce_tests/with_docker/test02_nbinotifications.py
index 27ff94b5947f6095c375b830052a7ec9694c4f37..5e5edfae16bb6324feb5c3a60d98e7cced884777 100644 (file)
--- a/tests/transportpce_tests/with_docker/test02_nbinotifications.py
+++ b/tests/transportpce_tests/with_docker/test02_nbinotifications.py
@@ -13,16 +13,19 @@
 
 import os
 import json
+# pylint: disable=wrong-import-order
 import sys
 import unittest
 import time
 import requests
 sys.path.append('transportpce_tests/common/')
-import test_utils
+# pylint: disable=wrong-import-position
+# pylint: disable=import-error
+import test_utils  # nopep8
 
 
 class TransportNbiNotificationstesting(unittest.TestCase):
-    processes = None
+    processes = []
     cr_serv_sample_data = {"input": {
         "sdnc-request-header": {
             "request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
@@ -38,36 +41,8 @@ class TransportNbiNotificationstesting(unittest.TestCase):
             "node-id": "XPDR-A1",
             "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": [{"index": 0}],
+            "rx-direction": [{"index": 0}],
             "optic-type": "gray"
         },
         "service-z-end": {
@@ -75,36 +50,8 @@ class TransportNbiNotificationstesting(unittest.TestCase):
             "node-id": "XPDR-C1",
             "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": [{"index": 0}],
+            "rx-direction": [{"index": 0}],
             "optic-type": "gray"
         },
         "due-date": "2016-11-28T00:00:01Z",
@@ -117,6 +64,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 +97,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):
@@ -207,7 +156,7 @@ class TransportNbiNotificationstesting(unittest.TestCase):
                 "group-id": "transportpceTest"
             }
         }
-        response = test_utils.get_notifications_service_request(data)
+        response = test_utils.get_notifications_process_service_request(data)
         self.assertEqual(response.status_code, requests.codes.no_content)
         time.sleep(2)
 
@@ -228,13 +177,14 @@ class TransportNbiNotificationstesting(unittest.TestCase):
                 "group-id": "transportpceTest"
             }
         }
-        response = test_utils.get_notifications_service_request(data)
+        response = test_utils.get_notifications_process_service_request(data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
-        self.assertEqual(res['output']['notification-service'][-1]['service-name'], 'service1')
-        self.assertEqual(res['output']['notification-service'][-1]['connection-type'], 'service')
-        self.assertEqual(res['output']['notification-service'][-1]['message'], 'ServiceCreate request failed ...')
-        self.assertEqual(res['output']['notification-service'][-1]['response-failed'],
+        self.assertEqual(res['output']['notifications-process-service'][-1]['service-name'], 'service1')
+        self.assertEqual(res['output']['notifications-process-service'][-1]['connection-type'], 'service')
+        self.assertEqual(res['output']['notifications-process-service'][-1]['message'],
+                         'ServiceCreate request failed ...')
+        self.assertEqual(res['output']['notifications-process-service'][-1]['response-failed'],
                          'PCE path computation failed !')
         time.sleep(2)
 
@@ -300,12 +250,12 @@ class TransportNbiNotificationstesting(unittest.TestCase):
                 "group-id": "transportpceTest"
             }
         }
-        response = test_utils.get_notifications_service_request(data)
+        response = test_utils.get_notifications_process_service_request(data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
-        self.assertEqual(res['output']['notification-service'][-1]['service-name'], 'service1')
-        self.assertEqual(res['output']['notification-service'][-1]['connection-type'], 'service')
-        self.assertEqual(res['output']['notification-service'][-1]['message'], 'Service implemented !')
+        self.assertEqual(res['output']['notifications-process-service'][-1]['service-name'], 'service1')
+        self.assertEqual(res['output']['notifications-process-service'][-1]['connection-type'], 'service')
+        self.assertEqual(res['output']['notifications-process-service'][-1]['message'], 'Service implemented !')
         time.sleep(2)
 
     def test_17_get_notifications_alarm_service1(self):
@@ -319,10 +269,10 @@ class TransportNbiNotificationstesting(unittest.TestCase):
         response = test_utils.get_notifications_alarm_service_request(data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
-        self.assertEqual(res['output']['notification-alarm-service'][-1]['service-name'], 'service1')
-        self.assertEqual(res['output']['notification-alarm-service'][-1]['connection-type'], 'service')
-        self.assertEqual(res['output']['notification-alarm-service'][-1]['operational-state'], 'inService')
-        self.assertEqual(res['output']['notification-alarm-service'][-1]['message'], 'The service is now inService')
+        self.assertEqual(res['output']['notifications-alarm-service'][-1]['service-name'], 'service1')
+        self.assertEqual(res['output']['notifications-alarm-service'][-1]['connection-type'], 'service')
+        self.assertEqual(res['output']['notifications-alarm-service'][-1]['operational-state'], 'inService')
+        self.assertEqual(res['output']['notifications-alarm-service'][-1]['message'], 'The service is now inService')
         time.sleep(2)
 
     def test_18_change_status_port_roadma_srg(self):
@@ -336,7 +286,8 @@ class TransportNbiNotificationstesting(unittest.TestCase):
             "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))
+                                    auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
+                                    timeout=test_utils.REQUEST_TIMEOUT)
         self.assertEqual(response.status_code, requests.codes.ok)
         time.sleep(2)
 
@@ -351,10 +302,10 @@ class TransportNbiNotificationstesting(unittest.TestCase):
         response = test_utils.get_notifications_alarm_service_request(data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
-        self.assertEqual(res['output']['notification-alarm-service'][-1]['service-name'], 'service1')
-        self.assertEqual(res['output']['notification-alarm-service'][-1]['connection-type'], 'service')
-        self.assertEqual(res['output']['notification-alarm-service'][-1]['operational-state'], 'outOfService')
-        self.assertEqual(res['output']['notification-alarm-service'][-1]['message'], 'The service is now outOfService')
+        self.assertEqual(res['output']['notifications-alarm-service'][-1]['service-name'], 'service1')
+        self.assertEqual(res['output']['notifications-alarm-service'][-1]['connection-type'], 'service')
+        self.assertEqual(res['output']['notifications-alarm-service'][-1]['operational-state'], 'outOfService')
+        self.assertEqual(res['output']['notifications-alarm-service'][-1]['message'], 'The service is now outOfService')
         time.sleep(2)
 
     def test_20_restore_status_port_roadma_srg(self):
@@ -368,7 +319,8 @@ class TransportNbiNotificationstesting(unittest.TestCase):
             "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))
+                                    auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
+                                    timeout=test_utils.REQUEST_TIMEOUT)
         self.assertEqual(response.status_code, requests.codes.ok)
         time.sleep(2)
 
@@ -391,12 +343,12 @@ class TransportNbiNotificationstesting(unittest.TestCase):
                 "group-id": "transportpceTest"
             }
         }
-        response = test_utils.get_notifications_service_request(data)
+        response = test_utils.get_notifications_process_service_request(data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
-        self.assertEqual(res['output']['notification-service'][-1]['service-name'], 'service1')
-        self.assertEqual(res['output']['notification-service'][-1]['connection-type'], 'service')
-        self.assertEqual(res['output']['notification-service'][-1]['message'], 'Service deleted !')
+        self.assertEqual(res['output']['notifications-process-service'][-1]['service-name'], 'service1')
+        self.assertEqual(res['output']['notifications-process-service'][-1]['connection-type'], 'service')
+        self.assertEqual(res['output']['notifications-process-service'][-1]['message'], 'Service deleted !')
         time.sleep(2)
 
     def test_24_disconnect_XPDRA(self):
Repository for Transport PCE project.