Outsource sims DS direct op on CP to test_utils 36/107836/6
authorguillaume.lambert <guillaume.lambert@orange.com>
Tue, 12 Sep 2023 14:23:12 +0000 (16:23 +0200)
committerguillaume.lambert <guillaume.lambert@orange.com>
Wed, 13 Sep 2023 08:38:05 +0000 (10:38 +0200)
JIRA: TRNSPRTPCE-754
Signed-off-by: guillaume.lambert <guillaume.lambert@orange.com>
Change-Id: I3217e2a1481c2437e5bb33b71711329c7f0ee735

tests/transportpce_tests/common/simulators.py
tests/transportpce_tests/common/test_utils.py
tests/transportpce_tests/hybrid/test01_device_change_notifications.py
tests/transportpce_tests/tapi/test03_tapi_device_change_notifications.py
tests/transportpce_tests/with_docker/test02_nbinotifications.py
tests/transportpce_tests/with_docker/test03_tapi_nbinotifications.py

index cb7294af65d0847262f2875b348381fda5155481..5eef9402223464cb8bb694129a70ade302f566e2 100644 (file)
@@ -8,28 +8,29 @@
 # http://www.apache.org/licenses/LICENSE-2.0
 ##############################################################################
 
+# pylint: disable=line-too-long
 SIMS = {
-    ('xpdra', '1.2.1'): {'port': '17830', 'configfile': 'oper-XPDRA.xml', 'logfile': 'xpdra-121.log'},
-    ('roadma', '1.2.1'): {'port': '17831', 'configfile': 'oper-ROADMA.xml', 'logfile': 'roadma-121.log'},
-    ('roadmb', '1.2.1'): {'port': '17832', 'configfile': 'oper-ROADMB.xml', 'logfile': 'roadmb-121.log'},
-    ('roadmc', '1.2.1'): {'port': '17833', 'configfile': 'oper-ROADMC.xml', 'logfile': 'roadmc-121.log'},
-    ('xpdrc', '1.2.1'): {'port': '17834', 'configfile': 'oper-XPDRC.xml', 'logfile': 'xpdrc-121.log'},
-    ('roadma-full', '1.2.1'): {'port': '17821', 'configfile': 'oper-ROADMA-full.xml', 'logfile': 'roadma-121.log'},
-    ('roadmc-full', '1.2.1'): {'port': '17823', 'configfile': 'oper-ROADMC-full.xml', 'logfile': 'roadmc-121.log'},
-    ('xpdra', '2.2.1'): {'port': '17840', 'configfile': 'oper-XPDRA.xml', 'logfile': 'xpdra-221.log'},
-    ('roadma', '2.2.1'): {'port': '17841', 'configfile': 'oper-ROADMA.xml', 'logfile': 'roadma-221.log'},
-    ('roadmb', '2.2.1'): {'port': '17842', 'configfile': 'oper-ROADMB.xml', 'logfile': 'roadmb-221.log'},
-    ('roadmc', '2.2.1'): {'port': '17843', 'configfile': 'oper-ROADMC.xml', 'logfile': 'roadmc-221.log'},
-    ('roadmd', '2.2.1'): {'port': '17847', 'configfile': 'oper-ROADMD.xml', 'logfile': 'roadmd-221.log'},
-    ('xpdrc', '2.2.1'): {'port': '17844', 'configfile': 'oper-XPDRC.xml', 'logfile': 'xpdrc-221.log'},
-    ('spdra', '2.2.1'): {'port': '17845', 'configfile': 'oper-SPDRA.xml', 'logfile': 'spdra-221.log'},
-    ('spdrc', '2.2.1'): {'port': '17846', 'configfile': 'oper-SPDRC.xml', 'logfile': 'spdrc-221.log'},
-    ('spdrb', '2.2.1'): {'port': '17848', 'configfile': 'oper-SPDRB.xml', 'logfile': 'spdrb-221.log'},
-    ('xpdra', '7.1'): {'port': '17850', 'configfile': 'oper-XPDRA.xml', 'logfile': 'xpdra-71.log'},
-    ('roadma', '7.1'): {'port': '17851', 'configfile': 'oper-ROADMA.xml', 'logfile': 'roadma-71.log'},
-    ('roadmb', '7.1'): {'port': '17852', 'configfile': 'oper-ROADMB.xml', 'logfile': 'roadmb-71.log'},
-    ('roadmc', '7.1'): {'port': '17853', 'configfile': 'oper-ROADMC.xml', 'logfile': 'roadmc-71.log'},
-    ('xpdrc', '7.1'): {'port': '17854', 'configfile': 'oper-XPDRC.xml', 'logfile': 'xpdrc-71.log'},
-    ('xpdra2', '7.1'): {'port': '17857', 'configfile': 'oper-XPDRA2.xml', 'logfile': 'xpdra2-71.log'},
-    ('xpdrc2', '7.1'): {'port': '17858', 'configfile': 'oper-XPDRC2.xml', 'logfile': 'xpdrc2-71.log'}
+    ('xpdra', '1.2.1'): {'port': '17830', 'configfile': 'oper-XPDRA.xml', 'logfile': 'xpdra-121.log', 'restconf_baseurl': 'http://127.0.0.1:8130/restconf'},
+    ('roadma', '1.2.1'): {'port': '17831', 'configfile': 'oper-ROADMA.xml', 'logfile': 'roadma-121.log', 'restconf_baseurl': 'http://127.0.0.1:8131/restconf'},
+    ('roadmb', '1.2.1'): {'port': '17832', 'configfile': 'oper-ROADMB.xml', 'logfile': 'roadmb-121.log', 'restconf_baseurl': 'http://127.0.0.1:8132/restconf'},
+    ('roadmc', '1.2.1'): {'port': '17833', 'configfile': 'oper-ROADMC.xml', 'logfile': 'roadmc-121.log', 'restconf_baseurl': 'http://127.0.0.1:8133/restconf'},
+    ('xpdrc', '1.2.1'): {'port': '17834', 'configfile': 'oper-XPDRC.xml', 'logfile': 'xpdrc-121.log', 'restconf_baseurl': 'http://127.0.0.1:8134/restconf'},
+    ('roadma-full', '1.2.1'): {'port': '17821', 'configfile': 'oper-ROADMA-full.xml', 'logfile': 'roadma-121.log', 'restconf_baseurl': 'http://127.0.0.1:8121/restconf'},
+    ('roadmc-full', '1.2.1'): {'port': '17823', 'configfile': 'oper-ROADMC-full.xml', 'logfile': 'roadmc-121.log', 'restconf_baseurl': 'http://127.0.0.1:8123/restconf'},
+    ('xpdra', '2.2.1'): {'port': '17840', 'configfile': 'oper-XPDRA.xml', 'logfile': 'xpdra-221.log', 'restconf_baseurl': 'http://127.0.0.1:8140/restconf'},
+    ('roadma', '2.2.1'): {'port': '17841', 'configfile': 'oper-ROADMA.xml', 'logfile': 'roadma-221.log', 'restconf_baseurl': 'http://127.0.0.1:8141/restconf'},
+    ('roadmb', '2.2.1'): {'port': '17842', 'configfile': 'oper-ROADMB.xml', 'logfile': 'roadmb-221.log', 'restconf_baseurl': 'http://127.0.0.1:8142/restconf'},
+    ('roadmc', '2.2.1'): {'port': '17843', 'configfile': 'oper-ROADMC.xml', 'logfile': 'roadmc-221.log', 'restconf_baseurl': 'http://127.0.0.1:8143/restconf'},
+    ('roadmd', '2.2.1'): {'port': '17847', 'configfile': 'oper-ROADMD.xml', 'logfile': 'roadmd-221.log', 'restconf_baseurl': 'http://127.0.0.1:8147/restconf'},
+    ('xpdrc', '2.2.1'): {'port': '17844', 'configfile': 'oper-XPDRC.xml', 'logfile': 'xpdrc-221.log', 'restconf_baseurl': 'http://127.0.0.1:8144/restconf'},
+    ('spdra', '2.2.1'): {'port': '17845', 'configfile': 'oper-SPDRA.xml', 'logfile': 'spdra-221.log', 'restconf_baseurl': 'http://127.0.0.1:8145/restconf'},
+    ('spdrc', '2.2.1'): {'port': '17846', 'configfile': 'oper-SPDRC.xml', 'logfile': 'spdrc-221.log', 'restconf_baseurl': 'http://127.0.0.1:8146/restconf'},
+    ('spdrb', '2.2.1'): {'port': '17848', 'configfile': 'oper-SPDRB.xml', 'logfile': 'spdrb-221.log', 'restconf_baseurl': 'http://127.0.0.1:8148/restconf'},
+    ('xpdra', '7.1'): {'port': '17850', 'configfile': 'oper-XPDRA.xml', 'logfile': 'xpdra-71.log', 'restconf_baseurl': 'http://127.0.0.1:8150/restconf'},
+    ('roadma', '7.1'): {'port': '17851', 'configfile': 'oper-ROADMA.xml', 'logfile': 'roadma-71.log', 'restconf_baseurl': 'http://127.0.0.1:8151/restconf'},
+    ('roadmb', '7.1'): {'port': '17852', 'configfile': 'oper-ROADMB.xml', 'logfile': 'roadmb-71.log', 'restconf_baseurl': 'http://127.0.0.1:8152/restconf'},
+    ('roadmc', '7.1'): {'port': '17853', 'configfile': 'oper-ROADMC.xml', 'logfile': 'roadmc-71.log', 'restconf_baseurl': 'http://127.0.0.1:8153/restconf'},
+    ('xpdrc', '7.1'): {'port': '17854', 'configfile': 'oper-XPDRC.xml', 'logfile': 'xpdrc-71.log', 'restconf_baseurl': 'http://127.0.0.1:8154/restconf'},
+    ('xpdra2', '7.1'): {'port': '17857', 'configfile': 'oper-XPDRA2.xml', 'logfile': 'xpdra2-71.log', 'restconf_baseurl': 'http://127.0.0.1:8157/restconf'},
+    ('xpdrc2', '7.1'): {'port': '17858', 'configfile': 'oper-XPDRC2.xml', 'logfile': 'xpdrc2-71.log', 'restconf_baseurl': 'http://127.0.0.1:8158/restconf'}
 }
