X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F1.2.1%2Ftest_pce.py;h=a1b35323b03b679e47eb5b6d7321ebd9f8bc2609;hb=19a9408ee823754859467888a5d2823bfd8655ab;hp=fb824f80da930f747fcf583e9d6fb36b9aeff4e9;hpb=c8bd98e0844257ae2e40e01d54546a40596df26d;p=transportpce.git diff --git a/tests/transportpce_tests/1.2.1/test_pce.py b/tests/transportpce_tests/1.2.1/test_pce.py index fb824f80d..a1b35323b 100644 --- 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)