X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F2.2.1%2Ftest_tapi.py;h=9467034383e8a8feb51d7db1c7c506b72975b072;hb=19a9408ee823754859467888a5d2823bfd8655ab;hp=e1c67477bd51fd382b0416c8bafece534f2feca0;hpb=c8bd98e0844257ae2e40e01d54546a40596df26d;p=transportpce.git

diff --git a/tests/transportpce_tests/2.2.1/test_tapi.py b/tests/transportpce_tests/2.2.1/test_tapi.py
index e1c67477b..946703438 100644
--- a/tests/transportpce_tests/2.2.1/test_tapi.py
+++ b/tests/transportpce_tests/2.2.1/test_tapi.py
@@ -8,7 +8,7 @@
 # http://www.apache.org/licenses/LICENSE-2.0
 ##############################################################################
 import os
-import re
+import sys
 import time
 import unittest
 
@@ -35,16 +35,17 @@ 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):
@@ -58,7 +59,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 +85,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 +94,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 +103,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 +112,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 +121,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 +130,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 +139,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 +148,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 +157,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 +172,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')