index 1272df7703bcd74aec551643beda73a0e8e0f567..cc0604abb6f6c15135eddcd4501b5b35de4b8e47 100644 (file)
@@ -676,3 +676,23 @@ def transportpce_api_rpc_request(api_module: str, rpc: str, payload: dict):
             return_output = res[return_key[RESTCONF_VERSION]]
     return {'status_code': response.status_code,
             'output': return_output}
+
+#
+# simulators datastore operations
+#
+
+
+def sims_update_cp_port(sim: tuple, circuitpack: str, port: str, payload: dict):
+    # pylint: disable=consider-using-f-string
+    url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/{}/ports/{}".format(
+        SIMS[sim]['restconf_baseurl'], circuitpack, port)
+    body = {"ports": [payload]}
+    print(sim)
+    print(url)
+    response = requests.request("PUT",
+                                url,
+                                data=json.dumps(body),
+                                headers=TYPE_APPLICATION_JSON,
+                                auth=(ODL_LOGIN, ODL_PWD),
+                                timeout=REQUEST_TIMEOUT)
+    return response.status_code == requests.codes.ok
index 88347bf3d8a0c5fae40ffb80929d44bcafe49959..c0d8c3e6474758e1ee147130b9d5651679ff1508 100644 (file)
@@ -10,7 +10,6 @@
 
 # pylint: disable=no-member
 # pylint: disable=too-many-public-methods
