add a method to check nodes configs in func tests

[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 b0daf88fb30551b38e9fd4a34d570a9c53358143..6b6958f73cf09e834b8fa04a7f9f49f533f053cb 100644 (file)
--- a/tests/transportpce_tests/common/test_utils.py
+++ b/tests/transportpce_tests/common/test_utils.py
@@ -34,6 +34,7 @@ ODL_LOGIN = "admin"
 ODL_PWD = "admin"
 NODES_LOGIN = "admin"
 NODES_PWD = "admin"
+URL_CONFIG_NETCONF_TOPO = "{}/config/network-topology:network-topology/topology/topology-netconf/"
 
 TYPE_APPLICATION_JSON = {'Content-Type': 'application/json', 'Accept': 'application/json'}
 TYPE_APPLICATION_XML = {'Content-Type': 'application/xml', 'Accept': 'application/xml'}
@@ -122,28 +123,70 @@ def install_karaf_feature(feature_name: str):
                           universal_newlines=True)
 
 
-def post_request(url, data, username, password):
+def get_request(url):
     return requests.request(
-        "POST", url, data=json.dumps(data),
-        headers=TYPE_APPLICATION_JSON, auth=(username, password))
+        "GET", url.format(RESTCONF_BASE_URL),
+        headers=TYPE_APPLICATION_JSON,
+        auth=(ODL_LOGIN, ODL_PWD))
+
+
+def post_request(url, data):
+    if data:
+        return requests.request(
+            "POST", url.format(RESTCONF_BASE_URL),
+            data=json.dumps(data),
+            headers=TYPE_APPLICATION_JSON,
+            auth=(ODL_LOGIN, ODL_PWD))
+    else:
+        return requests.request(
+            "POST", url.format(RESTCONF_BASE_URL),
+            headers=TYPE_APPLICATION_JSON,
+            auth=(ODL_LOGIN, ODL_PWD))
+
 
+def post_xmlrequest(url, data):
+    if data:
+        return requests.request(
+            "POST", url.format(RESTCONF_BASE_URL),
+            data=data,
+            headers=TYPE_APPLICATION_XML,
+            auth=(ODL_LOGIN, ODL_PWD))
 
-def put_request(url, data, username, password):
+
+def put_request(url, data):
     return requests.request(
-        "PUT", url, data=json.dumps(data), headers=TYPE_APPLICATION_JSON,
-        auth=(username, password))
+        "PUT", url.format(RESTCONF_BASE_URL),
+        data=json.dumps(data),
+        headers=TYPE_APPLICATION_JSON,
+        auth=(ODL_LOGIN, ODL_PWD))
 
 
-def delete_request(url, username, password):
+def put_xmlrequest(url, data):
     return requests.request(
-        "DELETE", url, headers=TYPE_APPLICATION_JSON,
-        auth=(username, password))
+        "PUT", url.format(RESTCONF_BASE_URL),
+        data=data,
+        headers=TYPE_APPLICATION_XML,
+        auth=(ODL_LOGIN, ODL_PWD))
+
+
+def rawput_request(url, data):
+    return requests.request(
+        "PUT", url.format(RESTCONF_BASE_URL),
+        data=data,
+        headers=TYPE_APPLICATION_JSON,
+        auth=(ODL_LOGIN, ODL_PWD))
+
+
+def delete_request(url):
+    return requests.request(
+        "DELETE", url.format(RESTCONF_BASE_URL),
+        headers=TYPE_APPLICATION_JSON,
+        auth=(ODL_LOGIN, ODL_PWD))
 
 
 def mount_device(node_id, sim):
-    url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/"
-           + node_id).format(RESTCONF_BASE_URL)
-    headers = {"node": [{
+    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,
@@ -151,7 +194,7 @@ def mount_device(node_id, sim):
         "netconf-node-topology:port": SIMS[sim]['port'],
         "netconf-node-topology:tcp-only": "false",
         "netconf-node-topology:pass-through": {}}]}
-    response = put_request(url, headers, ODL_LOGIN, ODL_PWD)
+    response = put_request(url, body)
     if wait_until_log_contains(TPCE_LOG, re.escape("Triggering notification stream NETCONF for node "+node_id), 60):
         print("Node "+node_id+" correctly added to tpce topology", end='... ', flush=True)
     else:
@@ -163,9 +206,8 @@ def mount_device(node_id, sim):
 
 
 def unmount_device(node_id):
-    url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/"
-           + node_id).format(RESTCONF_BASE_URL)
-    response = delete_request(url, ODL_LOGIN, ODL_PWD)
+    url = URL_CONFIG_NETCONF_TOPO+"node/"+node_id
+    response = delete_request(url)
     if wait_until_log_contains(TPCE_LOG, re.escape("onDeviceDisConnected: "+node_id), 60):
         print("Node "+node_id+" correctly deleted from tpce topology", end='... ', flush=True)
     else:
@@ -173,8 +215,9 @@ def unmount_device(node_id):
     return response
 
 
-def generate_link_data(xpdr_node: str, xpdr_num: str, network_num: str, rdm_node: str, srg_num: str,
-                       termination_num: str):
+def connect_xpdr_to_rdm_request(xpdr_node: str, xpdr_num: str, network_num: str,
+                                rdm_node: str, srg_num: str, termination_num: str):
+    url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links"
     data = {
         "networkutils:input": {
             "networkutils:links-input": {
@@ -187,7 +230,32 @@ def generate_link_data(xpdr_node: str, xpdr_num: str, network_num: str, rdm_node
             }
         }
     }
-    return data
+    return post_request(url, data)
+
+
+def connect_rdm_to_xpdr_request(xpdr_node: str, xpdr_num: str, network_num: str,
+                                rdm_node: str, srg_num: str, termination_num: str):
+    url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links"
+    data = {
+        "networkutils:input": {
+            "networkutils:links-input": {
+                "networkutils:xpdr-node": xpdr_node,
+                "networkutils:xpdr-num": xpdr_num,
+                "networkutils:network-num": network_num,
+                "networkutils:rdm-node": rdm_node,
+                "networkutils:srg-num": srg_num,
+                "networkutils:termination-point-num": termination_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
+    )
+    return get_request(url)
 
 
 def shutdown_process(process):
@@ -222,13 +290,13 @@ def wait_until_log_contains(log_file, regexp, time_to_wait=20):
             while True:
                 line = filelogs.readline()
                 if compiled_regexp.search(line):
-                    print("String found!", end=' ')
+                    print("Pattern found!", end=' ')
                     stringfound = True
                     break
                 if not line:
                     time.sleep(0.1)
     except TimeoutError:
-        print("String not found after "+str(time_to_wait), end=" seconds! ", flush=True)
+        print("Pattern not found after "+str(time_to_wait), end=" seconds! ", flush=True)
     except PermissionError:
         print("Permission Error when trying to access the log file", end=" ... ", flush=True)
     finally: