SIMS = simulators.SIMS
HONEYNODE_OK_START_MSG = 'Netconf SSH endpoint started successfully at 0.0.0.0'
-KARAF_OK_START_MSG = re.escape(
- "Blueprint container for bundle org.opendaylight.netconf.restconf")+".* was successfully created"
+KARAF_OK_START_MSG = "Blueprint container for bundle org.opendaylight.netconf.restconf.* was successfully created"
LIGHTY_OK_START_MSG = re.escape("lighty.io and RESTCONF-NETCONF started")
ODL_LOGIN = 'admin'
def post_request(url, data):
if data:
- print(json.dumps(data))
return requests.request(
"POST", url.format(RESTCONF_BASE_URL),
data=json.dumps(data),
'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, re.escape('Triggering notification stream NETCONF for node ' + node), 180):
+ 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)
#
-def get_portmapping(node: str):
- url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}',
- 'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}'}
- response = get_request(url[RESTCONF_VERSION].format('{}', node))
- res = response.json()
- return_key = {'rfc8040': 'transportpce-portmapping:nodes',
- 'draft-bierman02': 'nodes'}
- nodes = res[return_key[RESTCONF_VERSION]]
- return {'status_code': response.status_code,
- 'nodes': nodes}
+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 get_portmapping_node_info(node: str):
- url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}/node-info',
- 'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}/node-info'}
- response = get_request(url[RESTCONF_VERSION].format('{}', node))
- res = response.json()
- return_key = {'rfc8040': 'transportpce-portmapping:node-info',
- 'draft-bierman02': 'node-info'}
- if return_key[RESTCONF_VERSION] in res.keys():
- node_info = res[return_key[RESTCONF_VERSION]]
- else:
- node_info = res['errors']['error'][0]
- return {'status_code': response.status_code,
- 'node-info': node_info}
-
-
-def portmapping_request(node: str, mapping: str):
- url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}/mapping={}',
- 'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}/mapping/{}'}
- response = get_request(url[RESTCONF_VERSION].format('{}', node, mapping))
- res = response.json()
- return_key = {'rfc8040': 'transportpce-portmapping:mapping',
- 'draft-bierman02': 'mapping'}
- mapping = res[return_key[RESTCONF_VERSION]]
- return {'status_code': response.status_code,
- 'mapping': mapping}
-
+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 portmapping_switching_pool_request(node: str, switching_pool: str):
- url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}/switching-pool-lcp={}',
- 'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}/switching-pool-lcp/{}'}
- response = get_request(url[RESTCONF_VERSION].format('{}', node, switching_pool))
- res = response.json()
- return_key = {'rfc8040': 'transportpce-portmapping:switching-pool-lcp',
- 'draft-bierman02': 'switching-pool-lcp'}
- switching_pool = res[return_key[RESTCONF_VERSION]]
- return {'status_code': response.status_code,
- 'switching_pool': switching_pool}
-
-def portmapping_mc_capa_request(node: str, mc_capa: str):
- url = {'rfc8040': '{}/data/transportpce-portmapping:network/nodes={}/mc-capabilities={}',
- 'draft-bierman02': '{}/config/transportpce-portmapping:network/nodes/{}/mc-capabilities/{}'}
- response = get_request(url[RESTCONF_VERSION].format('{}', node, mc_capa))
+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:mc-capabilities',
- 'draft-bierman02': 'mc-capabilities'}
- capabilities = res[return_key[RESTCONF_VERSION]]
+ 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,
- 'mc-capabilities': capabilities}
+ attr: return_output}
#
# Topology operations
else:
format_args = ('{}', 'operational', network)
response = get_request(url[RESTCONF_VERSION].format(*format_args))
- res = response.json()
- return_key = {'rfc8040': 'ietf-network:network',
- 'draft-bierman02': 'network'}
- networks = res[return_key[RESTCONF_VERSION]]
+ 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/{}'}
else:
format_args = ('{}', 'operational', network, node)
response = get_request(url[RESTCONF_VERSION].format(*format_args))
- res = response.json()
- return_key = {'rfc8040': 'ietf-network:node',
- 'draft-bierman02': 'node'}
- node = res[return_key[RESTCONF_VERSION]][0]
+ 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}