diff --git a/sdx_controller/test/__init__.py b/sdx_controller/test/__init__.py index c703327..90ecf8f 100644 --- a/sdx_controller/test/__init__.py +++ b/sdx_controller/test/__init__.py @@ -42,3 +42,7 @@ class TestData: CONNECTION_REQ_V2_AMLIGHT_ZAOXI = ( REQUESTS_DIR / "test_request-amlight_zaoxi-p2p-v2.json" ) + + TOPOLOGY_FILE_AMLIGHT_v2 = TOPOLOGY_DIR / "ampath_v2.json" + TOPOLOGY_FILE_SAX_v2 = TOPOLOGY_DIR / "sax_v2.json" + TOPOLOGY_FILE_ZAOXI_v2 = TOPOLOGY_DIR / "zaoxi_v2.json" diff --git a/sdx_controller/test/test_l2vpn_controller.py b/sdx_controller/test/test_l2vpn_controller.py index c344401..69d4758 100644 --- a/sdx_controller/test/test_l2vpn_controller.py +++ b/sdx_controller/test/test_l2vpn_controller.py @@ -48,6 +48,21 @@ def __add_the_three_topologies(self): print(f"Adding topology: {topology.get('id')}") self.te_manager.add_topology(topology) + def __add_the_three_v2_topologies(self): + """ + A helper to add the three known topologies. + """ + for idx, topology_file in enumerate( + [ + TestData.TOPOLOGY_FILE_AMLIGHT_v2, + TestData.TOPOLOGY_FILE_SAX_v2, + TestData.TOPOLOGY_FILE_ZAOXI_v2, + ] + ): + topology = json.loads(topology_file.read_text()) + print(f"Adding topology: {topology.get('id')}") + self.te_manager.add_topology(topology) + def test_delete_connection_with_setup(self): """ Test case for delete_connection() @@ -570,6 +585,33 @@ def test_z105_getconnections_success(self): assert len(response.get_json()) != 0 + def test_issue_356(self): + """ + See https://github.com/atlanticwave-sdx/sdx-controller/issues/356 + """ + + self.__add_the_three_v2_topologies() + + connection_request = { + "name": "VLAN between AMPATH/300 and TENET/300", + "endpoints": [ + {"port_id": "urn:sdx:port:ampath.net:Ampath3:50", "vlan": "300"}, + {"port_id": "urn:sdx:port:tenet.ac.za:Tenet03:50", "vlan": "any"}, + ], + } + + response = self.client.open( + f"{BASE_PATH}/l2vpn/1.0", + method="POST", + data=json.dumps(connection_request), + content_type="application/json", + ) + + print(f"POST response body is : {response.data.decode('utf-8')}") + print(f"POST Response JSON is : {response.get_json()}") + + # self.assertStatus(response, 200) + if __name__ == "__main__": unittest.main()