-import json
 import unittest
 import time
 import requests
@@ -198,19 +197,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(1)
 
     def test_13_change_status_line_port_xpdra(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('xpdra', self.NODE_VERSION_121), '1%2F0%2F1-PLUG-NET', '1',
+                                                       {
             "port-name": "1",
             "logical-connection-point": "XPDR1-NETWORK1",
             "port-type": "CFP2",
             "circuit-id": "XPDRA-NETWORK",
             "administrative-state": "outOfService",
-            "port-qual": "xpdr-network"}]}
-        response = requests.request("PUT", url.format("http://127.0.0.1:8130/restconf"),
-                                    data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-                                    auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "xpdr-network"
+        }))
         time.sleep(2)
 
     def test_14_check_update_portmapping(self):
@@ -272,19 +267,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(1)
 
     def test_17_restore_status_line_port_xpdra(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('xpdra', self.NODE_VERSION_121), '1%2F0%2F1-PLUG-NET', '1',
+                                                       {
             "port-name": "1",
             "logical-connection-point": "XPDR1-NETWORK1",
             "port-type": "CFP2",
             "circuit-id": "XPDRA-NETWORK",
             "administrative-state": "inService",
-            "port-qual": "xpdr-network"}]}
-        response = requests.request("PUT", url.format("http://127.0.0.1:8130/restconf"),
-                                    data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-                                    auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "xpdr-network"
+        }))
         time.sleep(2)
 
     def test_18_check_update_portmapping_ok(self):
