Fix new pylint issues in functional tests

[transportpce.git] / tests / transportpce_tests / common / test_utils_rfc8040.py

diff --git a/tests/transportpce_tests/common/test_utils_rfc8040.py b/tests/transportpce_tests/common/test_utils_rfc8040.py
index 18d2a7d3e407f7d5210af17f20c72deeff2c2ccc..bcbe35ef467b68da1d6e7a524b2046db5d6228fd 100644 (file)
--- a/tests/transportpce_tests/common/test_utils_rfc8040.py
+++ b/tests/transportpce_tests/common/test_utils_rfc8040.py
@@ -29,8 +29,7 @@ import simulators
 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'
@@ -41,6 +40,8 @@ NODES_PWD = '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'
 
@@ -88,7 +89,8 @@ def get_request(url):
     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):
@@ -96,14 +98,16 @@ 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):
@@ -112,11 +116,13 @@ 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
@@ -375,6 +381,21 @@ def del_node_attribute_request(node: str, attribute: str, attribute_value: str):
 #
 
 
+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={}',
@@ -413,14 +434,32 @@ def get_ietf_network_request(network: str, content: str):
     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/{}'}
@@ -480,10 +519,13 @@ def get_ietf_network_node_request(network: str, node: str, content: str):
     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}
 
@@ -540,6 +582,21 @@ def get_ordm_serv_list_attr_request(attribute: str, value: str):
             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
 #