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