Fix more pylint warnings

[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 b70a4eeb5dbedda055726bfe65b3fee3f8700179..c18a4d1f5ee1288834881c70e5f9835ae3864ef9 100644 (file)
--- a/tests/transportpce_tests/common/test_utils.py
+++ b/tests/transportpce_tests/common/test_utils.py
@@ -13,6 +13,7 @@
 
 import json
 import os
+# pylint: disable=wrong-import-order
 import sys
 import re
 import signal
@@ -22,6 +23,7 @@ import time
 import psutil
 import requests
 
+# pylint: disable=import-error
 import simulators
 
 SIMS = simulators.SIMS
@@ -48,9 +50,15 @@ URL_SERV_CREATE = "{}/operations/org-openroadm-service:service-create"
 URL_SERV_DELETE = "{}/operations/org-openroadm-service:service-delete"
 URL_SERVICE_PATH = "{}/operations/transportpce-device-renderer:service-path"
 URL_OTN_SERVICE_PATH = "{}/operations/transportpce-device-renderer:otn-service-path"
+URL_TAPI_CREATE_CONNECTIVITY = "{}/operations/tapi-connectivity:create-connectivity-service"
+URL_TAPI_DELETE_CONNECTIVITY = "{}/operations/tapi-connectivity:delete-connectivity-service"
 URL_CREATE_OTS_OMS = "{}/operations/transportpce-device-renderer:create-ots-oms"
 URL_PATH_COMPUTATION_REQUEST = "{}/operations/transportpce-pce:path-computation-request"
 URL_FULL_PORTMAPPING = "{}/config/transportpce-portmapping:network"
+URL_TAPI_TOPOLOGY_DETAILS = "{}/operations/tapi-topology:get-topology-details"
+URL_TAPI_NODE_DETAILS = "{}/operations/tapi-topology:get-node-details"
+URL_TAPI_SIP_LIST = "{}/operations/tapi-common:get-service-interface-point-list"
+URL_TAPI_SERVICE_LIST = "{}/operations/tapi-connectivity:get-connectivity-service-list"
 
 TYPE_APPLICATION_JSON = {'Content-Type': 'application/json', 'Accept': 'application/json'}
 TYPE_APPLICATION_XML = {'Content-Type': 'application/xml', 'Accept': 'application/xml'}
@@ -125,7 +133,7 @@ def start_karaf():
     executable = os.path.join(
         os.path.dirname(os.path.realpath(__file__)),
         "..", "..", "..", KARAF_INSTALLDIR, "target", "assembly", "bin", "karaf")
-    with open('odl.log', 'w') as outfile:
+    with open('odl.log', 'w', encoding='utf-8') as outfile:
         return subprocess.Popen(
             ["sh", executable, "server"], stdout=outfile, stderr=outfile, stdin=None)
 
@@ -136,7 +144,7 @@ def start_lighty():
         os.path.dirname(os.path.realpath(__file__)),
         "..", "..", "..", "lighty", "target", "tpce",
         "clean-start-controller.sh")
-    with open(TPCE_LOG, 'w') as outfile:
+    with open(TPCE_LOG, 'w', encoding='utf-8') as outfile:
         return subprocess.Popen(
             ["sh", executable], stdout=outfile, stderr=outfile, stdin=None)
 
@@ -199,6 +207,7 @@ def put_xmlrequest(url, data):
         headers=TYPE_APPLICATION_XML,
         auth=(ODL_LOGIN, ODL_PWD))
 
+
 def put_jsonrequest(url, data):
     return requests.request(
         "PUT", url.format(RESTCONF_BASE_URL),
@@ -206,6 +215,7 @@ def put_jsonrequest(url, data):
         headers=TYPE_APPLICATION_JSON,
         auth=(ODL_LOGIN, ODL_PWD))
 
+
 def rawput_request(url, data):
     return requests.request(
         "PUT", url.format(RESTCONF_BASE_URL),
@@ -213,6 +223,7 @@ def rawput_request(url, data):
         headers=TYPE_APPLICATION_JSON,
         auth=(ODL_LOGIN, ODL_PWD))
 
+
 def rawpost_request(url, data):
     return requests.request(
         "POST", url.format(RESTCONF_BASE_URL),
@@ -439,6 +450,40 @@ def path_computation_request(requestid: str, servicename: str, serviceaend, serv
     return post_request(URL_PATH_COMPUTATION_REQUEST, {"input": attr})
 
 
+def tapi_create_connectivity_request(topologyidorname):
+    return post_request(URL_TAPI_CREATE_CONNECTIVITY, topologyidorname)
+
+
+def tapi_delete_connectivity_request(serviceidorname):
+    attr = {
+        "input": {
+            "service-id-or-name": serviceidorname}}
+    return post_request(URL_TAPI_DELETE_CONNECTIVITY, attr)
+
+
+def tapi_get_topology_details_request(topologyidorname):
+    attr = {
+        "input": {
+            "topology-id-or-name": topologyidorname}}
+    return post_request(URL_TAPI_TOPOLOGY_DETAILS, attr)
+
+
+def tapi_get_node_details_request(topologyidorname, nodeidorname):
+    attr = {
+        "input": {
+            "topology-id-or-name": topologyidorname,
+            "node-id-or-name": nodeidorname}}
+    return post_request(URL_TAPI_NODE_DETAILS, attr)
+
+
+def tapi_get_sip_details_request():
+    return post_request(URL_TAPI_SIP_LIST, "")
+
+
+def tapi_get_service_list_request():
+    return post_request(URL_TAPI_SERVICE_LIST, "")
+
+
 def shutdown_process(process):
     if process is not None:
         for child in psutil.Process(process.pid).children():
@@ -453,7 +498,7 @@ def start_honeynode(log_file: str, sim):
     sample_directory = os.path.join(os.path.dirname(os.path.realpath(__file__)),
                                     "..", "..", "sample_configs", "openroadm", sim[1])
     if os.path.isfile(executable):
-        with open(log_file, 'w') as outfile:
+        with open(log_file, 'w', encoding='utf-8') as outfile:
             return subprocess.Popen(
                 [executable, SIMS[sim]['port'], os.path.join(sample_directory, SIMS[sim]['configfile'])],
                 stdout=outfile, stderr=outfile)
@@ -469,7 +514,7 @@ def wait_until_log_contains(log_file, regexp, time_to_wait=60):
         with TimeOut(seconds=time_to_wait):
             while not os.path.exists(log_file):
                 time.sleep(0.2)
-            filelogs = open(log_file, 'r')
+            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)