Fix new pylint-3.0.0 detected issues

[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 cc0604abb6f6c15135eddcd4501b5b35de4b8e47..888a8e91fc228751b85ecd386c41bd49cd2f4537 100644 (file)
--- a/tests/transportpce_tests/common/test_utils.py
+++ b/tests/transportpce_tests/common/test_utils.py
@@ -22,6 +22,11 @@ import time
 
 import psutil
 import requests
+import urllib.parse
+
+from dict2xml import dict2xml
+from netconf_client.connect import connect_ssh
+from netconf_client.ncclient import Manager
 
 # pylint: disable=import-error
 import simulators
@@ -85,6 +90,19 @@ if 'USE_LIGHTY' in os.environ and os.environ['USE_LIGHTY'] == 'True':
 else:
     TPCE_LOG = KARAF_LOG
 
+if 'USE_SIMS' not in os.environ:
+    SIMS_TO_USE = 'lightynode'
+    SIMS_TYPE = 'lightynode'
+else:
+    SIMS_TO_USE = os.environ['USE_SIMS']
+    print("Forcing to use SIMS " + SIMS_TO_USE)
+    if SIMS_TO_USE != 'None' or 'SIMS_TYPE' not in os.environ:
+        SIMS_TYPE = SIMS_TO_USE
+    else:
+        SIMS_TYPE = os.environ['SIMS_TYPE']
+        print("Forcing to use SIMS type" + SIMS_TYPE)
+
+
 #
 # Basic HTTP operations
 #
@@ -162,14 +180,9 @@ def start_lightynode(log_file: str, sim):
 
 
 def start_sims(sims_list):
-    if 'USE_SIMS' in os.environ:
-        sims_to_use = os.environ['USE_SIMS']
-        print("Using SIMS " + sims_to_use)
-    else:
-        sims_to_use = 'lightynode'
-    if sims_to_use == 'None':
+    if SIMS_TO_USE == 'None':
         return None
-    if sims_to_use == 'honeynode':
+    if SIMS_TO_USE == 'honeynode':
         start_msg = HONEYNODE_OK_START_MSG
         start_method = start_honeynode
     else:
@@ -260,34 +273,29 @@ 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
     try:
         with TimeOut(seconds=time_to_wait):
             while not os.path.exists(log_file):
                 time.sleep(0.2)
-            filelogs = open(log_file, 'r', encoding='utf-8')
-            filelogs.seek(0, 2)
-            filefound = True
-            print("Searching for pattern '" + regexp + "' in " + os.path.basename(log_file), end='... ', flush=True)
-            compiled_regexp = re.compile(regexp)
-            while True:
-                line = filelogs.readline()
-                if compiled_regexp.search(line):
-                    print('Pattern found!', end=' ')
-                    stringfound = True
-                    break
-                if not line:
-                    time.sleep(0.1)
+            with open(log_file, 'r', encoding='utf-8') as filelogs:
+                filelogs.seek(0, 2)
+                print("Searching for pattern '" + regexp + "' in " + os.path.basename(log_file), end='... ', flush=True)
+                compiled_regexp = re.compile(regexp)
+                while True:
+                    line = filelogs.readline()
+                    if compiled_regexp.search(line):
+                        print('Pattern found!', end=' ')
+                        stringfound = True
+                        break
+                    if not line:
+                        time.sleep(0.1)
+        return stringfound
     except TimeoutError:
         print('Pattern not found after ' + str(time_to_wait), end=' seconds! ', flush=True)
+        return stringfound
     except PermissionError:
         print('Permission Error when trying to access the log file', end=' ... ', flush=True)
-    finally:
-        if filefound:
-            filelogs.close()
-        else:
-            print('log file does not exist or is not accessible... ', flush=True)
         return stringfound
 
 
@@ -683,12 +691,20 @@ def transportpce_api_rpc_request(api_module: str, rpc: str, payload: dict):
 
 
 def sims_update_cp_port(sim: tuple, circuitpack: str, port: str, payload: dict):
+    if SIMS_TYPE == 'lightynode':
+        return sims_update_cp_port_ntcf(sim, circuitpack, payload)
+    if SIMS_TYPE == 'honeynode':
+        return sims_update_cp_port_rest(sim, circuitpack, port, payload)
+    return False
+
+
+def sims_update_cp_port_rest(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)
+        SIMS[sim]['restconf_baseurl'],
+        urllib.parse.quote(circuitpack, safe=''),
+        urllib.parse.quote(port, safe=''))
     body = {"ports": [payload]}
-    print(sim)
-    print(url)
     response = requests.request("PUT",
                                 url,
                                 data=json.dumps(body),
@@ -696,3 +712,19 @@ def sims_update_cp_port(sim: tuple, circuitpack: str, port: str, payload: dict):
                                 auth=(ODL_LOGIN, ODL_PWD),
                                 timeout=REQUEST_TIMEOUT)
     return response.status_code == requests.codes.ok
+
+
+def sims_update_cp_port_ntcf(sim: tuple, circuitpack: str, payload: dict):
+    body = {"circuit-packs": {"circuit-pack-name": circuitpack, "ports": payload}}
+    xml_body = '<config><org-openroadm-device xmlns="http://org/openroadm/device">'
+    xml_body += dict2xml(body, indent="  ")
+    xml_body += '</org-openroadm-device></config>'
+    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.edit_config(xml_body, target="candidate", default_operation="merge")
+    if "None" in str(reply):
+        return True
+    return False