Power control support for 87.5GHz spectrum width

[transportpce.git] / tests / transportpce_tests / common / test_utils.py

diff --git a/tests/transportpce_tests/common/test_utils.py b/tests/transportpce_tests/common/test_utils.py
index c18a4d1f5ee1288834881c70e5f9835ae3864ef9..b6eb71cae984a48e667c214325739246cd8ee19e 100644 (file)
--- a/tests/transportpce_tests/common/test_utils.py
+++ b/tests/transportpce_tests/common/test_utils.py
@@ -10,6 +10,7 @@
 ##############################################################################
 
 # pylint: disable=no-member
+# pylint: disable=too-many-arguments
 
 import json
 import os
@@ -260,6 +261,27 @@ def mount_device(node_id, sim):
     return response
 
 
+def mount_tapi_device(node_id, sim):
+    url = URL_CONFIG_NETCONF_TOPO + "node/" + node_id
+    body = {"node": [{
+        "node-id": node_id,
+        "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": {}}]}
+    response = put_request(url, body)
+    if wait_until_log_contains(TPCE_LOG, re.escape(f"TAPI node for or node {node_id} successfully merged"), 200):
+        print("Node " + node_id + " correctly added to tpce topology", end='... ', flush=True)
+    else:
+        print("Node " + node_id + " still not added to tpce topology", end='... ', flush=True)
+        if response.status_code == requests.codes.ok:
+            print("It was probably loaded at start-up", end='... ', flush=True)
+        # TODO an else-clause to abort test would probably be nice here
+    return response
+
+
 def unmount_device(node_id):
     url = URL_CONFIG_NETCONF_TOPO + "node/" + node_id
     response = delete_request(url)
@@ -306,6 +328,19 @@ def connect_rdm_to_xpdr_request(xpdr_node: str, xpdr_num: str, network_num: str,
     return post_request(url, data)
 
 
+def connect_xpdr_to_rdm_tapi_request(xpdr_node: str, xpdr_num: str, rdm_node: str, srg_num: str):
+    url = "{}/operations/transportpce-tapinetworkutils:init-xpdr-rdm-tapi-link"
+    data = {
+        "input": {
+            "xpdr-node": xpdr_node,
+            "network-tp": xpdr_num,
+            "rdm-node": rdm_node,
+            "add-drop-tp": srg_num
+        }
+    }
+    return post_request(url, data)
+
+
 def check_netconf_node_request(node: str, suffix: str):
     url = URL_CONFIG_NETCONF_TOPO + (
         "node/" + node + "/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + suffix
@@ -406,7 +441,7 @@ def service_path_request(operation: str, servicename: str, wavenumber: str, node
         "renderer:operation": operation,
         "renderer:nodes": nodes,
         "renderer:center-freq": centerfreq,
-        "renderer:width": slotwidth,
+        "renderer:nmc-width": slotwidth,
         "renderer:min-freq": minfreq,
         "renderer:max-freq": maxfreq,
         "renderer:lower-spectral-slot-number": lowerslotnumber,
@@ -507,6 +542,7 @@ def start_honeynode(log_file: str, sim):
 
 def wait_until_log_contains(log_file, regexp, time_to_wait=60):
     # pylint: disable=lost-exception
+    # pylint: disable=consider-using-with
     stringfound = False
     filefound = False
     line = None