From 2cb4ee9215f28d3bc151f4254ed914fcf2dc0cf0 Mon Sep 17 00:00:00 2001 From: Sajith Sasidharan Date: Wed, 21 Aug 2024 10:43:10 -0500 Subject: [PATCH 1/2] Add test for when VLAN requested is "any" --- tests/test_te_manager.py | 76 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 76 insertions(+) diff --git a/tests/test_te_manager.py b/tests/test_te_manager.py index cf1d81b1..08f0a575 100644 --- a/tests/test_te_manager.py +++ b/tests/test_te_manager.py @@ -1097,3 +1097,79 @@ 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) + From a55c4f4144bfcc2a1f01401f02763e9b44c44551 Mon Sep 17 00:00:00 2001 From: Sajith Sasidharan Date: Wed, 21 Aug 2024 10:45:08 -0500 Subject: [PATCH 2/2] Reformat --- tests/test_te_manager.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/tests/test_te_manager.py b/tests/test_te_manager.py index 08f0a575..d2ad7696 100644 --- a/tests/test_te_manager.py +++ b/tests/test_te_manager.py @@ -1097,7 +1097,7 @@ 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 @@ -1122,9 +1122,9 @@ def test_connection_amlight_to_zaoxi_user_port_any(self): # 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}'") @@ -1172,4 +1172,3 @@ def test_connection_amlight_to_zaoxi_user_port_any(self): 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) -