+ nodes = res["output"]["topology"]["node"]
+ links = res["output"]["topology"]["link"]
+ self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
+ self.assertEqual(9, len(links), 'Topology should contain 9 links')
+ self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
+ 'Topology should contain 4 transitional links')
+ self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
+ 'Topology should contain 4 oms links')
+ self.assertEqual(1, count_object_with_double_key(links, "name", "value-name", "otn link name"),
+ 'Topology should contain 1 otn link')
+ for link in links:
+ if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
+ self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
+ 'OTU4 link should have an available capacity of 100 000 Mbps')
+ elif link["name"][0]["value-name"] == "transitional link name":
+ self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
+ 'link should have an available capacity of 100 Gbps')
+ self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
+
+ def test_33_create_ODU4_service(self):
+ self.cr_serv_sample_data["input"]["service-name"] = "service1-ODU4"
+ self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "ODU"
+ del self.cr_serv_sample_data["input"]["service-a-end"]["otu-service-rate"]
+ self.cr_serv_sample_data["input"]["service-a-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
+ self.cr_serv_sample_data["input"]["service-z-end"]["service-format"] = "ODU"
+ del self.cr_serv_sample_data["input"]["service-z-end"]["otu-service-rate"]
+ self.cr_serv_sample_data["input"]["service-z-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
+
+ response = test_utils.service_create_request(self.cr_serv_sample_data)
+ self.assertEqual(response.status_code, requests.codes.ok)
+ res = response.json()
+ self.assertIn('PCE calculation in progress',
+ res['output']['configuration-response-common']['response-message'])
+ time.sleep(self.WAITING)
+
+ def test_34_check_tapi_topology_T0(self):
+ url = "{}/operations/tapi-topology:get-topology-details"
+ data = {
+ "tapi-topology:input": {
+ "tapi-topology:topology-id-or-name": "T0 - Multi-layer topology"
+ }
+ }
+ response = test_utils.post_request(url, data)
+ self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
+ res = response.json()
+ nodes = res["output"]["topology"]["node"]
+ links = res["output"]["topology"]["link"]
+ self.assertEqual(9, len(nodes), 'Topology should contain 9 nodes')
+ self.assertEqual(10, len(links), 'Topology should contain 10 links')
+ self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "transitional link name"),
+ 'Topology should contain 4 transitional links')
+ self.assertEqual(4, count_object_with_double_key(links, "name", "value-name", "OMS link name"),
+ 'Topology should contain 4 oms links')
+ self.assertEqual(2, count_object_with_double_key(links, "name", "value-name", "otn link name"),
+ 'Topology should contain 2 otn links')
+ for link in links:
+ if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
+ self.assertEqual(0, link["available-capacity"]["total-size"]["value"],
+ 'OTU4 link should have an available capacity of 0 Mbps')
+ elif link["name"][0]["value"] == "ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
+ self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
+ 'ODU4 link should have an available capacity of 100 000 Mbps')
+ elif link["name"][0]["value-name"] == "transitional link name":
+ self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
+ 'link should have an available capacity of 100 Gbps')
+ self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
+
+ def test_35_connect_sprda_2_n2_to_roadma_pp3(self):
+ response = test_utils.connect_xpdr_to_rdm_request("SPDR-SA1", "2", "2",
+ "ROADM-A1", "1", "SRG1-PP3-TXRX")
+ self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
+ res = response.json()
+ self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
+ CREATED_SUCCESSFULLY)
+ time.sleep(2)
+
+ def test_36_connect_roadma_pp3_to_spdra_2_n2(self):
+ response = test_utils.connect_rdm_to_xpdr_request("SPDR-SA1", "2", "2",
+ "ROADM-A1", "1", "SRG1-PP3-TXRX")
+ self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
+ res = response.json()
+ self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
+ CREATED_SUCCESSFULLY)
+ time.sleep(2)