X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2Fcommon%2Ftest_utils_rfc8040.py;h=d53ddbf061f4c0e607ece5ec07178cb8bf40df25;hb=c015bf8edf32eaf6fe78caa8bfe20cb5acb02b31;hp=ae4716189868f53f4477a07e34761d93f7f8cfa7;hpb=6dc56e813eefe81ce293823936c500869b841e0a;p=transportpce.git diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py index ae4716189..d53ddbf06 100644 --- a/tests/transportpce_tests/common/test_utils_rfc8040.py +++ b/tests/transportpce_tests/common/test_utils_rfc8040.py @@ -494,6 +494,24 @@ def del_oms_attr_request(link: str,): 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 == '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)) + res = response.json() + return_key = {'rfc8040': 'ietf-network:node', + 'draft-bierman02': 'node'} + node = res[return_key[RESTCONF_VERSION]][0] + 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/{}'} @@ -506,165 +524,77 @@ def del_ietf_network_node_request(network: str, node: str, content: str): response = delete_request(url[RESTCONF_VERSION].format(*format_args)) return response + # -# TransportPCE internal API RPCs +# Service list operations # -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 connect_xpdr_to_rdm_request(payload: dict): - url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links" - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': {'links-input': payload}}, 'networkutils:') - else: - data = {'input': {'links-input': payload}} - return post_request(url, data) - - -def connect_rdm_to_xpdr_request(payload: dict): - url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links" - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': {'links-input': payload}}, 'networkutils:') - else: - data = {'input': {'links-input': payload}} - return post_request(url, data) - - -def device_renderer_create_ots_oms_request(nodeid: str, lcp: str): - url = "{}/operations/transportpce-device-renderer:create-ots-oms" - payload = { - 'node-id': nodeid, - 'logical-connection-point': lcp} - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': payload}, 'transportpce-device-renderer:') - else: - data = {'input': payload} - return post_request(url, data) - - -def device_renderer_service_path_request(payload: dict): - url = "{}/operations/transportpce-device-renderer:service-path" - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': payload}, 'transportpce-device-renderer:') - else: - data = {'input': payload} - response = post_request(url, data) +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': 'transportpce-device-renderer:output', - 'draft-bierman02': 'output'} - return_output = res[return_key[RESTCONF_VERSION]] - return {'status_code': response.status_code, - 'output': return_output} - - -def device_renderer_otn_service_path_request(payload: dict): - url = "{}/operations/transportpce-device-renderer:otn-service-path" - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': payload}, 'transportpce-device-renderer:') + 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: - data = {'input': payload} - response = post_request(url, data) - res = response.json() - return_key = {'rfc8040': 'transportpce-device-renderer:output', - 'draft-bierman02': 'output'} - return_output = res[return_key[RESTCONF_VERSION]] + response_attribute = res['errors']['error'][0] return {'status_code': response.status_code, - 'output': return_output} + 'service-list': response_attribute} -def renderer_service_implementation_request(payload: dict): - url = "{}/operations/transportpce-renderer:service-implementation-request" - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': payload}, 'transportpce-renderer:') +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/{}/{}'} + if RESTCONF_VERSION == 'rfc8040': + format_args = ('{}', attribute, value) else: - data = {'input': payload} - response = post_request(url, data) + format_args = ('{}', attribute, value) + response = get_request(url[RESTCONF_VERSION].format(*format_args)) res = response.json() - return_key = {'rfc8040': 'transportpce-renderer:output', - 'draft-bierman02': 'output'} - return_output = res[return_key[RESTCONF_VERSION]] - return {'status_code': response.status_code, - 'output': return_output} - - -def olm_get_pm_request(payload: dict): - url = "{}/operations/transportpce-olm:get-pm" - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': payload}, 'transportpce-olm:') + 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: - data = {'input': payload} - response = post_request(url, data) - res = response.json() - return_key = {'rfc8040': 'transportpce-olm:output', - 'draft-bierman02': 'output'} - return_output = res[return_key[RESTCONF_VERSION]] + response_attribute = res['errors']['error'][0] return {'status_code': response.status_code, - 'output': return_output} + attribute: response_attribute} -def olm_calculate_spanloss_base_request(payload: dict): - url = "{}/operations/transportpce-olm:calculate-spanloss-base" - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': payload}, 'transportpce-olm:') - else: - data = {'input': payload} - response = post_request(url, data) - res = response.json() - return_key = {'rfc8040': 'transportpce-olm:output', - 'draft-bierman02': 'output'} - return_output = res[return_key[RESTCONF_VERSION]] - return {'status_code': response.status_code, - 'output': return_output} +# +# TransportPCE internal API RPCs +# -def olm_service_power_setup_request(payload: dict): - url = "{}/operations/transportpce-olm:service-power-setup" - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': payload}, 'transportpce-olm:') - else: - data = {'input': payload} - response = post_request(url, data) - res = response.json() - return_key = {'rfc8040': 'transportpce-olm:output', - 'draft-bierman02': 'output'} - return_output = res[return_key[RESTCONF_VERSION]] - return {'status_code': response.status_code, - 'output': return_output} +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 olm_service_power_turndown_request(payload: dict): - url = "{}/operations/transportpce-olm:service-power-turndown" - if RESTCONF_VERSION == 'draft-bierman02': - data = prepend_dict_keys({'input': payload}, 'transportpce-olm:') +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) res = response.json() - return_key = {'rfc8040': 'transportpce-olm:output', - 'draft-bierman02': 'output'} - return_output = res[return_key[RESTCONF_VERSION]] - return {'status_code': response.status_code, - 'output': return_output} - - -def olm_calculate_spanloss_current_request(): - url = "{}/operations/transportpce-olm:calculate-spanloss-current" - response = post_request(url, None) - res = response.json() - return_key = {'rfc8040': 'transportpce-olm:output', + return_key = {'rfc8040': api_module + ':output', 'draft-bierman02': 'output'} return_output = res[return_key[RESTCONF_VERSION]] return {'status_code': response.status_code,