@@ -320,19 +311,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         self.test_12_get_eth_service1()
 
     def test_21_change_status_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C1',
+                                                       {
             "port-name": "C1",
             "logical-connection-point": "SRG1-PP1",
             "port-type": "client",
             "circuit-id": "SRG1",
             "administrative-state": "outOfService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "roadm-external"
+        }))
         time.sleep(2)
 
     def test_22_check_update_portmapping(self):
@@ -387,19 +374,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(1)
 
     def test_24_restore_status_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C1',
+                                                       {
             "port-name": "C1",
             "logical-connection-point": "SRG1-PP1",
             "port-type": "client",
             "circuit-id": "SRG1",
             "administrative-state": "inService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "roadm-external"
+        }))
         time.sleep(2)
 
     def test_25_check_update_portmapping_ok(self):
@@ -412,19 +395,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         self.test_12_get_eth_service1()
 
     def test_28_change_status_line_port_roadma_deg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2%2F0', 'L1',
+                                                       {
             "port-name": "L1",
             "logical-connection-point": "DEG2-TTP-TXRX",
             "port-type": "LINE",
             "circuit-id": "1",
             "administrative-state": "outOfService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "roadm-external"
+        }))
         time.sleep(2)
 
     def test_29_check_update_portmapping(self):
@@ -479,19 +458,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(1)
 
     def test_31_restore_status_line_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2%2F0', 'L1',
+                                                       {
             "port-name": "L1",
             "logical-connection-point": "DEG2-TTP-TXRX",
             "port-type": "LINE",
             "circuit-id": "1",
             "administrative-state": "inService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "roadm-external"
+        }))
         time.sleep(2)
 
     def test_32_check_update_portmapping_ok(self):
@@ -504,17 +479,13 @@ class TransportPCEFulltesting(unittest.TestCase):
         self.test_12_get_eth_service1()
 
     def test_35_change_status_line_port_xpdrc(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_71), '1%2F0%2F1-PLUG-NET', '1',
+                                                       {
             "port-name": "1",
             "port-type": "CFP2",
             "administrative-state": "outOfService",
-            "port-qual": "xpdr-network"}]}
-        response = requests.request("PUT", url.format("http://127.0.0.1:8154/restconf"),
-                                    data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-                                    auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "xpdr-network"
+        }))
         time.sleep(2)
 
     def test_36_check_update_portmapping(self):
