+
+#
+# Basic NetCONF device operations
+#
+
+
+def mount_device(node: str, sim: str):
+ url = {'rfc8040': '{}/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,
+ 'netconf-node-topology:username': NODES_LOGIN,
+ 'netconf-node-topology:password': NODES_PWD,
+ 'netconf-node-topology:host': '127.0.0.1',
+ 'netconf-node-topology:port': SIMS[sim]['port'],
+ 'netconf-node-topology:tcp-only': 'false',
+ 'netconf-node-topology:pass-through': {}}]}
+ response = put_request(url[RESTCONF_VERSION].format('{}', node), body)
+ if wait_until_log_contains(TPCE_LOG, 'Triggering notification stream NETCONF for node ' + node, 180):
+ print('Node ' + node + ' correctly added to tpce topology', end='... ', flush=True)
+ else:
+ print('Node ' + node + ' still not added to tpce topology', end='... ', flush=True)
+ if response.status_code == requests.codes.ok:
+ print('It was probably loaded at start-up', end='... ', flush=True)
+ # TODO an else-clause to abort test would probably be nice here
+ return response
+
+
+def unmount_device(node: str):
+ url = {'rfc8040': '{}/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):
+ print('Node ' + node + ' correctly deleted from tpce topology', end='... ', flush=True)
+ else:
+ print('Node ' + node + ' still not deleted from tpce topology', end='... ', flush=True)
+ return response
+
+
+def check_device_connection(node: str):
+ url = {'rfc8040': '{}/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',
+ 'draft-bierman02': 'node'}
+ if return_key[RESTCONF_VERSION] in res.keys():
+ connection_status = res[return_key[RESTCONF_VERSION]][0]['netconf-node-topology:connection-status']
+ else:
+ connection_status = res['errors']['error'][0]
+ return {'status_code': response.status_code,
+ 'connection-status': connection_status}
+
+
+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
+ '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',
+ 'draft-bierman02': 'org-openroadm-device'}
+ if return_key[RESTCONF_VERSION] in res.keys():
+ response_attribute = res[return_key[RESTCONF_VERSION]]
+ else:
+ response_attribute = res['errors']['error'][0]
+ return {'status_code': response.status_code,
+ 'org-openroadm-device': response_attribute}
+
+
+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
+ '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,
+ 'draft-bierman02': attribute}
+ if return_key[RESTCONF_VERSION] in res.keys():
+ response_attribute = res[return_key[RESTCONF_VERSION]]
+ else:
+ response_attribute = res['errors']['error'][0]
+ return {'status_code': response.status_code,
+ attribute: response_attribute}
+
+
+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
+ '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()
+ if attribute2 in res.keys():
+ response_attribute = res[attribute2]
+ else:
+ response_attribute = res['errors']['error'][0]
+ return {'status_code': response.status_code,
+ attribute2: response_attribute}
+
+
+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
+ '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
+
+#
+# Portmapping operations
+#
+
+
+def post_portmapping(payload: str):
+ url = {'rfc8040': '{}/data/transportpce-portmapping:network',
+ 'draft-bierman02': '{}/config/transportpce-portmapping:network'}
+ json_payload = json.loads(payload)
+ response = post_request(url[RESTCONF_VERSION].format('{}'), json_payload)
+ return {'status_code': response.status_code}
+
+
+def del_portmapping():
+ url = {'rfc8040': '{}/data/transportpce-portmapping:network',
+ 'draft-bierman02': '{}/config/transportpce-portmapping:network'}
+ response = delete_request(url[RESTCONF_VERSION].format('{}'))
+ return {'status_code': response.status_code}
+
+
+def get_portmapping_node_attr(node: str, attr: str, value: str):
+ # pylint: disable=consider-using-f-string
+ url = {'rfc8040': '{}/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': '={}', '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,
+ 'draft-bierman02': attr}
+ if return_key[RESTCONF_VERSION] in res.keys():
+ return_output = res[return_key[RESTCONF_VERSION]]
+ else:
+ return_output = res['errors']['error'][0]
+ return {'status_code': response.status_code,
+ attr: return_output}
+
+#
+# Topology operations
+#
+
+
+def get_ietf_network_request(network: str, content: str):
+ url = {'rfc8040': '{}/data/ietf-network:networks/network={}?content={}',
+ 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}'}
+ if RESTCONF_VERSION in ('rfc8040'):
+ format_args = ('{}', network, content)
+ elif content == 'config':
+ format_args = ('{}', content, network)
+ else:
+ format_args = ('{}', 'operational', network)
+ response = get_request(url[RESTCONF_VERSION].format(*format_args))
+ if bool(response):
+ res = response.json()
+ return_key = {'rfc8040': 'ietf-network:network',
+ 'draft-bierman02': 'network'}
+ networks = res[return_key[RESTCONF_VERSION]]
+ else:
+ networks = None
+ return {'status_code': response.status_code,
+ 'network': networks}
+
+
+def put_ietf_network(network: str, payload: str):
+ url = {'rfc8040': '{}/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)
+ return {'status_code': response.status_code}
+
+
+def del_ietf_network(network: str):
+ url = {'rfc8040': '{}/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}
+
+
+def get_ietf_network_link_request(network: str, link: str, content: str):
+ url = {'rfc8040': '{}/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'):
+ format_args = ('{}', network, link, content)
+ elif content == 'config':
+ format_args = ('{}', content, network, link)
+ else:
+ format_args = ('{}', 'operational', network, link)
+ response = get_request(url[RESTCONF_VERSION].format(*format_args))
+ res = response.json()
+ return_key = {'rfc8040': 'ietf-network-topology:link',
+ 'draft-bierman02': 'ietf-network-topology:link'}
+ link = res[return_key[RESTCONF_VERSION]][0]
+ return {'status_code': response.status_code,
+ 'link': link}
+
+
+def del_ietf_network_link_request(network: str, link: str, content: str):
+ url = {'rfc8040': '{}/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'):
+ format_args = ('{}', network, link, content)
+ elif content == 'config':
+ format_args = ('{}', content, network, link)
+ else:
+ format_args = ('{}', 'operational', network, link)
+ response = delete_request(url[RESTCONF_VERSION].format(*format_args))
+ return response
+
+
+def add_oms_attr_request(link: str, oms_attr: str):
+ url = {'rfc8040': '{}/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'
+ response = put_request(url2.format('{}', network, link), oms_attr)
+ return response
+
+
+def del_oms_attr_request(link: str,):
+ url = {'rfc8040': '{}/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'
+ response = delete_request(url2.format('{}', network, link))
+ return response
+
+
+def get_ietf_network_node_request(network: str, node: str, content: str):
+ url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}',
+ 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'}
+ if RESTCONF_VERSION in ('rfc8040'):
+ format_args = ('{}', network, node, content)
+ elif content == 'config':
+ format_args = ('{}', content, network, node)
+ else:
+ format_args = ('{}', 'operational', network, node)
+ response = get_request(url[RESTCONF_VERSION].format(*format_args))
+ if bool(response):
+ res = response.json()
+ return_key = {'rfc8040': 'ietf-network:node',
+ 'draft-bierman02': 'node'}
+ node = res[return_key[RESTCONF_VERSION]][0]
+ else:
+ node = None
+ return {'status_code': response.status_code,
+ 'node': node}
+
+
+def del_ietf_network_node_request(network: str, node: str, content: str):
+ url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}',
+ 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'}
+ if RESTCONF_VERSION in ('rfc8040'):
+ format_args = ('{}', network, node, content)
+ elif content == 'config':
+ format_args = ('{}', content, network, node)
+ else:
+ format_args = ('{}', 'operational', network, node)
+ response = delete_request(url[RESTCONF_VERSION].format(*format_args))
+ return response
+
+
+#
+# Service list operations
+#
+
+
+def get_ordm_serv_list_request():
+ url = {'rfc8040': '{}/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',
+ 'draft-bierman02': 'service-list'}
+ if return_key[RESTCONF_VERSION] in res.keys():
+ response_attribute = res[return_key[RESTCONF_VERSION]]
+ else:
+ response_attribute = res['errors']['error'][0]
+ return {'status_code': response.status_code,
+ 'service-list': response_attribute}
+
+
+def get_ordm_serv_list_attr_request(attribute: str, value: str):
+ url = {'rfc8040': '{}/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,
+ 'draft-bierman02': attribute}
+ if return_key[RESTCONF_VERSION] in res.keys():
+ response_attribute = res[return_key[RESTCONF_VERSION]]
+ else:
+ response_attribute = res['errors']['error'][0]
+ return {'status_code': response.status_code,
+ attribute: response_attribute}
+
+
+def get_serv_path_list_attr(attribute: str, value: str):
+ url = {'rfc8040': '{}/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,
+ 'draft-bierman02': attribute}
+ if return_key[RESTCONF_VERSION] in res.keys():
+ response_attribute = res[return_key[RESTCONF_VERSION]]
+ else:
+ response_attribute = res['errors']['error'][0]
+ return {'status_code': response.status_code,
+ attribute: response_attribute}
+
+
+#
+# TransportPCE internal API RPCs
+#
+
+
+def prepend_dict_keys(input_dict: dict, prefix: str):
+ return_dict = {}
+ for key, value in input_dict.items():
+ newkey = prefix + key
+ if isinstance(value, dict):
+ return_dict[newkey] = prepend_dict_keys(value, prefix)
+ # TODO: perhaps some recursion depth limit or another solution has to be considered here
+ # even if recursion depth is given by the input_dict argument
+ # direct (self-)recursive functions may carry unwanted side-effects such as ressource consumptions
+ else:
+ return_dict[newkey] = value
+ return return_dict
+
+
+def transportpce_api_rpc_request(api_module: str, rpc: str, payload: dict):
+ # pylint: disable=consider-using-f-string
+ url = "{}/operations/{}:{}".format('{}', api_module, rpc)
+ if payload is None:
+ data = None
+ elif RESTCONF_VERSION == 'draft-bierman02':
+ data = prepend_dict_keys({'input': payload}, api_module + ':')
+ else:
+ data = {'input': payload}
+ response = post_request(url, data)
+ if response.status_code == requests.codes.no_content:
+ return_output = None
+ else:
+ res = response.json()
+ return_key = {'rfc8040': api_module + ':output',
+ 'draft-bierman02': 'output'}
+ if response.status_code == requests.codes.internal_server_error:
+ return_output = res
+ else:
+ return_output = res[return_key[RESTCONF_VERSION]]
+ return {'status_code': response.status_code,
+ 'output': return_output}