X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F1.2.1%2Ftest_portmapping.py;h=6a9f9fc47ce60456df272647795f3819906e35be;hb=c834356f10cb8bc04c8b0a9eb2a438b2e9c50632;hp=a1daf17f270c66950da29426721010e4df476a76;hpb=802e7005667364a567c19bd50e9c167f63028cee;p=transportpce.git diff --git a/tests/transportpce_tests/1.2.1/test_portmapping.py b/tests/transportpce_tests/1.2.1/test_portmapping.py index a1daf17f2..6a9f9fc47 100644 --- a/tests/transportpce_tests/1.2.1/test_portmapping.py +++ b/tests/transportpce_tests/1.2.1/test_portmapping.py @@ -9,23 +9,30 @@ # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## +# pylint: disable=no-member +# pylint: disable=too-many-public-methods + import unittest import time import requests -from common import test_utils +import sys +sys.path.append('transportpce_tests/common/') +import test_utils class TransportPCEPortMappingTesting(unittest.TestCase): processes = None + NODE_VERSION = '1.2.1' @classmethod def setUpClass(cls): cls.processes = test_utils.start_tpce() - cls.processes = test_utils.start_sims(['xpdra', 'roadma']) + cls.processes = test_utils.start_sims([('xpdra', cls.NODE_VERSION), ('roadma', cls.NODE_VERSION)]) @classmethod def tearDownClass(cls): + # pylint: disable=not-an-iterable for process in cls.processes: test_utils.shutdown_process(process) print("all processes killed") @@ -34,26 +41,8 @@ class TransportPCEPortMappingTesting(unittest.TestCase): print("execution of {}".format(self.id().split(".")[-1])) time.sleep(10) -# def test_01_restconfAPI(self): -# response = test_utils.get_netconf_oper_request("controller-config") -# self.assertEqual(response.status_code, requests.codes.ok) -# res = response.json() -# self.assertEqual(res['node'] [0] ['netconf-node-topology:connection-status'], -# 'connected') - -# def test_02_restconfAPI(self): -# url = ("{}/config/transportpce-portmapping:network/nodes/controller-config") -# response = requests.request( -# "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) -# self.assertEqual(response.status_code, requests.codes.not_found) -# res = response.json() -# self.assertIn( -# {"error-type":"application", "error-tag":"data-missing", -# "error-message":"Request could not be completed because the relevant data model content does not exist "}, -# res['errors']['error']) - def test_01_rdm_device_connection(self): - response = test_utils.mount_device("ROADMA01", 'roadma') + response = test_utils.mount_device("ROADMA01", ('roadma', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_02_rdm_device_connected(self): @@ -66,10 +55,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(10) def test_03_rdm_portmapping_info(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/ROADMA01/node-info" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADMA01/node-info") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -82,43 +68,37 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(3) def test_04_rdm_portmapping_DEG1_TTP_TXRX(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/ROADMA01/mapping/DEG1-TTP-TXRX" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADMA01/mapping/DEG1-TTP-TXRX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0', - 'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'}, + 'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional', + 'port-admin-state': 'InService', 'port-oper-state': 'InService'}, res['mapping']) def test_05_rdm_portmapping_SRG1_PP7_TXRX(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/ROADMA01/mapping/SRG1-PP7-TXRX" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADMA01/mapping/SRG1-PP7-TXRX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0', - 'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional'}, + 'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional', + 'port-admin-state': 'InService', 'port-oper-state': 'InService'}, res['mapping']) def test_06_rdm_portmapping_SRG3_PP1_TXRX(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/ROADMA01/mapping/SRG3-PP1-TXRX" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("ROADMA01/mapping/SRG3-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0', - 'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional'}, + 'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional', + 'port-admin-state': 'InService', 'port-oper-state': 'InService'}, res['mapping']) def test_07_xpdr_device_connection(self): - response = test_utils.mount_device("XPDRA01", 'xpdra') + response = test_utils.mount_device("XPDRA01", ('xpdra', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_08_xpdr_device_connected(self): @@ -131,10 +111,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(10) def test_09_xpdr_portmapping_info(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/node-info" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/node-info") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -147,38 +124,31 @@ class TransportPCEPortMappingTesting(unittest.TestCase): time.sleep(3) def test_10_xpdr_portmapping_NETWORK1(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-NETWORK1" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET', 'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network', - 'lcp-hash-val': 'OSvMgUyP+mE='}, + 'lcp-hash-val': 'OSvMgUyP+mE=', + 'port-admin-state': 'InService', 'port-oper-state': 'InService'}, res['mapping']) def test_11_xpdr_portmapping_NETWORK2(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-NETWORK2" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK2") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET', 'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-CLIENT3', 'port-qual': 'xpdr-network', - 'lcp-hash-val': 'OSvMgUyP+mI='}, + 'lcp-hash-val': 'OSvMgUyP+mI=', + 'port-admin-state': 'InService', 'port-oper-state': 'InService'}, res['mapping']) def test_12_xpdr_portmapping_CLIENT1(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-CLIENT1" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -186,14 +156,12 @@ class TransportPCEPortMappingTesting(unittest.TestCase): 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT', 'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client', - 'lcp-hash-val': 'AO9UFkY/TLYw'}, + 'lcp-hash-val': 'AO9UFkY/TLYw', + 'port-admin-state': 'InService', 'port-oper-state': 'InService'}, res['mapping']) def test_13_xpdr_portmapping_CLIENT2(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-CLIENT2" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT2") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -201,14 +169,12 @@ class TransportPCEPortMappingTesting(unittest.TestCase): 'supporting-circuit-pack-name': '1/0/C2-PLUG-CLIENT', 'logical-connection-point': 'XPDR1-CLIENT2', 'port-direction': 'bidirectional', 'port-qual': 'xpdr-client', - 'lcp-hash-val': 'AO9UFkY/TLYz'}, + 'lcp-hash-val': 'AO9UFkY/TLYz', + 'port-admin-state': 'InService', 'port-oper-state': 'InService'}, res['mapping']) def test_14_xpdr_portmapping_CLIENT3(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-CLIENT3" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT3") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( @@ -216,21 +182,20 @@ class TransportPCEPortMappingTesting(unittest.TestCase): 'supporting-circuit-pack-name': '1/0/C3-PLUG-CLIENT', 'logical-connection-point': 'XPDR1-CLIENT3', 'connection-map-lcp': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional', - 'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLYy'}, + 'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLYy', + 'port-admin-state': 'InService', 'port-oper-state': 'InService'}, res['mapping']) def test_15_xpdr_portmapping_CLIENT4(self): - url = ("{}/config/transportpce-portmapping:network/" - "nodes/XPDRA01/mapping/XPDR1-CLIENT4" - ) - response = test_utils.get_request(url) + response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT4") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn( {'supporting-port': 'C4', 'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT', 'logical-connection-point': 'XPDR1-CLIENT4', 'port-direction': 'bidirectional', - 'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLY1'}, + 'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLY1', + 'port-admin-state': 'InService', 'port-oper-state': 'InService'}, res['mapping']) def test_16_xpdr_device_disconnection(self): @@ -239,7 +204,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): def test_17_xpdr_device_disconnected(self): response = test_utils.get_netconf_oper_request("XPDRA01") - self.assertEqual(response.status_code, requests.codes.not_found) + self.assertEqual(response.status_code, requests.codes.conflict) res = response.json() self.assertIn( {"error-type": "application", "error-tag": "data-missing", @@ -247,9 +212,8 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['errors']['error']) def test_18_xpdr_device_not_connected(self): - url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01") - response = test_utils.get_request(url) - self.assertEqual(response.status_code, requests.codes.not_found) + response = test_utils.portmapping_request("XPDRA01") + self.assertEqual(response.status_code, requests.codes.conflict) res = response.json() self.assertIn( {"error-type": "application", "error-tag": "data-missing", @@ -262,7 +226,7 @@ class TransportPCEPortMappingTesting(unittest.TestCase): def test_20_rdm_device_disconnected(self): response = test_utils.get_netconf_oper_request("ROADMA01") - self.assertEqual(response.status_code, requests.codes.not_found) + self.assertEqual(response.status_code, requests.codes.conflict) res = response.json() self.assertIn( {"error-type": "application", "error-tag": "data-missing", @@ -270,9 +234,8 @@ class TransportPCEPortMappingTesting(unittest.TestCase): res['errors']['error']) def test_21_rdm_device_not_connected(self): - url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA01") - response = test_utils.get_request(url) - self.assertEqual(response.status_code, requests.codes.not_found) + response = test_utils.portmapping_request("ROADMA01") + self.assertEqual(response.status_code, requests.codes.conflict) res = response.json() self.assertIn( {"error-type": "application", "error-tag": "data-missing",