@@ -569,17 +540,13 @@ class TransportPCEFulltesting(unittest.TestCase):
         time.sleep(1)
 
     def test_38_restore_status_line_port_xpdrc(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('xpdrc', self.NODE_VERSION_71), '1%2F0%2F1-PLUG-NET', '1',
+                                                       {
             "port-name": "1",
             "port-type": "CFP2",
             "administrative-state": "inService",
-            "port-qual": "xpdr-network"}]}
-        response = requests.request("PUT", url.format("http://127.0.0.1:8154/restconf"),
-                                    data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-                                    auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "xpdr-network"
+        }))
         time.sleep(2)
 
     def test_39_check_update_portmapping_ok(self):
@@ -592,19 +559,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         self.test_12_get_eth_service1()
 
     def test_42_change_status_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C2"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C2',
+                                                       {
             "port-name": "C2",
             "logical-connection-point": "SRG1-PP2",
             "port-type": "client",
             "circuit-id": "SRG1",
             "administrative-state": "outOfService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "roadm-external"
+        }))
         time.sleep(2)
 
     def test_43_check_update_portmapping(self):
index 5da1194a5e5c9c015183c884d781b98243ff981f..cc02a6ae8d6f0bacffe357dd0db739cb02d8a308 100644 (file)
@@ -503,19 +503,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         self.test_13_get_connectivity_service_Ethernet()
 
     def test_28_change_status_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C1',
+                                                       {
             "port-name": "C1",
             "logical-connection-point": "SRG1-PP1",
             "port-type": "client",
             "circuit-id": "SRG1",
             "administrative-state": "outOfService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "roadm-external"
+        }))
         time.sleep(2)
 
     def test_29_check_update_portmapping(self):
@@ -618,19 +614,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         self.test_20_check_update_connectivity_service_Ethernet()
 
     def test_35_restore_status_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C1',
+                                                       {
             "port-name": "C1",
             "logical-connection-point": "SRG1-PP1",
             "port-type": "client",
             "circuit-id": "SRG1",
             "administrative-state": "inService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "roadm-external"
+        }))
         time.sleep(2)
 
     def test_36_check_update_portmapping_ok(self):
@@ -672,19 +664,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         self.test_13_get_connectivity_service_Ethernet()
 
     def test_42_change_status_line_port_roadma_deg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2%2F0', 'L1',
+                                                       {
             "port-name": "L1",
             "logical-connection-point": "DEG2-TTP-TXRX",
             "port-type": "LINE",
             "circuit-id": "1",
             "administrative-state": "outOfService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "roadm-external"
+        }))
         time.sleep(2)
 
     def test_43_check_update_portmapping(self):
@@ -788,19 +776,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         self.test_20_check_update_connectivity_service_Ethernet()
 
     def test_49_restore_status_line_port_roadma_deg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '2%2F0', 'L1',
+                                                       {
             "port-name": "L1",
             "logical-connection-point": "DEG2-TTP-TXRX",
             "port-type": "LINE",
             "circuit-id": "1",
             "administrative-state": "inService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "roadm-external"
+        }))
         time.sleep(2)
 
     def test_50_check_update_portmapping_ok(self):
@@ -822,19 +806,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         self.test_13_get_connectivity_service_Ethernet()
 
     def test_56_change_status_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C2"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C2',
+                                                       {
             "port-name": "C2",
             "logical-connection-point": "SRG1-PP2",
             "port-type": "client",
             "circuit-id": "SRG1",
             "administrative-state": "outOfService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "roadm-external"
+        }))
         time.sleep(2)
 
     def test_57_check_update_portmapping(self):
@@ -964,19 +944,15 @@ class TransportPCEFulltesting(unittest.TestCase):
         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 
     def test_69_restore_status_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C2"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C2',
