X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F2.2.1%2Ftest13_tapi_full_multi_layer.py;h=4c8d3c9b035edd43a4299d859d65641f36e0d1aa;hb=88b66915d592d2f1602581bd6991b47b89673f71;hp=348610107c11654d2b61f3e72aced49a78aa20d5;hpb=a8343688872d307ada1b4521721313a8ac3a3e5e;p=transportpce.git diff --git a/tests/transportpce_tests/2.2.1/test13_tapi_full_multi_layer.py b/tests/transportpce_tests/2.2.1/test13_tapi_full_multi_layer.py index 348610107..4c8d3c9b0 100644 --- a/tests/transportpce_tests/2.2.1/test13_tapi_full_multi_layer.py +++ b/tests/transportpce_tests/2.2.1/test13_tapi_full_multi_layer.py @@ -132,22 +132,27 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(5) def test_01_connect_spdrA(self): - response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION)) + print("Connecting SPDRA") + response = test_utils.mount_tapi_device("SPDR-SA1", ('spdra', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_02_connect_spdrC(self): - response = test_utils.mount_device("SPDR-SC1", ('spdrc', self.NODE_VERSION)) + print("Connecting SPDRC") + response = test_utils.mount_tapi_device("SPDR-SC1", ('spdrc', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) def test_03_connect_rdmA(self): - response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION)) + print("Connecting ROADMA") + response = test_utils.mount_tapi_device("ROADM-A1", ('roadma', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) + time.sleep(2) def test_04_connect_rdmC(self): - response = test_utils.mount_device("ROADM-C1", ('roadmc', self.NODE_VERSION)) + print("Connecting ROADMC") + response = test_utils.mount_tapi_device("ROADM-C1", ('roadmc', self.NODE_VERSION)) self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) @@ -169,7 +174,16 @@ class TransportPCEtesting(unittest.TestCase): res["output"]["result"]) time.sleep(2) - def test_07_connect_sprdC_1_N1_to_roadmC_PP1(self): + def test_07_connect_roadmA_PP1_to_spdrA_1_N1_tapi(self): + response = test_utils.connect_xpdr_to_rdm_tapi_request("SPDR-SA1-XPDR1", "XPDR1-NETWORK1", + "ROADM-A1", "SRG1-PP1-TXRX") + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Link created in tapi topology', + res["output"]["result"]) + time.sleep(2) + + def test_08_connect_sprdC_1_N1_to_roadmC_PP1(self): response = test_utils.connect_xpdr_to_rdm_request("SPDR-SC1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok) @@ -178,7 +192,7 @@ class TransportPCEtesting(unittest.TestCase): res["output"]["result"]) time.sleep(2) - def test_08_connect_roadmC_PP1_to_spdrC_1_N1(self): + def test_09_connect_roadmC_PP1_to_spdrC_1_N1(self): response = test_utils.connect_rdm_to_xpdr_request("SPDR-SC1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok) @@ -187,7 +201,16 @@ class TransportPCEtesting(unittest.TestCase): res["output"]["result"]) time.sleep(2) - def test_09_add_omsAttributes_ROADMA_ROADMC(self): + def test_10_connect_roadmC_PP1_to_spdrC_1_N1_tapi(self): + response = test_utils.connect_xpdr_to_rdm_tapi_request("SPDR-SC1-XPDR1", "XPDR1-NETWORK1", + "ROADM-C1", "SRG1-PP1-TXRX") + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Link created in tapi topology', + res["output"]["result"]) + time.sleep(2) + + def test_11_add_omsAttributes_ROADMA_ROADMC(self): # Config ROADMA-ROADMC oms-attributes data = {"span": { "auto-spanloss": "true", @@ -204,7 +227,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.created) time.sleep(2) - def test_10_add_omsAttributes_ROADMC_ROADMA(self): + def test_12_add_omsAttributes_ROADMC_ROADMA(self): # Config ROADMC-ROADMA oms-attributes data = {"span": { "auto-spanloss": "true", @@ -221,7 +244,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(response.status_code, requests.codes.created) time.sleep(2) - def test_11_check_otn_topology(self): + def test_13_check_otn_topology(self): response = test_utils.get_otn_topo_request() self.assertEqual(response.status_code, requests.codes.ok) res = response.json() @@ -230,7 +253,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertNotIn('ietf-network-topology:link', res['network'][0]) time.sleep(2) - def test_12_check_openroadm_topology(self): + def test_14_check_openroadm_topology(self): response = test_utils.get_ordm_topo_request("") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() @@ -240,7 +263,7 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(nbLink, 22, 'There should be 22 openroadm links') time.sleep(2) - def test_13_get_tapi_topology_details(self): + def test_15_get_tapi_topology_details(self): response = test_utils.tapi_get_topology_details_request( "T0 - Full Multi-layer topology") time.sleep(2) @@ -249,10 +272,10 @@ class TransportPCEtesting(unittest.TestCase): nbNode = len(res['output']['topology']['node']) nbLink = len(res['output']['topology']['link']) self.assertEqual(nbNode, 14, 'There should be 14 TAPI nodes') - self.assertEqual(nbLink, 13, 'There should be 13 TAPI links') + self.assertEqual(nbLink, 15, 'There should be 15 TAPI links') time.sleep(2) - def test_14_check_sip_details(self): + def test_16_check_sip_details(self): response = test_utils.tapi_get_sip_details_request() self.assertEqual(response.status_code, requests.codes.ok) res = response.json() @@ -261,7 +284,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) # test create connectivity service from spdrA to spdrC for Photonic_media - def test_15_create_connectivity_service_PhotonicMedia(self): + def test_17_create_connectivity_service_PhotonicMedia(self): response = test_utils.tapi_create_connectivity_request(self.cr_serv_sample_data) time.sleep(self.WAITING) self.assertEqual(response.status_code, requests.codes.ok) @@ -288,9 +311,10 @@ class TransportPCEtesting(unittest.TestCase): res['output']['service']['end-point'][0]['name'][0]) self.assertDictEqual(dict(input_dict_3, **res['output']['service']['end-point'][1]['name'][0]), res['output']['service']['end-point'][1]['name'][0]) + # If the gate fails is because of the waiting time not being enough time.sleep(self.WAITING) - def test_16_get_service_PhotonicMedia(self): + def test_18_get_service_PhotonicMedia(self): response = test_utils.get_service_list_request( "services/" + str(self.uuid_services.pm)) self.assertEqual(response.status_code, requests.codes.ok) @@ -306,12 +330,12 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) # test create connectivity service from spdrA to spdrC for odu - def test_17_create_connectivity_service_ODU(self): + def test_19_create_connectivity_service_ODU(self): # pylint: disable=line-too-long self.cr_serv_sample_data["input"]["end-point"][0]["layer-protocol-name"] = "ODU" - self.cr_serv_sample_data["input"]["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = "eecbfa6e-57ab-3651-9606-c22c8ce73f18" + self.cr_serv_sample_data["input"]["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = "5efda776-f8de-3e0b-9bbd-2c702e210946" self.cr_serv_sample_data["input"]["end-point"][1]["layer-protocol-name"] = "ODU" - self.cr_serv_sample_data["input"]["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = "31f83b1f-29b2-3a8e-af9b-6423dbc5aa22" + self.cr_serv_sample_data["input"]["end-point"][1]["service-interface-point"]["service-interface-point-uuid"] = "8116d0af-39fa-3df5-bed2-dd2cd5e8217d" self.cr_serv_sample_data["input"]["connectivity-constraint"]["service-layer"] = "ODU" response = test_utils.tapi_create_connectivity_request(self.cr_serv_sample_data) @@ -340,9 +364,10 @@ class TransportPCEtesting(unittest.TestCase): res['output']['service']['end-point'][0]['name'][0]) self.assertDictEqual(dict(input_dict_3, **res['output']['service']['end-point'][1]['name'][0]), res['output']['service']['end-point'][1]['name'][0]) + # If the gate fails is because of the waiting time not being enough time.sleep(self.WAITING) - def test_18_get_service_ODU(self): + def test_20_get_service_ODU(self): response = test_utils.get_service_list_request( "services/" + str(self.uuid_services.odu)) self.assertEqual(response.status_code, requests.codes.ok) @@ -358,7 +383,7 @@ class TransportPCEtesting(unittest.TestCase): time.sleep(2) # test create connectivity service from spdrA to spdrC for dsr - def test_19_create_connectivity_service_DSR(self): + def test_21_create_connectivity_service_DSR(self): # pylint: disable=line-too-long self.cr_serv_sample_data["input"]["end-point"][0]["layer-protocol-name"] = "DSR" self.cr_serv_sample_data["input"]["end-point"][0]["service-interface-point"]["service-interface-point-uuid"] = "c14797a0-adcc-3875-a1fe-df8949d1a2d7" @@ -396,9 +421,10 @@ class TransportPCEtesting(unittest.TestCase): self.assertDictEqual(dict(input_dict_3, **res['output']['service']['end-point'][1]['name'][0]), res['output']['service']['end-point'][1]['name'][0]) + # The sleep here is okey as the DSR service creation is very fast time.sleep(self.WAITING) - def test_20_get_service_DSR(self): + def test_22_get_service_DSR(self): response = test_utils.get_service_list_request( "services/" + str(self.uuid_services.dsr)) self.assertEqual(response.status_code, requests.codes.ok) @@ -413,7 +439,7 @@ class TransportPCEtesting(unittest.TestCase): res['services'][0]['lifecycle-state'], 'planned') time.sleep(2) - def test_21_get_connectivity_service_list(self): + def test_23_get_connectivity_service_list(self): response = test_utils.tapi_get_service_list_request() self.assertEqual(response.status_code, requests.codes.ok) res = response.json() @@ -428,32 +454,32 @@ class TransportPCEtesting(unittest.TestCase): self.assertEqual(ele['operational-state'], 'ENABLED') self.assertEqual(ele['service-layer'], 'ODU') nbconnection = len(ele['connection']) - self.assertEqual(nbconnection, 3, 'There should be 3 connections') + self.assertEqual(nbconnection, 1, 'There should be 3 connections') elif ele['uuid'] == self.uuid_services.dsr: self.assertEqual(ele['operational-state'], 'ENABLED') self.assertEqual(ele['service-layer'], 'DSR') nbconnection = len(ele['connection']) - self.assertEqual(nbconnection, 1, 'There should be 1 connection') + self.assertEqual(nbconnection, 4, 'There should be 4 connections') else: self.fail("get connectivity service failed") time.sleep(2) - def test_22_delete_connectivity_service_DSR(self): + def test_24_delete_connectivity_service_DSR(self): response = test_utils.tapi_delete_connectivity_request(self.uuid_services.dsr) self.assertEqual(response.status_code, requests.codes.no_content) time.sleep(self.WAITING) - def test_23_delete_connectivity_service_ODU(self): + def test_25_delete_connectivity_service_ODU(self): response = test_utils.tapi_delete_connectivity_request(self.uuid_services.odu) self.assertEqual(response.status_code, requests.codes.no_content) time.sleep(self.WAITING) - def test_24_delete_connectivity_service_PhotonicMedia(self): + def test_26_delete_connectivity_service_PhotonicMedia(self): response = test_utils.tapi_delete_connectivity_request(self.uuid_services.pm) self.assertEqual(response.status_code, requests.codes.no_content) time.sleep(self.WAITING) - def test_25_get_no_tapi_services(self): + def test_27_get_no_tapi_services(self): response = test_utils.tapi_get_service_list_request() res = response.json() self.assertIn( @@ -463,7 +489,7 @@ class TransportPCEtesting(unittest.TestCase): res['errors']['error']) time.sleep(2) - def test_26_get_no_openroadm_services(self): + def test_28_get_no_openroadm_services(self): response = test_utils.get_service_list_request("") self.assertEqual(response.status_code, requests.codes.conflict) res = response.json() @@ -473,22 +499,22 @@ class TransportPCEtesting(unittest.TestCase): res['errors']['error']) time.sleep(2) - def test_27_disconnect_spdrA(self): + def test_29_disconnect_spdrA(self): response = test_utils.unmount_device("SPDR-SA1") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) - def test_28_disconnect_spdrC(self): + def test_30_disconnect_spdrC(self): response = test_utils.unmount_device("SPDR-SC1") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) - def test_29_disconnect_roadmA(self): + def test_31_disconnect_roadmA(self): response = test_utils.unmount_device("ROADM-A1") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) - def test_30_disconnect_roadmC(self): + def test_32_disconnect_roadmC(self): response = test_utils.unmount_device("ROADM-C1") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)