X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2Fcommon%2Ftest_utils.py;h=1c0bf202cfedc8e2efce9e0f146637fabb5254ab;hb=db87ba733190c2e59997e00e0fe9199c9c74234d;hp=5a043a7f0f3f50b7f21dc3d1a2ff6f76f914db9e;hpb=97d0dba6ebff111c98efd9632c3e1c2b613614ab;p=transportpce.git diff --git a/tests/transportpce_tests/common/test_utils.py b/tests/transportpce_tests/common/test_utils.py index 5a043a7f0..1c0bf202c 100644 --- 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 @@ -29,7 +34,8 @@ import simulators SIMS = simulators.SIMS HONEYNODE_OK_START_MSG = 'Netconf SSH endpoint started successfully at 0.0.0.0' -KARAF_OK_START_MSG = "Blueprint container for bundle org.opendaylight.netconf.restconf.* was successfully created" +LIGHTYNODE_OK_START_MSG = 'Data tree change listeners registered' +KARAF_OK_START_MSG = "Transportpce controller started" LIGHTY_OK_START_MSG = re.escape("lighty.io and RESTCONF-NETCONF started") ODL_LOGIN = 'admin' @@ -47,6 +53,9 @@ CODE_SHOULD_BE_201 = 'Http status code should be 201' T100GE = 'Transponder 100GE' T0_MULTILAYER_TOPO = 'T0 - Multi-layer topology' T0_FULL_MULTILAYER_TOPO = 'T0 - Full Multi-layer topology' +T100GE_UUID = 'cf51c729-3699-308a-a7d0-594c6a62ebbb' +T0_MULTILAYER_TOPO_UUID = '747c670e-7a07-3dab-b379-5b1cd17402a3' +T0_FULL_MULTILAYER_TOPO_UUID = '393f09a4-0a0b-3d82-a4f6-1fbbc14ca1a7' SIM_LOG_DIRECTORY = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'log') @@ -58,8 +67,8 @@ else: RESTCONF_PORT = 8181 RESTCONF_PATH_PREFIX = {'rfc8040': '/rests', - 'lighty': '/restconf', 'draft-bierman02': '/restconf'} + if 'USE_ODL_RESTCONF_VERSION' in os.environ: RESTCONF_VERSION = os.environ['USE_ODL_RESTCONF_VERSION'] if RESTCONF_VERSION not in RESTCONF_PATH_PREFIX: @@ -84,6 +93,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 # @@ -133,12 +155,47 @@ def post_request(url, data): # +def start_honeynode(log_file: str, sim): + executable = os.path.join(os.path.dirname(os.path.realpath(__file__)), + '..', '..', 'honeynode', sim[1], 'honeynode-simulator', 'honeycomb-tpce') + 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', encoding='utf-8') as outfile: + return subprocess.Popen( + [executable, SIMS[sim]['port'], os.path.join(sample_directory, SIMS[sim]['configfile'])], + stdout=outfile, stderr=outfile) + return None + + +def start_lightynode(log_file: str, sim): + executable = os.path.join(os.path.dirname(os.path.realpath(__file__)), + '..', '..', 'lightynode', 'lightynode-openroadm-device', 'start-device.sh') + 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', encoding='utf-8') as outfile: + return subprocess.Popen( + [executable, "-v" + sim[1], "-p" + SIMS[sim]['port'], "-f" + os.path.join(sample_directory, + SIMS[sim]['configfile'])], + stdout=outfile, stderr=outfile) + return None + + def start_sims(sims_list): + if SIMS_TO_USE == 'None': + return None + if SIMS_TO_USE == 'honeynode': + start_msg = HONEYNODE_OK_START_MSG + start_method = start_honeynode + else: + start_msg = LIGHTYNODE_OK_START_MSG + start_method = start_lightynode for sim in sims_list: print('starting simulator ' + sim[0] + ' in OpenROADM device version ' + sim[1] + '...') log_file = os.path.join(SIM_LOG_DIRECTORY, SIMS[sim]['logfile']) - process = start_honeynode(log_file, sim) - if wait_until_log_contains(log_file, HONEYNODE_OK_START_MSG, 100): + process = start_method(log_file, sim) + if wait_until_log_contains(log_file, start_msg, 100): print('simulator for ' + sim[0] + ' started') else: print('simulator for ' + sim[0] + ' failed to start') @@ -151,6 +208,9 @@ def start_sims(sims_list): def start_tpce(): + if 'NO_ODL_STARTUP' in os.environ: + print('No OpenDaylight instance to start!') + return [] print('starting OpenDaylight...') if 'USE_LIGHTY' in os.environ and os.environ['USE_LIGHTY'] == 'True': process = start_lighty() @@ -158,7 +218,7 @@ def start_tpce(): else: process = start_karaf() start_msg = KARAF_OK_START_MSG - if wait_until_log_contains(TPCE_LOG, start_msg, time_to_wait=300): + if wait_until_log_contains(TPCE_LOG, start_msg, time_to_wait=100): print('OpenDaylight started !') else: print('OpenDaylight failed to start !') @@ -196,7 +256,9 @@ def install_karaf_feature(feature_name: str): executable = os.path.join( os.path.dirname(os.path.realpath(__file__)), '..', '..', '..', KARAF_INSTALLDIR, 'target', 'assembly', 'bin', 'client') - return subprocess.run([executable], +# FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-701 +# -b option needed below because of Karaf client bug reporte in the JIRA ticket mentioned above + return subprocess.run([executable, '-b'], input='feature:install ' + feature_name + '\n feature:list | grep ' + feature_name + ' \n logout \n', universal_newlines=True, check=False) @@ -210,51 +272,33 @@ def shutdown_process(process): process.send_signal(signal.SIGINT) -def start_honeynode(log_file: str, sim): - executable = os.path.join(os.path.dirname(os.path.realpath(__file__)), - '..', '..', 'honeynode', sim[1], 'honeynode-simulator', 'honeycomb-tpce') - 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', encoding='utf-8') as outfile: - return subprocess.Popen( - [executable, SIMS[sim]['port'], os.path.join(sample_directory, SIMS[sim]['configfile'])], - stdout=outfile, stderr=outfile) - return None - - 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 @@ -281,7 +325,6 @@ class TimeOut: def mount_device(node: str, sim: str): url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}', - 'lighty': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}', 'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}'} body = {'node': [{ 'node-id': node, @@ -304,7 +347,6 @@ def mount_device(node: str, sim: str): def unmount_device(node: str): url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}', - 'lighty': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}', 'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}'} response = delete_request(url[RESTCONF_VERSION].format('{}', node)) if wait_until_log_contains(TPCE_LOG, re.escape("onDeviceDisConnected: " + node), 180): @@ -316,12 +358,10 @@ def unmount_device(node: str): def check_device_connection(node: str): url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}?content=nonconfig', - 'lighty': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}?content=nonconfig', 'draft-bierman02': '{}/operational/network-topology:network-topology/topology/topology-netconf/node/{}'} response = get_request(url[RESTCONF_VERSION].format('{}', node)) res = response.json() return_key = {'rfc8040': 'network-topology:node', - 'lighty': 'network-topology:node', 'draft-bierman02': 'node'} if return_key[RESTCONF_VERSION] in res.keys(): connection_status = res[return_key[RESTCONF_VERSION]][0]['netconf-node-topology:connection-status'] @@ -334,12 +374,10 @@ def check_device_connection(node: str): def check_node_request(node: str): # pylint: disable=line-too-long url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device?content=config', # nopep8 - 'lighty': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device?content=config', # nopep8 'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device'} # nopep8 response = get_request(url[RESTCONF_VERSION].format('{}', node)) res = response.json() return_key = {'rfc8040': 'org-openroadm-device:org-openroadm-device', - 'lighty': 'org-openroadm-device:org-openroadm-device', 'draft-bierman02': 'org-openroadm-device'} if return_key[RESTCONF_VERSION] in res.keys(): response_attribute = res[return_key[RESTCONF_VERSION]] @@ -352,17 +390,19 @@ def check_node_request(node: str): def check_node_attribute_request(node: str, attribute: str, attribute_value: str): # pylint: disable=line-too-long url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}={}?content=nonconfig', # nopep8 - 'lighty': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}={}?content=nonconfig', # nopep8 'draft-bierman02': '{}/operational/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}/{}'} # nopep8 response = get_request(url[RESTCONF_VERSION].format('{}', node, attribute, attribute_value)) res = response.json() return_key = {'rfc8040': 'org-openroadm-device:' + attribute, - 'lighty': 'org-openroadm-device:' + attribute, 'draft-bierman02': attribute} if return_key[RESTCONF_VERSION] in res.keys(): response_attribute = res[return_key[RESTCONF_VERSION]] - else: + elif 'errors' in res.keys(): response_attribute = res['errors']['error'][0] + else: + # status code 400 invalid request + response_attribute = res['message'] + ' ' + res['url'] + print(response_attribute) return {'status_code': response.status_code, attribute: response_attribute} @@ -370,7 +410,6 @@ def check_node_attribute_request(node: str, attribute: str, attribute_value: str def check_node_attribute2_request(node: str, attribute: str, attribute_value: str, attribute2: str): # pylint: disable=line-too-long url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}={}/{}?content=config', # nopep8 - 'lighty': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}={}/{}?content=config', # nopep8 'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}/{}/{}'} # nopep8 response = get_request(url[RESTCONF_VERSION].format('{}', node, attribute, attribute_value, attribute2)) res = response.json() @@ -385,7 +424,6 @@ def check_node_attribute2_request(node: str, attribute: str, attribute_value: st def del_node_attribute_request(node: str, attribute: str, attribute_value: str): # pylint: disable=line-too-long url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}={}', # nopep8 - 'lighty': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}={}', # nopep8 'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}/{}'} # nopep8 response = delete_request(url[RESTCONF_VERSION].format('{}', node, attribute, attribute_value)) return response @@ -397,7 +435,6 @@ def del_node_attribute_request(node: str, attribute: str, attribute_value: str): def post_portmapping(payload: str): url = {'rfc8040': '{}/data/transportpce-portmapping:network', - 'lighty': '{}/data/transportpce-portmapping:network', 'draft-bierman02': '{}/config/transportpce-portmapping:network'} json_payload = json.loads(payload) response = post_request(url[RESTCONF_VERSION].format('{}'), json_payload) @@ -406,7 +443,6 @@ def post_portmapping(payload: str): def del_portmapping(): url = {'rfc8040': '{}/data/transportpce-portmapping:network', - 'lighty': '{}/data/transportpce-portmapping:network', 'draft-bierman02': '{}/config/transportpce-portmapping:network'} response = delete_request(url[RESTCONF_VERSION].format('{}')) return {'status_code': response.status_code} @@ -415,20 +451,18 @@ def del_portmapping(): def get_portmapping_node_attr(node: str, attr: str, value: str): # pylint: disable=consider-using-f-string url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}', - 'lighty': '{}/data/transportpce-portmapping:network/nodes={}', 'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}'} target_url = url[RESTCONF_VERSION].format('{}', node) if attr is not None: target_url = (target_url + '/{}').format('{}', attr) if value is not None: - suffix = {'rfc8040': '={}', 'lighty': '={}', 'draft-bierman02': '/{}'} + suffix = {'rfc8040': '={}', 'draft-bierman02': '/{}'} target_url = (target_url + suffix[RESTCONF_VERSION]).format('{}', value) else: attr = 'nodes' response = get_request(target_url) res = response.json() return_key = {'rfc8040': 'transportpce-portmapping:' + attr, - 'lighty': 'transportpce-portmapping:' + attr, 'draft-bierman02': attr} if return_key[RESTCONF_VERSION] in res.keys(): return_output = res[return_key[RESTCONF_VERSION]] @@ -444,9 +478,8 @@ def get_portmapping_node_attr(node: str, attr: str, value: str): def get_ietf_network_request(network: str, content: str): url = {'rfc8040': '{}/data/ietf-network:networks/network={}?content={}', - 'lighty': '{}/data/ietf-network:networks/network={}?content={}', 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}'} - if RESTCONF_VERSION in ('rfc8040', 'lighty'): + if RESTCONF_VERSION in ('rfc8040'): format_args = ('{}', network, content) elif content == 'config': format_args = ('{}', content, network) @@ -456,7 +489,6 @@ def get_ietf_network_request(network: str, content: str): if bool(response): res = response.json() return_key = {'rfc8040': 'ietf-network:network', - 'lighty': 'ietf-network:network', 'draft-bierman02': 'network'} networks = res[return_key[RESTCONF_VERSION]] else: @@ -467,7 +499,6 @@ def get_ietf_network_request(network: str, content: str): def put_ietf_network(network: str, payload: str): url = {'rfc8040': '{}/data/ietf-network:networks/network={}', - 'lighty': '{}/data/ietf-network:networks/network={}', 'draft-bierman02': '{}/config/ietf-network:networks/network/{}'} json_payload = json.loads(payload) response = put_request(url[RESTCONF_VERSION].format('{}', network), json_payload) @@ -476,7 +507,6 @@ def put_ietf_network(network: str, payload: str): def del_ietf_network(network: str): url = {'rfc8040': '{}/data/ietf-network:networks/network={}', - 'lighty': '{}/data/ietf-network:networks/network={}', 'draft-bierman02': '{}/config/ietf-network:networks/network/{}'} response = delete_request(url[RESTCONF_VERSION].format('{}', network)) return {'status_code': response.status_code} @@ -484,9 +514,8 @@ def del_ietf_network(network: str): def get_ietf_network_link_request(network: str, link: str, content: str): url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}?content={}', - 'lighty': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}?content={}', 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/ietf-network-topology:link/{}'} - if RESTCONF_VERSION in ('rfc8040', 'lighty'): + if RESTCONF_VERSION in ('rfc8040'): format_args = ('{}', network, link, content) elif content == 'config': format_args = ('{}', content, network, link) @@ -495,7 +524,6 @@ def get_ietf_network_link_request(network: str, link: str, content: str): response = get_request(url[RESTCONF_VERSION].format(*format_args)) res = response.json() return_key = {'rfc8040': 'ietf-network-topology:link', - 'lighty': 'ietf-network-topology:link', 'draft-bierman02': 'ietf-network-topology:link'} link = res[return_key[RESTCONF_VERSION]][0] return {'status_code': response.status_code, @@ -504,9 +532,8 @@ def get_ietf_network_link_request(network: str, link: str, content: str): def del_ietf_network_link_request(network: str, link: str, content: str): url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}?content={}', - 'lighty': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}?content={}', 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/ietf-network-topology:link/{}'} - if RESTCONF_VERSION in ('rfc8040', 'lighty'): + if RESTCONF_VERSION in ('rfc8040'): format_args = ('{}', network, link, content) elif content == 'config': format_args = ('{}', content, network, link) @@ -518,7 +545,6 @@ def del_ietf_network_link_request(network: str, link: str, content: str): def add_oms_attr_request(link: str, oms_attr: str): url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}', - 'lighty': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}', 'draft-bierman02': '{}/config/ietf-network:networks/network/{}/ietf-network-topology:link/{}'} url2 = url[RESTCONF_VERSION] + '/org-openroadm-network-topology:OMS-attributes/span' network = 'openroadm-topology' @@ -528,7 +554,6 @@ def add_oms_attr_request(link: str, oms_attr: str): def del_oms_attr_request(link: str,): url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}', - 'lighty': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}', 'draft-bierman02': '{}/config/ietf-network:networks/network/{}/ietf-network-topology:link/{}'} url2 = url[RESTCONF_VERSION] + '/org-openroadm-network-topology:OMS-attributes/span' network = 'openroadm-topology' @@ -538,9 +563,8 @@ def del_oms_attr_request(link: str,): def get_ietf_network_node_request(network: str, node: str, content: str): url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}', - 'lighty': '{}/data/ietf-network:networks/network={}/node={}?content={}', 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'} - if RESTCONF_VERSION in ('rfc8040', 'lighty'): + if RESTCONF_VERSION in ('rfc8040'): format_args = ('{}', network, node, content) elif content == 'config': format_args = ('{}', content, network, node) @@ -550,7 +574,6 @@ def get_ietf_network_node_request(network: str, node: str, content: str): if bool(response): res = response.json() return_key = {'rfc8040': 'ietf-network:node', - 'lighty': 'ietf-network:node', 'draft-bierman02': 'node'} node = res[return_key[RESTCONF_VERSION]][0] else: @@ -561,9 +584,8 @@ def get_ietf_network_node_request(network: str, node: str, content: str): def del_ietf_network_node_request(network: str, node: str, content: str): url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}', - 'lighty': '{}/data/ietf-network:networks/network={}/node={}?content={}', 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'} - if RESTCONF_VERSION in ('rfc8040', 'lighty'): + if RESTCONF_VERSION in ('rfc8040'): format_args = ('{}', network, node, content) elif content == 'config': format_args = ('{}', content, network, node) @@ -580,12 +602,10 @@ def del_ietf_network_node_request(network: str, node: str, content: str): def get_ordm_serv_list_request(): url = {'rfc8040': '{}/data/org-openroadm-service:service-list?content=nonconfig', - 'lighty': '{}/data/org-openroadm-service:service-list?content=nonconfig', 'draft-bierman02': '{}/operational/org-openroadm-service:service-list/'} response = get_request(url[RESTCONF_VERSION]) res = response.json() return_key = {'rfc8040': 'org-openroadm-service:service-list', - 'lighty': 'org-openroadm-service:service-list', 'draft-bierman02': 'service-list'} if return_key[RESTCONF_VERSION] in res.keys(): response_attribute = res[return_key[RESTCONF_VERSION]] @@ -597,13 +617,11 @@ def get_ordm_serv_list_request(): def get_ordm_serv_list_attr_request(attribute: str, value: str): url = {'rfc8040': '{}/data/org-openroadm-service:service-list/{}={}?content=nonconfig', - 'lighty': '{}/data/org-openroadm-service:service-list/{}={}?content=nonconfig', 'draft-bierman02': '{}/operational/org-openroadm-service:service-list/{}/{}'} format_args = ('{}', attribute, value) response = get_request(url[RESTCONF_VERSION].format(*format_args)) res = response.json() return_key = {'rfc8040': 'org-openroadm-service:' + attribute, - 'lighty': 'org-openroadm-service:' + attribute, 'draft-bierman02': attribute} if return_key[RESTCONF_VERSION] in res.keys(): response_attribute = res[return_key[RESTCONF_VERSION]] @@ -615,12 +633,10 @@ def get_ordm_serv_list_attr_request(attribute: str, value: str): def get_serv_path_list_attr(attribute: str, value: str): url = {'rfc8040': '{}/data/transportpce-service-path:service-path-list/{}={}?content=nonconfig', - 'lighty': '{}/data/transportpce-service-path:service-path-list/{}={}?content=nonconfig', 'draft-bierman02': '{}/operational/transportpce-service-path:service-path-list/{}/{}'} response = get_request(url[RESTCONF_VERSION].format('{}', attribute, value)) res = response.json() return_key = {'rfc8040': 'transportpce-service-path:' + attribute, - 'lighty': 'transportpce-service-path:' + attribute, 'draft-bierman02': attribute} if return_key[RESTCONF_VERSION] in res.keys(): response_attribute = res[return_key[RESTCONF_VERSION]] @@ -664,7 +680,6 @@ def transportpce_api_rpc_request(api_module: str, rpc: str, payload: dict): else: res = response.json() return_key = {'rfc8040': api_module + ':output', - 'lighty': api_module + ':output', 'draft-bierman02': 'output'} if response.status_code == requests.codes.internal_server_error: return_output = res @@ -672,3 +687,47 @@ def transportpce_api_rpc_request(api_module: str, rpc: str, payload: dict): return_output = res[return_key[RESTCONF_VERSION]] return {'status_code': response.status_code, 'output': return_output} + +# +# simulators datastore operations +# + + +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'], + urllib.parse.quote(circuitpack, safe=''), + urllib.parse.quote(port, safe='')) + body = {"ports": [payload]} + response = requests.request("PUT", + url, + data=json.dumps(body), + headers=TYPE_APPLICATION_JSON, + 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 = '' + xml_body += dict2xml(body, indent=" ") + xml_body += '' + 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