diff --git a/tests/test_te_manager.py b/tests/test_te_manager.py index cf1d81b1..d2ad7696 100644 --- a/tests/test_te_manager.py +++ b/tests/test_te_manager.py @@ -1097,3 +1097,78 @@ def test_get_failed_links(self): failed_links = temanager.get_failed_links() self.assertEqual(failed_links, expected_failed_links) + + def test_connection_amlight_to_zaoxi_user_port_any(self): + """ + Exercise a connection request between Amlight and Zaoxi, with + VLAN set to "any". + """ + temanager = TEManager(topology_data=None) + + for path in ( + TestData.TOPOLOGY_FILE_AMLIGHT_USER_PORT, + TestData.TOPOLOGY_FILE_SAX, + TestData.TOPOLOGY_FILE_ZAOXI, + ): + topology = json.loads(path.read_text()) + temanager.add_topology(topology) + + graph = temanager.generate_graph_te() + + connection_request = json.loads( + TestData.CONNECTION_REQ_AMLIGHT_ZAOXI_USER_PORT_v2.read_text() + ) + + # Rewrite the request to have VLAN of "any". + connection_request["endpoints"][0]["vlan"] = "any" + connection_request["endpoints"][1]["vlan"] = "any" + + print(f"connection_request: {connection_request}") + + traffic_matrix = temanager.generate_traffic_matrix(connection_request) + + print(f"Generated graph: '{graph}', traffic matrix: '{traffic_matrix}'") + + self.assertIsNotNone(graph) + self.assertIsNotNone(traffic_matrix) + + conn = temanager.requests_connectivity(traffic_matrix) + print(f"Graph connectivity: {conn}") + + solution = TESolver(graph, traffic_matrix).solve() + print(f"TESolver result: {solution}") + + self.assertIsNotNone(solution.connection_map) + + links = temanager.get_links_on_path(solution) + print(f"Links on path: {links}") + + # Make a flat list of links in connection solution dict, and + # check that we have the same number of links. + values = sum([v for v in solution.connection_map.values()], []) + self.assertEqual(len(links), len(values)) + + breakdown = temanager.generate_connection_breakdown( + solution, connection_request + ) + print(f"breakdown: {json.dumps(breakdown)}") + + # Now assert that the breakdown has everything we need. + zaoxi = breakdown.get("urn:sdx:topology:zaoxi.net") + sax = breakdown.get("urn:sdx:topology:sax.net") + amlight = breakdown.get("urn:sdx:topology:amlight.net") + + for segment in [zaoxi, sax, amlight]: + self.assertIsInstance(segment, dict) + self.assertIsInstance(segment.get("name"), str) + self.assertIsInstance(segment.get("dynamic_backup_path"), bool) + self.assertIsInstance(segment.get("uni_a"), dict) + self.assertIsInstance(segment.get("uni_a").get("tag"), dict) + self.assertIsInstance(segment.get("uni_a").get("tag").get("value"), int) + self.assertIsInstance(segment.get("uni_a").get("tag").get("tag_type"), int) + self.assertIsInstance(segment.get("uni_a").get("port_id"), str) + self.assertIsInstance(segment.get("uni_z"), dict) + self.assertIsInstance(segment.get("uni_z").get("tag"), dict) + self.assertIsInstance(segment.get("uni_z").get("tag").get("value"), int) + self.assertIsInstance(segment.get("uni_z").get("tag").get("tag_type"), int) + self.assertIsInstance(segment.get("uni_z").get("port_id"), str)