X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2Fcommon%2Ftest_utils.py;h=90f237172783a0cc599c738c0ba30bc3d540f27a;hb=b1b3bafd549bb501937cea5c976d5344608b6ed3;hp=1c0bf202cfedc8e2efce9e0f146637fabb5254ab;hpb=c16c716c06381a8bc52859550b02f9e084fea79a;p=transportpce.git
diff --git a/tests/transportpce_tests/common/test_utils.py b/tests/transportpce_tests/common/test_utils.py
index 1c0bf202c..90f237172 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
@@ -327,13 +328,19 @@ def mount_device(node: str, sim: str):
url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}',
'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}'}
body = {'node': [{
- 'node-id': node,
- 'netconf-node-topology:username': NODES_LOGIN,
- 'netconf-node-topology:password': NODES_PWD,
- 'netconf-node-topology:host': '127.0.0.1',
- 'netconf-node-topology:port': SIMS[sim]['port'],
- 'netconf-node-topology:tcp-only': 'false',
- 'netconf-node-topology:pass-through': {}}]}
+ "node-id": node,
+ "netconf-node-topology:host": "127.0.0.1",
+ "netconf-node-topology:port": SIMS[sim]['port'],
+ "netconf-node-topology:login-password-unencrypted": {
+ "netconf-node-topology:username": NODES_LOGIN,
+ "netconf-node-topology:password": NODES_PWD
+ },
+ "netconf-node-topology:tcp-only": "false",
+ "netconf-node-topology:reconnect-on-changed-schema": "false",
+ "netconf-node-topology:connection-timeout-millis": "20000",
+ "netconf-node-topology:default-request-timeout-millis": "60000",
+ "netconf-node-topology:max-connection-attempts": "0",
+ "netconf-node-topology:keepalive-delay": "120"}]}
response = put_request(url[RESTCONF_VERSION].format('{}', node), body)
if wait_until_log_contains(TPCE_LOG, 'Triggering notification stream NETCONF for node ' + node, 180):
print('Node ' + node + ' correctly added to tpce topology', end='... ', flush=True)
@@ -731,3 +738,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 = ''
+ xml_body += dict2xml(payload, indent=" ")
+ xml_body += ''
+ new_xml = xml_body.replace("/org-openroadm-device:org-openroadm-device/org-openroadm-device:interface[org-openroadm-device:name='OTS-DEG2-TTP-TXRX']",
+ "/a:org-openroadm-device/a:interface[a:name='OTS-DEG2-TTP-TXRX']")
+ 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