- print("Node " + node_id + " still not deleted from tpce topology", end='... ', flush=True)
- return response
-
-
-def connect_xpdr_to_rdm_request(xpdr_node: str, xpdr_num: str, network_num: str,
- rdm_node: str, srg_num: str, termination_num: str):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links"
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": xpdr_node,
- "networkutils:xpdr-num": xpdr_num,
- "networkutils:network-num": network_num,
- "networkutils:rdm-node": rdm_node,
- "networkutils:srg-num": srg_num,
- "networkutils:termination-point-num": termination_num
- }
- }
- }
- return post_request(url, data)
-
-
-def connect_rdm_to_xpdr_request(xpdr_node: str, xpdr_num: str, network_num: str,
- rdm_node: str, srg_num: str, termination_num: str):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links"
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": xpdr_node,
- "networkutils:xpdr-num": xpdr_num,
- "networkutils:network-num": network_num,
- "networkutils:rdm-node": rdm_node,
- "networkutils:srg-num": srg_num,
- "networkutils:termination-point-num": termination_num
- }
- }
- }
- return post_request(url, data)
-
-
-def connect_xpdr_to_rdm_tapi_request(xpdr_node: str, xpdr_num: str, rdm_node: str, srg_num: str):
- url = "{}/operations/transportpce-tapinetworkutils:init-xpdr-rdm-tapi-link"
- data = {
- "input": {
- "xpdr-node": xpdr_node,
- "network-tp": xpdr_num,
- "rdm-node": rdm_node,
- "add-drop-tp": srg_num
- }
- }
- return post_request(url, data)
-
-
-def check_netconf_node_request(node: str, suffix: str):
- url = URL_CONFIG_NETCONF_TOPO + (
- "node/" + node + "/yang-ext:mount/org-openroadm-device:org-openroadm-device/" + suffix
- )
- return get_request(url)
-
-
-def get_netconf_oper_request(node: str):
- url = "{}/operational/network-topology:network-topology/topology/topology-netconf/node/" + node
- return get_request(url)
-
-
-def get_ordm_topo_request(suffix: str):
- url = URL_CONFIG_ORDM_TOPO + suffix
- return get_request(url)
-
-
-def add_oms_attr_request(link: str, attr):
- url = URL_CONFIG_ORDM_TOPO + (
- "ietf-network-topology:link/" + link + "/org-openroadm-network-topology:OMS-attributes/span"
- )
- return put_request(url, attr)
-
-
-def del_oms_attr_request(link: str):
- url = URL_CONFIG_ORDM_TOPO + (
- "ietf-network-topology:link/" + link + "/org-openroadm-network-topology:OMS-attributes/span"
- )
- return delete_request(url)
-
-
-def get_clli_net_request():
- return get_request(URL_CONFIG_CLLI_NET)
-
-
-def get_ordm_net_request():
- return get_request(URL_CONFIG_ORDM_NET)
-
-
-def get_otn_topo_request():
- return get_request(URL_CONFIG_OTN_TOPO)
-
-
-def del_link_request(link: str):
- url = URL_CONFIG_ORDM_TOPO + ("ietf-network-topology:link/" + link)
- return delete_request(url)
-
-
-def del_node_request(node: str):
- url = URL_CONFIG_CLLI_NET + ("node/" + node)
- return delete_request(url)
-
-
-def portmapping_request(suffix: str):
- url = URL_PORTMAPPING + suffix
- return get_request(url)
-
-
-def get_notifications_process_service_request(attr):
- return post_request(URL_GET_NBINOTIFICATIONS_PROCESS_SERV, attr)
-
-
-def get_notifications_alarm_service_request(attr):
- return post_request(URL_GET_NBINOTIFICATIONS_ALARM_SERV, attr)
-
-
-def get_service_list_request(suffix: str):
- return get_request(URL_OPER_SERV_LIST + suffix)
-
-
-def get_service_path_list_request(suffix: str):
- return get_request(URL_OPER_SERV_PATH_LIST + suffix)
-
-
-def service_create_request(attr):
- return post_request(URL_SERV_CREATE, attr)
-
-
-def service_delete_request(servicename: str,
- requestid="e3028bae-a90f-4ddd-a83f-cf224eba0e58",
- notificationurl="http://localhost:8585/NotificationServer/notify"):
- attr = {"input": {
- "sdnc-request-header": {
- "request-id": requestid,
- "rpc-action": "service-delete",
- "request-system-id": "appname",
- "notification-url": notificationurl},
- "service-delete-req-info": {
- "service-name": servicename,
- "tail-retention": "no"}}}
- return post_request(URL_SERV_DELETE, attr)
+ 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]]
+ elif 'errors' in res.keys():
+ response_attribute = res['errors']['error'][0]
+ else:
+ # status code 400 invalid request
+ response_attribute = res['message'] + ' ' + res['url']
+ print(response_attribute)
+ 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}