Code Review / transportpce.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | review | tree
raw | inline | side by side
fix some pylint issues
[transportpce.git] / tests / transportpce_tests / 2.2.1 / test_tapi.py
diff --git a/tests/transportpce_tests/2.2.1/test_tapi.py b/tests/transportpce_tests/2.2.1/test_tapi.py
index e1c67477bd51fd382b0416c8bafece534f2feca0..496a33e65d88d38f5eccd6a35b460f9e424cc340 100644 (file)
--- a/tests/transportpce_tests/2.2.1/test_tapi.py
+++ b/tests/transportpce_tests/2.2.1/test_tapi.py
@@ -7,8 +7,15 @@
 # which accompanies this distribution, and is available at
 # http://www.apache.org/licenses/LICENSE-2.0
 ##############################################################################
+
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+# some pylint false positives specific to tapi test
+# pylint: disable=unsubscriptable-object
+# pylint: disable=unsupported-assignment-operation
+
 import os
-import re
+import sys
 import time
 import unittest
 
@@ -27,7 +34,8 @@ class TransportTapitesting(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.init_failed = False
-
+        os.environ['JAVA_MIN_MEM'] = '1024M'
+        os.environ['JAVA_MAX_MEM'] = '4069M'
         cls.processes = test_utils.start_tpce()
         # TAPI feature is not installed by default in Karaf
         if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
@@ -35,19 +43,21 @@ class TransportTapitesting(unittest.TestCase):
             result = test_utils.install_karaf_feature("odl-transportpce-tapi")
             if result.returncode != 0:
                 cls.init_failed = True
-            print("Restarting opendaylight...")
+            print("Restarting OpenDaylight...")
             test_utils.shutdown_process(cls.processes[0])
-            cls.processes = test_utils.start_tpce()
+            cls.processes[0] = test_utils.start_karaf()
+            test_utils.process_list[0] = cls.processes[0]
             cls.init_failed = not test_utils.wait_until_log_contains(
-                test_utils.karaf_log, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
+                test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
         if cls.init_failed:
-            print("tapi installaiton feature failed...")
-            test_utils.shutdown_process(cls.odl_processes[0])
-            exit(2)
-        cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc', 'spdrav2'])
+            print("tapi installation feature failed...")
+            test_utils.shutdown_process(cls.processes[0])
+            sys.exit(2)
+        cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc', 'spdra'])
 
     @classmethod
     def tearDownClass(cls):
+        # pylint: disable=not-an-iterable
         for process in cls.processes:
             test_utils.shutdown_process(process)
         print("all processes killed")
@@ -58,7 +68,7 @@ class TransportTapitesting(unittest.TestCase):
         print("execution of {}".format(self.id().split(".")[-1]))
 
     def test_00_connect_spdr_sa1(self):
-        response = test_utils.mount_device("SPDR-SA1", 'spdrav2')
+        response = test_utils.mount_device("SPDR-SA1", 'spdra')
         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
         time.sleep(10)
         # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
@@ -84,9 +94,8 @@ class TransportTapitesting(unittest.TestCase):
         time.sleep(20)
 
     def test_05_connect_xprda_n1_to_roadma_pp1(self):
-        url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
-        data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
-        response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+        response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
+                                                          "ROADM-A1", "1", "SRG1-PP1-TXRX")
         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
         res = response.json()
         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
@@ -94,9 +103,8 @@ class TransportTapitesting(unittest.TestCase):
         time.sleep(2)
 
     def test_06_connect_roadma_pp1_to_xpdra_n1(self):
-        url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
-        data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
-        response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+        response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
+                                                          "ROADM-A1", "1", "SRG1-PP1-TXRX")
         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
         res = response.json()
         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
@@ -104,9 +112,8 @@ class TransportTapitesting(unittest.TestCase):
         time.sleep(2)
 
     def test_07_connect_xprdc_n1_to_roadmc_pp1(self):
-        url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
-        data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
-        response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+        response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
+                                                          "ROADM-C1", "1", "SRG1-PP1-TXRX")
         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
         res = response.json()
         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
@@ -114,9 +121,8 @@ class TransportTapitesting(unittest.TestCase):
         time.sleep(2)
 
     def test_08_connect_roadmc_pp1_to_xpdrc_n1(self):
-        url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
-        data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
-        response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+        response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
+                                                          "ROADM-C1", "1", "SRG1-PP1-TXRX")
         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
         res = response.json()
         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
@@ -124,9 +130,8 @@ class TransportTapitesting(unittest.TestCase):
         time.sleep(2)
 
     def test_09_connect_xprda_n2_to_roadma_pp2(self):
-        url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
-        data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
-        response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+        response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2",
+                                                          "ROADM-A1", "1", "SRG1-PP2-TXRX")
         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
         res = response.json()
         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
@@ -134,9 +139,8 @@ class TransportTapitesting(unittest.TestCase):
         time.sleep(2)
 
     def test_10_connect_roadma_pp2_to_xpdra_n2(self):
-        url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
-        data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
-        response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+        response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "2",
+                                                          "ROADM-A1", "1", "SRG1-PP2-TXRX")
         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
         res = response.json()
         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
@@ -144,9 +148,8 @@ class TransportTapitesting(unittest.TestCase):
         time.sleep(2)
 
     def test_11_connect_xprdc_n2_to_roadmc_pp2(self):
-        url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
-        data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
-        response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+        response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "2",
+                                                          "ROADM-C1", "1", "SRG1-PP2-TXRX")
         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
         res = response.json()
         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
@@ -154,9 +157,8 @@ class TransportTapitesting(unittest.TestCase):
         time.sleep(2)
 
     def test_12_connect_roadmc_pp2_to_xpdrc_n2(self):
-        url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
-        data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
-        response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+        response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "2",
+                                                          "ROADM-C1", "1", "SRG1-PP2-TXRX")
         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
         res = response.json()
         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
@@ -164,14 +166,14 @@ class TransportTapitesting(unittest.TestCase):
         time.sleep(2)
 
     def test_13_get_tapi_openroadm_topology(self):
-        url = "{}/operations/tapi-topology:get-topology-details".format(test_utils.RESTCONF_BASE_URL)
+        url = "{}/operations/tapi-topology:get-topology-details"
         data = {
             "tapi-topology:input": {
                 "tapi-topology:topology-id-or-name": "openroadm-topology"
             }
         }
 
-        response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
         res = response.json()
         self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'There should be 1 node')
@@ -179,14 +181,14 @@ class TransportTapitesting(unittest.TestCase):
                          'There should be 4 owned-node-edge-points')
 
     def test_14_get_tapi_otn_topology(self):
-        url = "{}/operations/tapi-topology:get-topology-details".format(test_utils.RESTCONF_BASE_URL)
+        url = "{}/operations/tapi-topology:get-topology-details"
         data = {
             "tapi-topology:input": {
                 "tapi-topology:topology-id-or-name": "otn-topology"
             }
         }
 
-        response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
         res = response.json()
         self.assertEqual(len(res["output"]["topology"]["node"]), 4, 'There should be 4 nodes')
Repository for Transport PCE project.