##############################################################################
# pylint: disable=no-member
+# pylint: disable=too-many-arguments
import json
import os
+# pylint: disable=wrong-import-order
import sys
import re
import signal
import psutil
import requests
+# pylint: disable=import-error
import simulators
SIMS = simulators.SIMS
URL_CONFIG_ORDM_NET = "{}/config/ietf-network:networks/network/openroadm-network/"
URL_PORTMAPPING = "{}/config/transportpce-portmapping:network/nodes/"
URL_OPER_SERV_LIST = "{}/operational/org-openroadm-service:service-list/"
+URL_OPER_SERV_PATH_LIST = "{}/operational/transportpce-service-path:service-path-list/"
URL_GET_NBINOTIFICATIONS_PROCESS_SERV = "{}/operations/nbi-notifications:get-notifications-process-service/"
URL_GET_NBINOTIFICATIONS_ALARM_SERV = "{}/operations/nbi-notifications:get-notifications-alarm-service/"
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_TAPI_GET_CONNECTIVITY = "{}/operations/tapi-connectivity:get-connectivity-service-details"
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_NEP_DETAILS = "{}/operations/tapi-topology:get-node-edge-point-details"
+URL_TAPI_SIP_LIST = "{}/operations/tapi-common:get-service-interface-point-list"
+URL_TAPI_SERVICE_LIST = "{}/operations/tapi-connectivity:get-connectivity-service-list"
+URL_TAPI_NOTIFICATION_SUBS_SERVICE = "{}/operations/tapi-notification:create-notification-subscription-service"
+URL_TAPI_GET_NOTIFICATION_LIST = "{}/operations/tapi-notification:get-notification-list"
TYPE_APPLICATION_JSON = {'Content-Type': 'application/json', 'Accept': 'application/json'}
TYPE_APPLICATION_XML = {'Content-Type': 'application/xml', 'Accept': 'application/xml'}
+REQUEST_TIMEOUT = 10
+
CODE_SHOULD_BE_200 = 'Http status code should be 200'
CODE_SHOULD_BE_201 = 'Http status code should be 201'
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)
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)
return requests.request(
"GET", url.format(RESTCONF_BASE_URL),
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
def post_request(url, data):
"POST", url.format(RESTCONF_BASE_URL),
data=json.dumps(data),
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
return requests.request(
"POST", url.format(RESTCONF_BASE_URL),
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
def post_xmlrequest(url, data):
"POST", url.format(RESTCONF_BASE_URL),
data=data,
headers=TYPE_APPLICATION_XML,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
return None
"PUT", url.format(RESTCONF_BASE_URL),
data=json.dumps(data),
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
def put_xmlrequest(url, data):
"PUT", url.format(RESTCONF_BASE_URL),
data=data,
headers=TYPE_APPLICATION_XML,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
def put_jsonrequest(url, data):
"PUT", url.format(RESTCONF_BASE_URL),
data=data,
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
def rawput_request(url, data):
"PUT", url.format(RESTCONF_BASE_URL),
data=data,
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
def rawpost_request(url, data):
"POST", url.format(RESTCONF_BASE_URL),
data=data,
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
def delete_request(url):
return requests.request(
"DELETE", url.format(RESTCONF_BASE_URL),
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
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)
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
def get_service_list_request(suffix: str):
- url = URL_OPER_SERV_LIST + suffix
- return get_request(url)
+ return get_request(URL_OPER_SERV_LIST + suffix)
+
+
+def get_service_path_list_request(suffix: str):
+ return get_request(URL_OPER_SERV_PATH_LIST + suffix)
def service_create_request(attr):
def service_path_request(operation: str, servicename: str, wavenumber: str, nodes, centerfreq: str,
slotwidth: int, minfreq: float, maxfreq: float, lowerslotnumber: int,
- higherslotnumber: int):
+ higherslotnumber: int, modulation_format="dp-qpsk"):
attr = {"renderer:input": {
"renderer:service-name": servicename,
"renderer:wave-number": wavenumber,
- "renderer:modulation-format": "dp-qpsk",
+ "renderer:modulation-format": modulation_format,
"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,
"service-handler-header": {"request-id": requestid},
"service-a-end": serviceaend,
"service-z-end": servicezend,
- "pce-metric": metric}
+ "pce-routing-metric": metric}
if hardconstraints:
attr.update({"hard-constraints": hardconstraints})
if softconstraints:
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_get_connectivity_request(serviceidorname):
+ attr = {
+ "input": {
+ "service-id-or-name": serviceidorname}}
+ return post_request(URL_TAPI_GET_CONNECTIVITY, attr)
+
+
+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_node_edge_point_details_request(topologyidorname, nodeidorname, nepidorname):
+ attr = {
+ "input": {
+ "topology-id-or-name": topologyidorname,
+ "node-id-or-name": nodeidorname,
+ "ep-id-or-name": nepidorname}}
+ return post_request(URL_TAPI_NEP_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 tapi_create_notification_subscription_service_request(attr):
+ return post_request(URL_TAPI_NOTIFICATION_SUBS_SERVICE, attr)
+
+
+def tapi_get_notifications_list_request(attr):
+ return post_request(URL_TAPI_GET_NOTIFICATION_LIST, attr)
+
+
def shutdown_process(process):
if process is not None:
for child in psutil.Process(process.pid).children():
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)
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
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)
Code Review / transportpce.git / blobdiff