generalize RESTconf requests methods in func tests

[transportpce.git] / tests / transportpce_tests / 1.2.1 / test_pce.py

diff --git a/tests/transportpce_tests/1.2.1/test_pce.py b/tests/transportpce_tests/1.2.1/test_pce.py
index fb824f80da930f747fcf583e9d6fb36b9aeff4e9..a1b35323b03b679e47eb5b6d7321ebd9f8bc2609 100644 (file)
--- a/tests/transportpce_tests/1.2.1/test_pce.py
+++ b/tests/transportpce_tests/1.2.1/test_pce.py
@@ -8,15 +8,11 @@
 # http://www.apache.org/licenses/LICENSE-2.0
 ##############################################################################
 
+import unittest
 import json
 import os
-import psutil
-import requests
-import signal
-import shutil
-import subprocess
 import time
-import unittest
+import requests
 from common import test_utils
 
 
@@ -62,21 +58,16 @@ class TransportPCEtesting(unittest.TestCase):
 
      # Load simple bidirectional topology
     def test_01_load_simple_topology_bi(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = self.simple_topo_bi_dir_data
-        response = requests.request(
-            "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_XML,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology"
+        data = self.simple_topo_bi_dir_data
+        response = test_utils.put_xmlrequest(url, data)
         self.assertEqual(response.status_code, requests.codes.ok)
         time.sleep(2)
 
     # Get existing nodeId
     def test_02_get_nodeId(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertEqual(
@@ -85,10 +76,8 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Get existing linkId
     def test_03_get_linkId(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology/link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology/link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertEqual(
@@ -98,9 +87,8 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Path Computation success
     def test_04_path_computation_xpdr_bi(self):
-        url = ("{}/operations/transportpce-pce:path-computation-request"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = {"input": {
+        url = "{}/operations/transportpce-pce:path-computation-request"
+        data = {"input": {
                 "service-name": "service-1",
                 "resource-reserve": "true",
                 "pce-metric": "hop-count",
@@ -121,9 +109,7 @@ class TransportPCEtesting(unittest.TestCase):
                 }
                 }
                 }
-        response = requests.request(
-            "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Path is calculated',
@@ -132,9 +118,8 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Path Computation success
     def test_05_path_computation_rdm_bi(self):
-        url = ("{}/operations/transportpce-pce:path-computation-request"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = {"input": {
+        url = "{}/operations/transportpce-pce:path-computation-request"
+        data = {"input": {
                 "service-name": "service-1",
                 "resource-reserve": "true",
                 "pce-metric": "hop-count",
@@ -155,9 +140,7 @@ class TransportPCEtesting(unittest.TestCase):
                 }
                 }
                 }
-        response = requests.request(
-            "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Path is calculated',
@@ -166,39 +149,30 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Delete topology
     def test_06_delete_simple_topology_bi(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "DELETE", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology"
+        response = test_utils.delete_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         time.sleep(2)
 
     # Test deleted topology
     def test_07_test_topology_simple_bi_deleted(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, 404)
         time.sleep(1)
 
     # Load simple bidirectional topology
     def test_08_load_simple_topology_uni(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = self.simple_topo_uni_dir_data
-        response = requests.request(
-            "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_XML,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology"
+        data = self.simple_topo_uni_dir_data
+        response = test_utils.put_xmlrequest(url, data)
         self.assertEqual(response.status_code, 201)
         time.sleep(2)
 
     # Get existing nodeId
     def test_09_get_nodeId(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertEqual(
@@ -208,10 +182,8 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Get existing linkId
     def test_10_get_linkId(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertEqual(
@@ -221,9 +193,8 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Path Computation success
     def test_11_path_computation_xpdr_uni(self):
-        url = ("{}/operations/transportpce-pce:path-computation-request"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = {"input": {
+        url = "{}/operations/transportpce-pce:path-computation-request"
+        data = {"input": {
                 "service-name": "service-1",
                 "resource-reserve": "true",
                 "pce-metric": "hop-count",
@@ -244,9 +215,7 @@ class TransportPCEtesting(unittest.TestCase):
                 }
                 }
                 }
-        response = requests.request(
-            "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Path is calculated',
@@ -255,9 +224,8 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Path Computation success
     def test_12_path_computation_rdm_uni(self):
-        url = ("{}/operations/transportpce-pce:path-computation-request"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = {"input": {
+        url = "{}/operations/transportpce-pce:path-computation-request"
+        data = {"input": {
                 "service-name": "service1",
                 "resource-reserve": "true",
                 "service-handler-header": {
@@ -278,9 +246,7 @@ class TransportPCEtesting(unittest.TestCase):
                 "pce-metric": "hop-count"
                 }
                 }
-        response = requests.request(
-            "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Path is calculated',
@@ -301,39 +267,30 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Delete topology
     def test_13_delete_simple_topology(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "DELETE", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology"
+        response = test_utils.delete_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         time.sleep(2)
 
     # Test deleted topology
     def test_14_test_topology_simple_deleted(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, 404)
         time.sleep(1)
 
     # Load complex topology
     def test_15_load_complex_topology(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = self.complex_topo_uni_dir_data
-        response = requests.request(
-            "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_XML,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology"
+        data = self.complex_topo_uni_dir_data
+        response = test_utils.put_xmlrequest(url, data)
         self.assertEqual(response.status_code, 201)
         time.sleep(2)
 
     # Get existing nodeId
     def test_16_get_nodeId(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertEqual(
@@ -343,17 +300,14 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Test failed path computation
     def test_17_fail_path_computation(self):
-        url = ("{}/operations/transportpce-pce:path-computation-request"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = {"input": {
+        url = "{}/operations/transportpce-pce:path-computation-request"
+        data = {"input": {
                 "service-handler-header": {
                     "request-id": "request-1"
                 }
                 }
                 }
-        response = requests.request(
-            "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Service Name is not set',
@@ -362,9 +316,8 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Test1 success path computation
     def test_18_success1_path_computation(self):
-        url = ("{}/operations/transportpce-pce:path-computation-request"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = {"input": {
+        url = "{}/operations/transportpce-pce:path-computation-request"
+        data = {"input": {
                 "service-name": "service1",
                 "resource-reserve": "true",
                 "service-handler-header": {
@@ -450,9 +403,7 @@ class TransportPCEtesting(unittest.TestCase):
                 "locally-protected-links": "true"
                 }
                 }
-        response = requests.request(
-            "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Path is calculated',
@@ -461,9 +412,8 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Test2 success path computation with path description
     def test_19_success2_path_computation(self):
-        url = ("{}/operations/transportpce-pce:path-computation-request"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = {"input": {
+        url = "{}/operations/transportpce-pce:path-computation-request"
+        data = {"input": {
                 "service-name": "service 1",
                 "resource-reserve": "true",
                 "service-handler-header": {
@@ -484,9 +434,7 @@ class TransportPCEtesting(unittest.TestCase):
                 "pce-metric": "hop-count"
                 }
                 }
-        response = requests.request(
-            "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Path is calculated',
@@ -499,9 +447,8 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Test3 success path computation with hard-constraints exclude
     def test_20_success3_path_computation(self):
-        url = ("{}/operations/transportpce-pce:path-computation-request"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = {"input": {
+        url = "{}/operations/transportpce-pce:path-computation-request"
+        data = {"input": {
                 "service-name": "service 1",
                 "resource-reserve": "true",
                 "service-handler-header": {
@@ -527,9 +474,7 @@ class TransportPCEtesting(unittest.TestCase):
                 "pce-metric": "hop-count"
                 }
                 }
-        response = requests.request(
-            "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Path is calculated',
@@ -542,9 +487,8 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
     def test_21_path_computation_before_oms_attribute_deletion(self):
-        url = ("{}/operations/transportpce-pce:path-computation-request"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = {"input": {
+        url = "{}/operations/transportpce-pce:path-computation-request"
+        data = {"input": {
                 "service-name": "service 1",
                 "resource-reserve": "true",
                 "service-handler-header": {
@@ -565,9 +509,7 @@ class TransportPCEtesting(unittest.TestCase):
                 "pce-metric": "hop-count"
                 }
                 }
-        response = requests.request(
-            "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Path is calculated',
@@ -588,17 +530,15 @@ class TransportPCEtesting(unittest.TestCase):
     def test_22_delete_oms_attribute_in_openroadm13toopenroadm12_link(self):
         url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
                "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2/org-openroadm-network-topology:OMS-attributes/span"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "DELETE", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+               )
+        response = test_utils.delete_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         time.sleep(2)
 
     # Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
     def test_23_path_computation_after_oms_attribute_deletion(self):
-        url = ("{}/operations/transportpce-pce:path-computation-request"
-               .format(test_utils.RESTCONF_BASE_URL))
-        body = {"input": {
+        url = "{}/operations/transportpce-pce:path-computation-request"
+        data = {"input": {
                 "service-name": "service 1",
                 "resource-reserve": "true",
                 "service-handler-header": {
@@ -619,9 +559,7 @@ class TransportPCEtesting(unittest.TestCase):
                 "pce-metric": "hop-count"
                 }
                 }
-        response = requests.request(
-            "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
-            auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        response = test_utils.post_request(url, data)
         self.assertEqual(response.status_code, requests.codes.ok)
         res = response.json()
         self.assertIn('Path is calculated',
@@ -640,19 +578,15 @@ class TransportPCEtesting(unittest.TestCase):
 
     # Delete complex topology
     def test_24_delete_complex_topology(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "DELETE", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology"
+        response = test_utils.delete_request(url)
         self.assertEqual(response.status_code, requests.codes.ok)
         time.sleep(2)
 
     # Test deleted complex topology
     def test_25_test_topology_complex_deleted(self):
-        url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
-               .format(test_utils.RESTCONF_BASE_URL))
-        response = requests.request(
-            "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+        url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
+        response = test_utils.get_request(url)
         self.assertEqual(response.status_code, 404)
         time.sleep(1)