X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2Fcommon%2Ftest_utils.py;h=5d2789c1b024bd901f9a61c68ccc773f725e72cf;hb=122239adbcc8e857aa9eff4a45f6ab1900dfe885;hp=888a8e91fc228751b85ecd386c41bd49cd2f4537;hpb=65e9c4506ec5f5b3449fac0a258020ebb1085337;p=transportpce.git

diff --git a/tests/transportpce_tests/common/test_utils.py b/tests/transportpce_tests/common/test_utils.py
index 888a8e91f..5d2789c1b 100644
--- a/tests/transportpce_tests/common/test_utils.py
+++ b/tests/transportpce_tests/common/test_utils.py
@@ -28,6 +28,7 @@ from dict2xml import dict2xml
 from netconf_client.connect import connect_ssh
 from netconf_client.ncclient import Manager
 
+
 # pylint: disable=import-error
 import simulators
 
@@ -53,6 +54,9 @@ CODE_SHOULD_BE_201 = 'Http status code should be 201'
 T100GE = 'Transponder 100GE'
 T0_MULTILAYER_TOPO = 'T0 - Multi-layer topology'
 T0_FULL_MULTILAYER_TOPO = 'T0 - Full Multi-layer topology'
+T100GE_UUID = 'cf51c729-3699-308a-a7d0-594c6a62ebbb'
+T0_MULTILAYER_TOPO_UUID = '747c670e-7a07-3dab-b379-5b1cd17402a3'
+T0_FULL_MULTILAYER_TOPO_UUID = '393f09a4-0a0b-3d82-a4f6-1fbbc14ca1a7'
 
 SIM_LOG_DIRECTORY = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'log')
 
@@ -728,3 +732,42 @@ def sims_update_cp_port_ntcf(sim: tuple, circuitpack: str, payload: dict):
     if "None" in str(reply):
         return True
     return False
+
+
+def sims_update_pm_interact(sim: tuple, payload: dict):
+    if SIMS_TYPE == 'lightynode':
+        return sims_update_pm_interact_ntcf(sim, payload)
+    if SIMS_TYPE == 'honeynode':
+        return sims_update_pm_interact_rest(sim, payload)
+    return False
+
+
+def sims_update_pm_interact_rest(sim: tuple, payload: dict):
+    # pylint: disable=consider-using-f-string
+    url = "{}/operations/pm-handling:pm-interact".format(SIMS[sim]['restconf_baseurl'])
+    body = {"input": payload}
+    response = requests.request("POST",
+                                url,
+                                data=json.dumps(body),
+                                headers=TYPE_APPLICATION_JSON,
+                                auth=(ODL_LOGIN, ODL_PWD),
+                                timeout=REQUEST_TIMEOUT)
+    return response.status_code == requests.codes.ok
+
+
+def sims_update_pm_interact_ntcf(sim: tuple, payload: dict):
+    # pylint: disable=line-too-long
+    xml_body = '<pm-interact xmlns="http://honeynode-simulator/pm-handling">'
+    xml_body += dict2xml(payload, indent="  ")
+    xml_body += '</pm-interact>'
+    new_xml = xml_body.replace("<pm-resource-instance>/org-openroadm-device:org-openroadm-device/org-openroadm-device:interface[org-openroadm-device:name='OTS-DEG2-TTP-TXRX']</pm-resource-instance>",
+                               "<pm-resource-instance xmlns:a=\"http://org/openroadm/device\">/a:org-openroadm-device/a:interface[a:name='OTS-DEG2-TTP-TXRX']</pm-resource-instance>")
+    with connect_ssh(host='127.0.0.1',
+                     port=int(SIMS[sim]['port']),
+                     username=NODES_LOGIN,
+                     password=NODES_PWD) as session:
+        mgr = Manager(session, timeout=120)
+        reply = mgr.dispatch(new_xml)
+    if "netconf_client.ncclient.RPCReply" in str(reply):
+        return True
+    return False