tests/transportpce_tests/common/test_utils.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2021 Orange, Inc. and others.  All rights reserved.
   5 #
   6 # All rights reserved. This program and the accompanying materials
   7 # are made available under the terms of the Apache License, Version 2.0
   8 # which accompanies this distribution, and is available at
   9 # http://www.apache.org/licenses/LICENSE-2.0
  10 ##############################################################################
  11
  12 # pylint: disable=no-member
  13
  14 import json
  15 import os
  16 # pylint: disable=wrong-import-order
  17 import sys
  18 import re
  19 import signal
  20 import subprocess
  21 import time
  22
  23 import psutil
  24 import requests
  25 import urllib.parse
  26
  27 from dict2xml import dict2xml
  28 from netconf_client.connect import connect_ssh
  29 from netconf_client.ncclient import Manager
  30
  31
  32 # pylint: disable=import-error
  33 import simulators
  34
  35 SIMS = simulators.SIMS
  36
  37 HONEYNODE_OK_START_MSG = 'Netconf SSH endpoint started successfully at 0.0.0.0'
  38 LIGHTYNODE_OK_START_MSG = 'Data tree change listeners registered'
  39 KARAF_OK_START_MSG = "Transportpce controller started"
  40 LIGHTY_OK_START_MSG = re.escape("lighty.io and RESTCONF-NETCONF started")
  41
  42 ODL_LOGIN = 'admin'
  43 ODL_PWD = 'admin'
  44 NODES_LOGIN = 'admin'
  45 NODES_PWD = 'admin'
  46
  47 TYPE_APPLICATION_JSON = {'Content-Type': 'application/json', 'Accept': 'application/json'}
  48 TYPE_APPLICATION_XML = {'Content-Type': 'application/xml', 'Accept': 'application/xml'}
  49
  50 REQUEST_TIMEOUT = 10
  51
  52 CODE_SHOULD_BE_200 = 'Http status code should be 200'
  53 CODE_SHOULD_BE_201 = 'Http status code should be 201'
  54 T100GE = 'Transponder 100GE'
  55 T0_MULTILAYER_TOPO = 'T0 - Multi-layer topology'
  56 T0_FULL_MULTILAYER_TOPO = 'T0 - Full Multi-layer topology'
  57 T100GE_UUID = 'cf51c729-3699-308a-a7d0-594c6a62ebbb'
  58 T0_MULTILAYER_TOPO_UUID = '747c670e-7a07-3dab-b379-5b1cd17402a3'
  59 T0_FULL_MULTILAYER_TOPO_UUID = '393f09a4-0a0b-3d82-a4f6-1fbbc14ca1a7'
  60
  61 SIM_LOG_DIRECTORY = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'log')
  62
  63 process_list = []
  64
  65 if 'USE_ODL_ALT_RESTCONF_PORT' in os.environ:
  66     RESTCONF_PORT = os.environ['USE_ODL_ALT_RESTCONF_PORT']
  67 else:
  68     RESTCONF_PORT = 8181
  69
  70 RESTCONF_PATH_PREFIX = {'rfc8040': '/rests',
  71                         'draft-bierman02': '/restconf'}
  72
  73 if 'USE_ODL_RESTCONF_VERSION' in os.environ:
  74     RESTCONF_VERSION = os.environ['USE_ODL_RESTCONF_VERSION']
  75     if RESTCONF_VERSION not in RESTCONF_PATH_PREFIX:
  76         print('unsupported RESTCONF version ' + RESTCONF_VERSION)
  77         sys.exit(3)
  78 else:
  79     RESTCONF_VERSION = 'rfc8040'
  80
  81 RESTCONF_BASE_URL = 'http://localhost:' + str(RESTCONF_PORT) + RESTCONF_PATH_PREFIX[RESTCONF_VERSION]
  82
  83 if 'USE_ODL_ALT_KARAF_INSTALL_DIR' in os.environ:
  84     KARAF_INSTALLDIR = os.environ['USE_ODL_ALT_KARAF_INSTALL_DIR']
  85 else:
  86     KARAF_INSTALLDIR = 'karaf'
  87
  88 KARAF_LOG = os.path.join(
  89     os.path.dirname(os.path.realpath(__file__)),
  90     '..', '..', '..', KARAF_INSTALLDIR, 'target', 'assembly', 'data', 'log', 'karaf.log')
  91
  92 if 'USE_LIGHTY' in os.environ and os.environ['USE_LIGHTY'] == 'True':
  93     TPCE_LOG = 'odl-' + str(os.getpid()) + '.log'
  94 else:
  95     TPCE_LOG = KARAF_LOG
  96
  97 if 'USE_SIMS' not in os.environ:
  98     SIMS_TO_USE = 'lightynode'
  99     SIMS_TYPE = 'lightynode'
 100 else:
 101     SIMS_TO_USE = os.environ['USE_SIMS']
 102     print("Forcing to use SIMS " + SIMS_TO_USE)
 103     if SIMS_TO_USE != 'None' or 'SIMS_TYPE' not in os.environ:
 104         SIMS_TYPE = SIMS_TO_USE
 105     else:
 106         SIMS_TYPE = os.environ['SIMS_TYPE']
 107         print("Forcing to use SIMS type" + SIMS_TYPE)
 108
 109
 110 #
 111 # Basic HTTP operations
 112 #
 113
 114
 115 def get_request(url):
 116     return requests.request(
 117         'GET', url.format(RESTCONF_BASE_URL),
 118         headers=TYPE_APPLICATION_JSON,
 119         auth=(ODL_LOGIN, ODL_PWD),
 120         timeout=REQUEST_TIMEOUT)
 121
 122
 123 def put_request(url, data):
 124     return requests.request(
 125         'PUT', url.format(RESTCONF_BASE_URL),
 126         data=json.dumps(data),
 127         headers=TYPE_APPLICATION_JSON,
 128         auth=(ODL_LOGIN, ODL_PWD),
 129         timeout=REQUEST_TIMEOUT)
 130
 131
 132 def delete_request(url):
 133     return requests.request(
 134         'DELETE', url.format(RESTCONF_BASE_URL),
 135         headers=TYPE_APPLICATION_JSON,
 136         auth=(ODL_LOGIN, ODL_PWD),
 137         timeout=REQUEST_TIMEOUT)
 138
 139
 140 def post_request(url, data):
 141     if data:
 142         return requests.request(
 143             "POST", url.format(RESTCONF_BASE_URL),
 144             data=json.dumps(data),
 145             headers=TYPE_APPLICATION_JSON,
 146             auth=(ODL_LOGIN, ODL_PWD),
 147             timeout=REQUEST_TIMEOUT)
 148     return requests.request(
 149         "POST", url.format(RESTCONF_BASE_URL),
 150         headers=TYPE_APPLICATION_JSON,
 151         auth=(ODL_LOGIN, ODL_PWD),
 152         timeout=REQUEST_TIMEOUT)
 153
 154 #
 155 # Process management
 156 #
 157
 158
 159 def start_honeynode(log_file: str, sim):
 160     executable = os.path.join(os.path.dirname(os.path.realpath(__file__)),
 161                               '..', '..', 'honeynode', sim[1], 'honeynode-simulator', 'honeycomb-tpce')
 162     sample_directory = os.path.join(os.path.dirname(os.path.realpath(__file__)),
 163                                     '..', '..', 'sample_configs', 'openroadm', sim[1])
 164     if os.path.isfile(executable):
 165         with open(log_file, 'w', encoding='utf-8') as outfile:
 166             return subprocess.Popen(
 167                 [executable, SIMS[sim]['port'], os.path.join(sample_directory, SIMS[sim]['configfile'])],
 168                 stdout=outfile, stderr=outfile)
 169     return None
 170
 171
 172 def start_lightynode(log_file: str, sim):
 173     executable = os.path.join(os.path.dirname(os.path.realpath(__file__)),
 174                               '..', '..', 'lightynode', 'lightynode-openroadm-device', 'start-device.sh')
 175     sample_directory = os.path.join(os.path.dirname(os.path.realpath(__file__)),
 176                                     '..', '..', 'sample_configs', 'openroadm', sim[1])
 177     if os.path.isfile(executable):
 178         with open(log_file, 'w', encoding='utf-8') as outfile:
 179             return subprocess.Popen(
 180                 [executable, "-v" + sim[1], "-p" + SIMS[sim]['port'], "-f" + os.path.join(sample_directory,
 181                                                                                           SIMS[sim]['configfile'])],
 182                 stdout=outfile, stderr=outfile)
 183     return None
 184
 185
 186 def start_sims(sims_list):
 187     if SIMS_TO_USE == 'None':
 188         return None
 189     if SIMS_TO_USE == 'honeynode':
 190         start_msg = HONEYNODE_OK_START_MSG
 191         start_method = start_honeynode
 192     else:
 193         start_msg = LIGHTYNODE_OK_START_MSG
 194         start_method = start_lightynode
 195     for sim in sims_list:
 196         print('starting simulator ' + sim[0] + ' in OpenROADM device version ' + sim[1] + '...')
 197         log_file = os.path.join(SIM_LOG_DIRECTORY, SIMS[sim]['logfile'])
 198         process = start_method(log_file, sim)
 199         if wait_until_log_contains(log_file, start_msg, 100):
 200             print('simulator for ' + sim[0] + ' started')
 201         else:
 202             print('simulator for ' + sim[0] + ' failed to start')
 203             shutdown_process(process)
 204             for pid in process_list:
 205                 shutdown_process(pid)
 206             sys.exit(3)
 207         process_list.append(process)
 208     return process_list
 209
 210
 211 def start_tpce():
 212     if 'NO_ODL_STARTUP' in os.environ:
 213         print('No OpenDaylight instance to start!')
 214         return []
 215     print('starting OpenDaylight...')
 216     if 'USE_LIGHTY' in os.environ and os.environ['USE_LIGHTY'] == 'True':
 217         process = start_lighty()
 218         start_msg = LIGHTY_OK_START_MSG
 219     else:
 220         process = start_karaf()
 221         start_msg = KARAF_OK_START_MSG
 222     if wait_until_log_contains(TPCE_LOG, start_msg, time_to_wait=100):
 223         print('OpenDaylight started !')
 224     else:
 225         print('OpenDaylight failed to start !')
 226         shutdown_process(process)
 227         for pid in process_list:
 228             shutdown_process(pid)
 229         sys.exit(1)
 230     process_list.append(process)
 231     return process_list
 232
 233
 234 def start_karaf():
 235     print('starting KARAF TransportPCE build...')
 236     executable = os.path.join(
 237         os.path.dirname(os.path.realpath(__file__)),
 238         '..', '..', '..', KARAF_INSTALLDIR, 'target', 'assembly', 'bin', 'karaf')
 239     with open('odl.log', 'w', encoding='utf-8') as outfile:
 240         return subprocess.Popen(
 241             ['sh', executable, 'server'], stdout=outfile, stderr=outfile, stdin=None)
 242
 243
 244 def start_lighty():
 245     print('starting LIGHTY.IO TransportPCE build...')
 246     executable = os.path.join(
 247         os.path.dirname(os.path.realpath(__file__)),
 248         '..', '..', '..', 'lighty', 'target', 'tpce',
 249         'clean-start-controller.sh')
 250     with open(TPCE_LOG, 'w', encoding='utf-8') as outfile:
 251         return subprocess.Popen(
 252             ['sh', executable], stdout=outfile, stderr=outfile, stdin=None)
 253
 254
 255 def install_karaf_feature(feature_name: str):
 256     print('installing feature ' + feature_name)
 257     executable = os.path.join(
 258         os.path.dirname(os.path.realpath(__file__)),
 259         '..', '..', '..', KARAF_INSTALLDIR, 'target', 'assembly', 'bin', 'client')
 260 # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-701
 261 # -b option needed below because of Karaf client bug reporte in the JIRA ticket mentioned above
 262     return subprocess.run([executable, '-b'],
 263                           input='feature:install ' + feature_name + '\n feature:list | grep '
 264                           + feature_name + ' \n logout \n',
 265                           universal_newlines=True, check=False)
 266
 267
 268 def shutdown_process(process):
 269     if process is not None:
 270         for child in psutil.Process(process.pid).children():
 271             child.send_signal(signal.SIGINT)
 272             child.wait()
 273         process.send_signal(signal.SIGINT)
 274
 275
 276 def wait_until_log_contains(log_file, regexp, time_to_wait=60):
 277     # pylint: disable=lost-exception
 278     # pylint: disable=consider-using-with
 279     stringfound = False
 280     line = None
 281     try:
 282         with TimeOut(seconds=time_to_wait):
 283             while not os.path.exists(log_file):
 284                 time.sleep(0.2)
 285             with open(log_file, 'r', encoding='utf-8') as filelogs:
 286                 filelogs.seek(0, 2)
 287                 print("Searching for pattern '" + regexp + "' in " + os.path.basename(log_file), end='... ', flush=True)
 288                 compiled_regexp = re.compile(regexp)
 289                 while True:
 290                     line = filelogs.readline()
 291                     if compiled_regexp.search(line):
 292                         print('Pattern found!', end=' ')
 293                         stringfound = True
 294                         break
 295                     if not line:
 296                         time.sleep(0.1)
 297         return stringfound
 298     except TimeoutError:
 299         print('Pattern not found after ' + str(time_to_wait), end=' seconds! ', flush=True)
 300         return stringfound
 301     except PermissionError:
 302         print('Permission Error when trying to access the log file', end=' ... ', flush=True)
 303         return stringfound
 304
 305
 306 class TimeOut:
 307     def __init__(self, seconds=1, error_message='Timeout'):
 308         self.seconds = seconds
 309         self.error_message = error_message
 310
 311     def handle_timeout(self, signum, frame):
 312         raise TimeoutError(self.error_message)
 313
 314     def __enter__(self):
 315         signal.signal(signal.SIGALRM, self.handle_timeout)
 316         signal.alarm(self.seconds)
 317
 318     def __exit__(self, type, value, traceback):
 319         # pylint: disable=W0622
 320         signal.alarm(0)
 321
 322 #
 323 # Basic NetCONF device operations
 324 #
 325
 326
 327 def mount_device(node: str, sim: str):
 328     url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}',
 329            'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}'}
 330     body = {'node': [{
 331         'node-id': node,
 332         'netconf-node-topology:username': NODES_LOGIN,
 333         'netconf-node-topology:password': NODES_PWD,
 334         'netconf-node-topology:host': '127.0.0.1',
 335         'netconf-node-topology:port': SIMS[sim]['port'],
 336         'netconf-node-topology:tcp-only': 'false',
 337         'netconf-node-topology:pass-through': {}}]}
 338     response = put_request(url[RESTCONF_VERSION].format('{}', node), body)
 339     if wait_until_log_contains(TPCE_LOG, 'Triggering notification stream NETCONF for node ' + node, 180):
 340         print('Node ' + node + ' correctly added to tpce topology', end='... ', flush=True)
 341     else:
 342         print('Node ' + node + ' still not added to tpce topology', end='... ', flush=True)
 343         if response.status_code == requests.codes.ok:
 344             print('It was probably loaded at start-up', end='... ', flush=True)
 345         # TODO an else-clause to abort test would probably be nice here
 346     return response
 347
 348
 349 def unmount_device(node: str):
 350     url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}',
 351            'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}'}
 352     response = delete_request(url[RESTCONF_VERSION].format('{}', node))
 353     if wait_until_log_contains(TPCE_LOG, re.escape("onDeviceDisConnected: " + node), 180):
 354         print('Node ' + node + ' correctly deleted from tpce topology', end='... ', flush=True)
 355     else:
 356         print('Node ' + node + ' still not deleted from tpce topology', end='... ', flush=True)
 357     return response
 358
 359
 360 def check_device_connection(node: str):
 361     url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}?content=nonconfig',
 362            'draft-bierman02': '{}/operational/network-topology:network-topology/topology/topology-netconf/node/{}'}
 363     response = get_request(url[RESTCONF_VERSION].format('{}', node))
 364     res = response.json()
 365     return_key = {'rfc8040': 'network-topology:node',
 366                   'draft-bierman02': 'node'}
 367     if return_key[RESTCONF_VERSION] in res.keys():
 368         connection_status = res[return_key[RESTCONF_VERSION]][0]['netconf-node-topology:connection-status']
 369     else:
 370         connection_status = res['errors']['error'][0]
 371     return {'status_code': response.status_code,
 372             'connection-status': connection_status}
 373
 374
 375 def check_node_request(node: str):
 376     # pylint: disable=line-too-long
 377     url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device?content=config',  # nopep8
 378            'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device'}  # nopep8
 379     response = get_request(url[RESTCONF_VERSION].format('{}', node))
 380     res = response.json()
 381     return_key = {'rfc8040': 'org-openroadm-device:org-openroadm-device',
 382                   'draft-bierman02': 'org-openroadm-device'}
 383     if return_key[RESTCONF_VERSION] in res.keys():
 384         response_attribute = res[return_key[RESTCONF_VERSION]]
 385     else:
 386         response_attribute = res['errors']['error'][0]
 387     return {'status_code': response.status_code,
 388             'org-openroadm-device': response_attribute}
 389
 390
 391 def check_node_attribute_request(node: str, attribute: str, attribute_value: str):
 392     # pylint: disable=line-too-long
 393     url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}={}?content=nonconfig',  # nopep8
 394            'draft-bierman02': '{}/operational/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}/{}'}  # nopep8
 395     response = get_request(url[RESTCONF_VERSION].format('{}', node, attribute, attribute_value))
 396     res = response.json()
 397     return_key = {'rfc8040': 'org-openroadm-device:' + attribute,
 398                   'draft-bierman02': attribute}
 399     if return_key[RESTCONF_VERSION] in res.keys():
 400         response_attribute = res[return_key[RESTCONF_VERSION]]
 401     elif 'errors' in res.keys():
 402         response_attribute = res['errors']['error'][0]
 403     else:
 404         # status code 400 invalid request
 405         response_attribute = res['message'] + ' ' + res['url']
 406         print(response_attribute)
 407     return {'status_code': response.status_code,
 408             attribute: response_attribute}
 409
 410
 411 def check_node_attribute2_request(node: str, attribute: str, attribute_value: str, attribute2: str):
 412     # pylint: disable=line-too-long
 413     url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}={}/{}?content=config',  # nopep8
 414            'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}/{}/{}'}  # nopep8
 415     response = get_request(url[RESTCONF_VERSION].format('{}', node, attribute, attribute_value, attribute2))
 416     res = response.json()
 417     if attribute2 in res.keys():
 418         response_attribute = res[attribute2]
 419     else:
 420         response_attribute = res['errors']['error'][0]
 421     return {'status_code': response.status_code,
 422             attribute2: response_attribute}
 423
 424
 425 def del_node_attribute_request(node: str, attribute: str, attribute_value: str):
 426     # pylint: disable=line-too-long
 427     url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}={}',  # nopep8
 428            'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/{}/{}'}  # nopep8
 429     response = delete_request(url[RESTCONF_VERSION].format('{}', node, attribute, attribute_value))
 430     return response
 431
 432 #
 433 # Portmapping operations
 434 #
 435
 436
 437 def post_portmapping(payload: str):
 438     url = {'rfc8040': '{}/data/transportpce-portmapping:network',
 439            'draft-bierman02': '{}/config/transportpce-portmapping:network'}
 440     json_payload = json.loads(payload)
 441     response = post_request(url[RESTCONF_VERSION].format('{}'), json_payload)
 442     return {'status_code': response.status_code}
 443
 444
 445 def del_portmapping():
 446     url = {'rfc8040': '{}/data/transportpce-portmapping:network',
 447            'draft-bierman02': '{}/config/transportpce-portmapping:network'}
 448     response = delete_request(url[RESTCONF_VERSION].format('{}'))
 449     return {'status_code': response.status_code}
 450
 451
 452 def get_portmapping_node_attr(node: str, attr: str, value: str):
 453     # pylint: disable=consider-using-f-string
 454     url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}',
 455            'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}'}
 456     target_url = url[RESTCONF_VERSION].format('{}', node)
 457     if attr is not None:
 458         target_url = (target_url + '/{}').format('{}', attr)
 459         if value is not None:
 460             suffix = {'rfc8040': '={}', 'draft-bierman02': '/{}'}
 461             target_url = (target_url + suffix[RESTCONF_VERSION]).format('{}', value)
 462     else:
 463         attr = 'nodes'
 464     response = get_request(target_url)
 465     res = response.json()
 466     return_key = {'rfc8040': 'transportpce-portmapping:' + attr,
 467                   'draft-bierman02': attr}
 468     if return_key[RESTCONF_VERSION] in res.keys():
 469         return_output = res[return_key[RESTCONF_VERSION]]
 470     else:
 471         return_output = res['errors']['error'][0]
 472     return {'status_code': response.status_code,
 473             attr: return_output}
 474
 475 #
 476 # Topology operations
 477 #
 478
 479
 480 def get_ietf_network_request(network: str, content: str):
 481     url = {'rfc8040': '{}/data/ietf-network:networks/network={}?content={}',
 482            'draft-bierman02': '{}/{}/ietf-network:networks/network/{}'}
 483     if RESTCONF_VERSION in ('rfc8040'):
 484         format_args = ('{}', network, content)
 485     elif content == 'config':
 486         format_args = ('{}', content, network)
 487     else:
 488         format_args = ('{}', 'operational', network)
 489     response = get_request(url[RESTCONF_VERSION].format(*format_args))
 490     if bool(response):
 491         res = response.json()
 492         return_key = {'rfc8040': 'ietf-network:network',
 493                       'draft-bierman02': 'network'}
 494         networks = res[return_key[RESTCONF_VERSION]]
 495     else:
 496         networks = None
 497     return {'status_code': response.status_code,
 498             'network': networks}
 499
 500
 501 def put_ietf_network(network: str, payload: str):
 502     url = {'rfc8040': '{}/data/ietf-network:networks/network={}',
 503            'draft-bierman02': '{}/config/ietf-network:networks/network/{}'}
 504     json_payload = json.loads(payload)
 505     response = put_request(url[RESTCONF_VERSION].format('{}', network), json_payload)
 506     return {'status_code': response.status_code}
 507
 508
 509 def del_ietf_network(network: str):
 510     url = {'rfc8040': '{}/data/ietf-network:networks/network={}',
 511            'draft-bierman02': '{}/config/ietf-network:networks/network/{}'}
 512     response = delete_request(url[RESTCONF_VERSION].format('{}', network))
 513     return {'status_code': response.status_code}
 514
 515
 516 def get_ietf_network_link_request(network: str, link: str, content: str):
 517     url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}?content={}',
 518            'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/ietf-network-topology:link/{}'}
 519     if RESTCONF_VERSION in ('rfc8040'):
 520         format_args = ('{}', network, link, content)
 521     elif content == 'config':
 522         format_args = ('{}', content, network, link)
 523     else:
 524         format_args = ('{}', 'operational', network, link)
 525     response = get_request(url[RESTCONF_VERSION].format(*format_args))
 526     res = response.json()
 527     return_key = {'rfc8040': 'ietf-network-topology:link',
 528                   'draft-bierman02': 'ietf-network-topology:link'}
 529     link = res[return_key[RESTCONF_VERSION]][0]
 530     return {'status_code': response.status_code,
 531             'link': link}
 532
 533
 534 def del_ietf_network_link_request(network: str, link: str, content: str):
 535     url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}?content={}',
 536            'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/ietf-network-topology:link/{}'}
 537     if RESTCONF_VERSION in ('rfc8040'):
 538         format_args = ('{}', network, link, content)
 539     elif content == 'config':
 540         format_args = ('{}', content, network, link)
 541     else:
 542         format_args = ('{}', 'operational', network, link)
 543     response = delete_request(url[RESTCONF_VERSION].format(*format_args))
 544     return response
 545
 546
 547 def add_oms_attr_request(link: str, oms_attr: str):
 548     url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}',
 549            'draft-bierman02': '{}/config/ietf-network:networks/network/{}/ietf-network-topology:link/{}'}
 550     url2 = url[RESTCONF_VERSION] + '/org-openroadm-network-topology:OMS-attributes/span'
 551     network = 'openroadm-topology'
 552     response = put_request(url2.format('{}', network, link), oms_attr)
 553     return response
 554
 555
 556 def del_oms_attr_request(link: str,):
 557     url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}',
 558            'draft-bierman02': '{}/config/ietf-network:networks/network/{}/ietf-network-topology:link/{}'}
 559     url2 = url[RESTCONF_VERSION] + '/org-openroadm-network-topology:OMS-attributes/span'
 560     network = 'openroadm-topology'
 561     response = delete_request(url2.format('{}', network, link))
 562     return response
 563
 564
 565 def get_ietf_network_node_request(network: str, node: str, content: str):
 566     url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}',
 567            'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'}
 568     if RESTCONF_VERSION in ('rfc8040'):
 569         format_args = ('{}', network, node, content)
 570     elif content == 'config':
 571         format_args = ('{}', content, network, node)
 572     else:
 573         format_args = ('{}', 'operational', network, node)
 574     response = get_request(url[RESTCONF_VERSION].format(*format_args))
 575     if bool(response):
 576         res = response.json()
 577         return_key = {'rfc8040': 'ietf-network:node',
 578                       'draft-bierman02': 'node'}
 579         node = res[return_key[RESTCONF_VERSION]][0]
 580     else:
 581         node = None
 582     return {'status_code': response.status_code,
 583             'node': node}
 584
 585
 586 def del_ietf_network_node_request(network: str, node: str, content: str):
 587     url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}',
 588            'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'}
 589     if RESTCONF_VERSION in ('rfc8040'):
 590         format_args = ('{}', network, node, content)
 591     elif content == 'config':
 592         format_args = ('{}', content, network, node)
 593     else:
 594         format_args = ('{}', 'operational', network, node)
 595     response = delete_request(url[RESTCONF_VERSION].format(*format_args))
 596     return response
 597
 598
 599 #
 600 # Service list operations
 601 #
 602
 603
 604 def get_ordm_serv_list_request():
 605     url = {'rfc8040': '{}/data/org-openroadm-service:service-list?content=nonconfig',
 606            'draft-bierman02': '{}/operational/org-openroadm-service:service-list/'}
 607     response = get_request(url[RESTCONF_VERSION])
 608     res = response.json()
 609     return_key = {'rfc8040': 'org-openroadm-service:service-list',
 610                   'draft-bierman02': 'service-list'}
 611     if return_key[RESTCONF_VERSION] in res.keys():
 612         response_attribute = res[return_key[RESTCONF_VERSION]]
 613     else:
 614         response_attribute = res['errors']['error'][0]
 615     return {'status_code': response.status_code,
 616             'service-list': response_attribute}
 617
 618
 619 def get_ordm_serv_list_attr_request(attribute: str, value: str):
 620     url = {'rfc8040': '{}/data/org-openroadm-service:service-list/{}={}?content=nonconfig',
 621            'draft-bierman02': '{}/operational/org-openroadm-service:service-list/{}/{}'}
 622     format_args = ('{}', attribute, value)
 623     response = get_request(url[RESTCONF_VERSION].format(*format_args))
 624     res = response.json()
 625     return_key = {'rfc8040': 'org-openroadm-service:' + attribute,
 626                   'draft-bierman02': attribute}
 627     if return_key[RESTCONF_VERSION] in res.keys():
 628         response_attribute = res[return_key[RESTCONF_VERSION]]
 629     else:
 630         response_attribute = res['errors']['error'][0]
 631     return {'status_code': response.status_code,
 632             attribute: response_attribute}
 633
 634
 635 def get_serv_path_list_attr(attribute: str, value: str):
 636     url = {'rfc8040': '{}/data/transportpce-service-path:service-path-list/{}={}?content=nonconfig',
 637            'draft-bierman02': '{}/operational/transportpce-service-path:service-path-list/{}/{}'}
 638     response = get_request(url[RESTCONF_VERSION].format('{}', attribute, value))
 639     res = response.json()
 640     return_key = {'rfc8040': 'transportpce-service-path:' + attribute,
 641                   'draft-bierman02': attribute}
 642     if return_key[RESTCONF_VERSION] in res.keys():
 643         response_attribute = res[return_key[RESTCONF_VERSION]]
 644     else:
 645         response_attribute = res['errors']['error'][0]
 646     return {'status_code': response.status_code,
 647             attribute: response_attribute}
 648
 649
 650 #
 651 # TransportPCE internal API RPCs
 652 #
 653
 654
 655 def prepend_dict_keys(input_dict: dict, prefix: str):
 656     return_dict = {}
 657     for key, value in input_dict.items():
 658         newkey = prefix + key
 659         if isinstance(value, dict):
 660             return_dict[newkey] = prepend_dict_keys(value, prefix)
 661             # TODO: perhaps some recursion depth limit or another solution has to be considered here
 662             # even if recursion depth is given by the input_dict argument
 663             # direct (self-)recursive functions may carry unwanted side-effects such as ressource consumptions
 664         else:
 665             return_dict[newkey] = value
 666     return return_dict
 667
 668
 669 def transportpce_api_rpc_request(api_module: str, rpc: str, payload: dict):
 670     # pylint: disable=consider-using-f-string
 671     url = "{}/operations/{}:{}".format('{}', api_module, rpc)
 672     if payload is None:
 673         data = None
 674     elif RESTCONF_VERSION == 'draft-bierman02':
 675         data = prepend_dict_keys({'input': payload}, api_module + ':')
 676     else:
 677         data = {'input': payload}
 678     response = post_request(url, data)
 679     if response.status_code == requests.codes.no_content:
 680         return_output = None
 681     else:
 682         res = response.json()
 683         return_key = {'rfc8040': api_module + ':output',
 684                       'draft-bierman02': 'output'}
 685         if response.status_code == requests.codes.internal_server_error:
 686             return_output = res
 687         else:
 688             return_output = res[return_key[RESTCONF_VERSION]]
 689     return {'status_code': response.status_code,
 690             'output': return_output}
 691
 692 #
 693 # simulators datastore operations
 694 #
 695
 696
 697 def sims_update_cp_port(sim: tuple, circuitpack: str, port: str, payload: dict):
 698     if SIMS_TYPE == 'lightynode':
 699         return sims_update_cp_port_ntcf(sim, circuitpack, payload)
 700     if SIMS_TYPE == 'honeynode':
 701         return sims_update_cp_port_rest(sim, circuitpack, port, payload)
 702     return False
 703
 704
 705 def sims_update_cp_port_rest(sim: tuple, circuitpack: str, port: str, payload: dict):
 706     # pylint: disable=consider-using-f-string
 707     url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/{}/ports/{}".format(
 708         SIMS[sim]['restconf_baseurl'],
 709         urllib.parse.quote(circuitpack, safe=''),
 710         urllib.parse.quote(port, safe=''))
 711     body = {"ports": [payload]}
 712     response = requests.request("PUT",
 713                                 url,
 714                                 data=json.dumps(body),
 715                                 headers=TYPE_APPLICATION_JSON,
 716                                 auth=(ODL_LOGIN, ODL_PWD),
 717                                 timeout=REQUEST_TIMEOUT)
 718     return response.status_code == requests.codes.ok
 719
 720
 721 def sims_update_cp_port_ntcf(sim: tuple, circuitpack: str, payload: dict):
 722     body = {"circuit-packs": {"circuit-pack-name": circuitpack, "ports": payload}}
 723     xml_body = '<config><org-openroadm-device xmlns="http://org/openroadm/device">'
 724     xml_body += dict2xml(body, indent="  ")
 725     xml_body += '</org-openroadm-device></config>'
 726     with connect_ssh(host='127.0.0.1',
 727                      port=int(SIMS[sim]['port']),
 728                      username=NODES_LOGIN,
 729                      password=NODES_PWD) as session:
 730         mgr = Manager(session, timeout=120)
 731         reply = mgr.edit_config(xml_body, target="candidate", default_operation="merge")
 732     if "None" in str(reply):
 733         return True
 734     return False
 735
 736
 737 def sims_update_pm_interact(sim: tuple, payload: dict):
 738     if SIMS_TYPE == 'lightynode':
 739         return sims_update_pm_interact_ntcf(sim, payload)
 740     if SIMS_TYPE == 'honeynode':
 741         return sims_update_pm_interact_rest(sim, payload)
 742     return False
 743
 744
 745 def sims_update_pm_interact_rest(sim: tuple, payload: dict):
 746     # pylint: disable=consider-using-f-string
 747     url = "{}/operations/pm-handling:pm-interact".format(SIMS[sim]['restconf_baseurl'])
 748     body = {"input": payload}
 749     response = requests.request("POST",
 750                                 url,
 751                                 data=json.dumps(body),
 752                                 headers=TYPE_APPLICATION_JSON,
 753                                 auth=(ODL_LOGIN, ODL_PWD),
 754                                 timeout=REQUEST_TIMEOUT)
 755     return response.status_code == requests.codes.ok
 756
 757
 758 def sims_update_pm_interact_ntcf(sim: tuple, payload: dict):
 759     # pylint: disable=line-too-long
 760     xml_body = '<pm-interact xmlns="http://honeynode-simulator/pm-handling">'
 761     xml_body += dict2xml(payload, indent="  ")
 762     xml_body += '</pm-interact>'
 763     new_xml = xml_body.replace("<pm-resource-instance>/org-openroadm-device:org-openroadm-device/org-openroadm-device:interface[org-openroadm-device:name='OTS-DEG2-TTP-TXRX']</pm-resource-instance>",
 764                                "<pm-resource-instance xmlns:a=\"http://org/openroadm/device\">/a:org-openroadm-device/a:interface[a:name='OTS-DEG2-TTP-TXRX']</pm-resource-instance>")
 765     with connect_ssh(host='127.0.0.1',
 766                      port=int(SIMS[sim]['port']),
 767                      username=NODES_LOGIN,
 768                      password=NODES_PWD) as session:
 769         mgr = Manager(session, timeout=120)
 770         reply = mgr.dispatch(new_xml)
 771     if "netconf_client.ncclient.RPCReply" in str(reply):
 772         return True
 773     return False