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'
TYPE_APPLICATION_JSON = {'Content-Type': 'application/json', 'Accept': 'application/json'}
TYPE_APPLICATION_XML = {'Content-Type': 'application/xml', 'Accept': 'application/xml'}
+REQUEST_TIMEOUT = 10
+
CODE_SHOULD_BE_200 = 'Http status code should be 200'
CODE_SHOULD_BE_201 = 'Http status code should be 201'
+T100GE = 'Transponder 100GE'
+T0_MULTILAYER_TOPO = 'T0 - Multi-layer topology'
+T0_FULL_MULTILAYER_TOPO = 'T0 - Full Multi-layer topology'
SIM_LOG_DIRECTORY = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'log')
return requests.request(
'GET', url.format(RESTCONF_BASE_URL),
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
def put_request(url, data):
'PUT', url.format(RESTCONF_BASE_URL),
data=json.dumps(data),
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
def delete_request(url):
return requests.request(
'DELETE', url.format(RESTCONF_BASE_URL),
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
def post_request(url, data):
"POST", url.format(RESTCONF_BASE_URL),
data=json.dumps(data),
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
return requests.request(
"POST", url.format(RESTCONF_BASE_URL),
headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+ auth=(ODL_LOGIN, ODL_PWD),
+ timeout=REQUEST_TIMEOUT)
#
# Process management
#
+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={}',
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}
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
#
else:
data = {'input': payload}
response = post_request(url, data)
- res = response.json()
- return_key = {'rfc8040': api_module + ':output',
- 'draft-bierman02': 'output'}
- return_output = res[return_key[RESTCONF_VERSION]]
+ 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}