X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2Fcommon%2Ftest_utils_rfc8040.py;h=d83c1d9df288ce49c345abfee62d0d21d29e521e;hb=8ec42577fc9bb12bed1a7689cd4a17487e3c1ff6;hp=c65c51ddd7af7eda3e26d853984105ea0ad17309;hpb=b007d00d5264bd6d28b461ef6a600135f1d513f3;p=transportpce.git diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py index c65c51ddd..d83c1d9df 100644 --- a/tests/transportpce_tests/common/test_utils_rfc8040.py +++ b/tests/transportpce_tests/common/test_utils_rfc8040.py @@ -304,7 +304,7 @@ def unmount_device(node: str): def check_device_connection(node: str): - url = {'rfc8040': '{}/data/network-topology:network-topology/topology=topology-netconf/node={}', + 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() @@ -318,79 +318,41 @@ def check_device_connection(node: str): 'connection-status': connection_status} -def check_interface_request(node: str, interface: str): +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/interface={}', # nopep8 - 'draft-bierman02': '{}/operational/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/{}'} # nopep8 - response = get_request(url[RESTCONF_VERSION].format('{}', node, interface)) + 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:interface', - 'draft-bierman02': 'interface'} + return_key = {'rfc8040': 'org-openroadm-device:' + attribute, + 'draft-bierman02': attribute} if return_key[RESTCONF_VERSION] in res.keys(): - interface = res[return_key[RESTCONF_VERSION]] + response_attribute = res[return_key[RESTCONF_VERSION]] else: - interface = res['errors']['error'][0] + response_attribute = res['errors']['error'][0] return {'status_code': response.status_code, - 'interface': interface} - - -def del_interface_request(node: str, interface: 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/interface={}', # nopep8 - 'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/{}'} # nopep8 - # draft-bierman02: note this is config here and not operational as previously in check_interface_request - response = delete_request(url[RESTCONF_VERSION].format('{}', node, interface)) - return response - - -def check_roadm_connections_request(node: str, connections: 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/roadm-connections={}', # nopep8 - 'draft-bierman02': '{}/operational/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/roadm-connections/{}'} # nopep8 - response = get_request(url[RESTCONF_VERSION].format('{}', node, connections)) - res = response.json() - return_key = {'rfc8040': 'org-openroadm-device:roadm-connections', - 'draft-bierman02': 'roadm-connections'} - if return_key[RESTCONF_VERSION] in res.keys(): - roadm_connections = res[return_key[RESTCONF_VERSION]] - else: - roadm_connections = res['errors']['error'][0] - return {'status_code': response.status_code, - 'roadm-connections': roadm_connections} - + attribute: response_attribute} -def del_roadm_connections_request(node: str, connections: 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/roadm-connections={}', # nopep8 - 'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/roadm-connections/{}'} # nopep8 - # draft-bierman02: note this is config here and not operational as previously in check_interface_request - response = delete_request(url[RESTCONF_VERSION].format('{}', node, connections)) - return response - -def check_circuit_packs_request(node: str, circuit_packs: str): +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/circuit-packs={}', # nopep8 - 'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/circuit-packs/{}'} # nopep8 - # draft-bierman02: note this is config here and not operational as previously in check_interface_request - # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-591 - response = get_request(url[RESTCONF_VERSION].format('{}', node, circuit_packs)) + 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() - return_key = {'rfc8040': 'org-openroadm-device:circuit-packs', - 'draft-bierman02': 'circuit-packs'} - if return_key[RESTCONF_VERSION] in res.keys(): - cp = res[return_key[RESTCONF_VERSION]] + if attribute2 in res.keys(): + response_attribute = res[attribute2] else: - cp = res['errors']['error'][0] + response_attribute = res['errors']['error'][0] return {'status_code': response.status_code, - 'circuit-packs': cp} + attribute2: response_attribute} -def del_circuit_packs_request(node: str, circuit_packs: str): +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/circuit-packs={}', # nopep8 - 'draft-bierman02': '{}/config/network-topology:network-topology/topology/topology-netconf/node/{}/yang-ext:mount/org-openroadm-device:org-openroadm-device/circuit-packs/{}'} # nopep8 - response = delete_request(url[RESTCONF_VERSION].format('{}', node, circuit_packs)) + 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 # @@ -437,6 +399,18 @@ def portmapping_request(node: str, mapping: str): 'mapping': mapping} +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/{}'} @@ -533,7 +507,7 @@ def del_ietf_network_node_request(network: str, node: str, content: str): return response # -# TransportPCE network-utils and service-path operations +# TransportPCE internal API RPCs # @@ -569,10 +543,31 @@ def connect_rdm_to_xpdr_request(payload: dict): return post_request(url, data) -def device_renderer_service_path_request(payload: dict): - url = "{}/operations/transportpce-device-renderer:service-path" +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(payload, 'transportpce-device-renderer:') + data = prepend_dict_keys({'input': payload}, 'transportpce-device-renderer:') else: - data = payload + data = {'input': payload} return post_request(url, data) + + +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': api_module + ':output', + 'draft-bierman02': 'output'} + return_output = res[return_key[RESTCONF_VERSION]] + return {'status_code': response.status_code, + 'output': return_output}