+                                                       {
             "port-name": "C2",
             "logical-connection-point": "SRG1-PP2",
             "port-type": "client",
             "circuit-id": "SRG1",
             "administrative-state": "inService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
+            "port-qual": "roadm-external"
+        }))
         time.sleep(2)
 
     def test_70_clean_openroadm_topology(self):
index c3ce3d6c195cadcc7bc0333b32c4c8563a08d6ec..2cbf4e2a5efb7743d968b54c0cb30cbb35b957b4 100644 (file)
@@ -12,7 +12,6 @@
 # pylint: disable=too-many-public-methods
 
 import os
-import json
 # pylint: disable=wrong-import-order
 import sys
 import unittest
@@ -262,20 +261,15 @@ class TransportNbiNotificationstesting(unittest.TestCase):
                          'The service is now inService')
 
     def test_18_change_status_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION), '3%2F0', 'C1',
+                                                       {
             "port-name": "C1",
             "logical-connection-point": "SRG1-PP1",
             "port-type": "client",
             "circuit-id": "SRG1",
             "administrative-state": "outOfService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
-        time.sleep(2)
+            "port-qual": "roadm-external"
+        }))
 
     def test_19_get_notifications_alarm_service1(self):
         response = test_utils.transportpce_api_rpc_request(
@@ -288,20 +282,15 @@ class TransportNbiNotificationstesting(unittest.TestCase):
                          'The service is now outOfService')
 
     def test_20_restore_status_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION), '3%2F0', 'C1',
+                                                       {
             "port-name": "C1",
             "logical-connection-point": "SRG1-PP1",
             "port-type": "client",
             "circuit-id": "SRG1",
             "administrative-state": "inService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
-        time.sleep(2)
+            "port-qual": "roadm-external"
+        }))
 
     def test_21_get_notifications_alarm_service1(self):
         self.test_17_get_notifications_alarm_service1()
index 1a82ceaba55bd0a84fda7fb83498b0d009c9afc4..a58ef7a0553e077f5a3ced54751362f593030c81 100644 (file)
@@ -12,7 +12,6 @@
 # pylint: disable=too-many-public-methods
 
 import os
-import json
 # pylint: disable=wrong-import-order
 import sys
 import unittest
@@ -331,21 +330,15 @@ class TransportNbiNotificationstesting(unittest.TestCase):
         time.sleep(2)
 
     def test_15_change_status_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C1',
+                                                       {
             "port-name": "C1",
             "logical-connection-point": "SRG1-PP1",
             "port-type": "client",
             "circuit-id": "SRG1",
             "administrative-state": "outOfService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
-        # If the gate fails is because of the waiting time not being enough
-        time.sleep(2)
+            "port-qual": "roadm-external"
+        }))
 
     def test_16_get_tapi_notifications_connectivity_service_Ethernet(self):
         self.cr_get_notif_list_input_data["subscription-id-or-name"] = str(self.uuid_subscriptions.eth)
@@ -359,21 +352,15 @@ class TransportNbiNotificationstesting(unittest.TestCase):
         time.sleep(2)
 
     def test_17_restore_status_port_roadma_srg(self):
-        url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
-        body = {"ports": [{
+        self.assertTrue(test_utils.sims_update_cp_port(('roadma', self.NODE_VERSION_221), '3%2F0', 'C1',
+                                                       {
             "port-name": "C1",
             "logical-connection-point": "SRG1-PP1",
             "port-type": "client",
             "circuit-id": "SRG1",
             "administrative-state": "inService",
-            "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),
-                                    timeout=test_utils.REQUEST_TIMEOUT)
-        self.assertEqual(response.status_code, requests.codes.ok)
-        # If the gate fails is because of the waiting time not being enough
-        time.sleep(2)
+            "port-qual": "roadm-external"
+        }))
 
     def test_18_get_tapi_notifications_connectivity_service_Ethernet(self):
         self.cr_get_notif_list_input_data["subscription-id-or-name"] = str(self.uuid_subscriptions.eth)