Merge pull request #244 from atlanticwave-sdx/215.restore-vlan-reserv…

…ations

Save/restore vlan reservations

src/sdx_pce/topology/temanager.py

@@ -127,6 +127,47 @@ def get_connections(self) -> List[ConnectionRequest]:
             connections.append(solution.request_id)
         return connections
 
+    @property
+    def vlan_tags_table(self) -> dict:
+        """
+        Return the current VLAN tags table.
+        """
+        return self._vlan_tags_table
+
+    @vlan_tags_table.setter
+    def vlan_tags_table(self, table: dict):
+        """
+        Set VLAN tags table.
+        """
+        # Ensure that the input is in correct shape.
+        if not isinstance(table, dict):
+            raise ValidationError(f"table ({table}) is not a dict")
+
+        for domain, ports in table.items():
+            if not isinstance(domain, str):
+                raise ValidationError(f"domain ({domain}) is not a str")
+
+            for port_id, labels in ports.items():
+                if not isinstance(port_id, str):
+                    raise ValidationError(f"port_id ({port_id}) is not a str")
+
+                if not isinstance(labels, dict):
+                    raise ValidationError(f"labels ({labels}) is not a dict")
+
+        # We should allow VLAN table to be restored only during
+        # startup.  If the table has VLANs that are in use, it means
+        # that we're in the wrong state.
+        for domain, ports in self._vlan_tags_table.items():
+            for port_id, labels in ports.items():
+                for vlan, status in labels.items():
+                    if status is not UNUSED_VLAN:
+                        raise ValidationError(
+                            f"Error: VLAN table is not empty:"
+                            f"(domain: {domain}, port: {port_id}, vlan: {vlan})"
+                        )
+
+        self._vlan_tags_table = table
+
     def _update_vlan_tags_table(self, domain_name: str, port_map: dict):
         """
         Update VLAN tags table in a non-disruptive way, meaning: only add new
@@ -1023,8 +1064,8 @@ def _reserve_vlan(
         domain: str,
         port: dict,
         request_id: str,
-        tag: str = None,
-        upstream_egress_vlan: str = None,
+        tag: Optional[str] = None,
+        upstream_egress_vlan: Optional[str] = None,
     ):
         """
         Find unused VLANs for given domain/port and mark them in-use.
@@ -1170,7 +1211,7 @@ def delete_connection(self, request_id: str):
         # Now it is the time to update the bandwidth of the links after breakdowns are successfully generated
         self.update_link_bandwidth(solution, reduce=False)
 
-    def get_connection_solution(self, request_id: str) -> ConnectionSolution:
+    def get_connection_solution(self, request_id: str) -> Optional[ConnectionSolution]:
         """
         Get a connection solution by request ID.
         """

tests/test_te_manager.py

@@ -7,7 +7,7 @@
 from sdx_pce.load_balancing.te_solver import TESolver
 from sdx_pce.models import ConnectionRequest, ConnectionSolution, TrafficMatrix
 from sdx_pce.topology.temanager import TEManager
-from sdx_pce.utils.exceptions import TEError, UnknownRequestError
+from sdx_pce.utils.exceptions import TEError, UnknownRequestError, ValidationError
 
 from . import TestData
 
@@ -1815,3 +1815,106 @@ def _vlan_meets_request(self, requested_vlan: str, assigned_vlan: int) -> bool:
             return requested_vlan in requested_range
 
         raise Exception("invalid state!")
+
+    def test_vlan_tags_table(self):
+        """
+        Test saving/restoring VLAN tags table.
+        """
+        temanager = TEManager(topology_data=None)
+
+        for topology_file in [
+            TestData.TOPOLOGY_FILE_AMLIGHT_v2,
+            TestData.TOPOLOGY_FILE_ZAOXI_v2,
+            TestData.TOPOLOGY_FILE_SAX_v2,
+        ]:
+            temanager.add_topology(json.loads(topology_file.read_text()))
+
+        # Test getter.
+        table1 = temanager.vlan_tags_table
+        self.assertIsInstance(table1, dict)
+
+        # Test setter
+        temanager.vlan_tags_table = table1
+        table2 = temanager.vlan_tags_table
+        self.assertIsInstance(table2, dict)
+        self.assertEqual(table1, table2)
+
+    def test_vlan_tags_table_error_checks(self):
+        """
+        Test error checks when restoring VLAN tags table.
+        """
+        temanager = TEManager(topology_data=None)
+
+        for topology_file in [
+            TestData.TOPOLOGY_FILE_AMLIGHT_v2,
+            TestData.TOPOLOGY_FILE_ZAOXI_v2,
+            TestData.TOPOLOGY_FILE_SAX_v2,
+        ]:
+            temanager.add_topology(json.loads(topology_file.read_text()))
+
+        # input must be a dict.
+        with self.assertRaises(ValidationError) as ctx:
+            temanager.vlan_tags_table = list()
+            self.assertTrue("table ([]) is not a dict" in str(ctx.exception))
+
+        # port_id keys in the input must be a string.
+        with self.assertRaises(ValidationError) as ctx:
+            temanager.vlan_tags_table = {"domain1": {1: {1: None}}}
+            self.assertTrue("port_id (1) is not a str" in str(ctx.exception))
+
+        # the "inner" VLAN allocation table must be a dict.
+        with self.assertRaises(ValidationError) as ctx:
+            temanager.vlan_tags_table = {"domain1": {"port1": (1, None)}}
+            self.assertTrue("labels ((1, None)) is not a dict" in str(ctx.exception))
+
+    def test_vlan_tags_table_ensure_no_existing_allocations(self):
+        """
+        Test that restoring VLAN tables works when there are no
+        existing allocations.
+        """
+        temanager = TEManager(topology_data=None)
+
+        for topology_file in [
+            TestData.TOPOLOGY_FILE_AMLIGHT_v2,
+            TestData.TOPOLOGY_FILE_ZAOXI_v2,
+            TestData.TOPOLOGY_FILE_SAX_v2,
+        ]:
+            temanager.add_topology(json.loads(topology_file.read_text()))
+
+        connection_request = {
+            "name": "check-existing-vlan-allocations",
+            "id": "id-check-existing-vlan-allocations",
+            "endpoints": [
+                {"port_id": "urn:sdx:port:ampath.net:Ampath1:50", "vlan": "100:200"},
+                {"port_id": "urn:sdx:port:tenet.ac.za:Tenet01:1", "vlan": "100:200"},
+            ],
+        }
+
+        graph = temanager.generate_graph_te()
+        traffic_matrix = temanager.generate_traffic_matrix(connection_request)
+
+        print(f"Generated graph: '{graph}', traffic matrix: '{traffic_matrix}'")
+        self.assertIsNotNone(graph)
+        self.assertIsNotNone(traffic_matrix)
+
+        solution = TESolver(graph, traffic_matrix).solve()
+        self.assertIsNotNone(solution)
+
+        print(f"TESolver result: {solution}")
+        breakdown = temanager.generate_connection_breakdown(
+            solution, connection_request
+        )
+
+        print(f"Breakdown: {breakdown}")
+        self.assertIsNotNone(breakdown)
+
+        # The VLAN table should have some allocations now, and we
+        # should not be able to change its state.
+        with self.assertRaises(ValidationError) as ctx:
+            temanager.vlan_tags_table = {"domain1": {"port1": {1: None}}}
+            expected_error = (
+                "Error: VLAN table is not empty:"
+                "(domain: urn:sdx:topology:ampath.net, port: "
+                "urn:sdx:port:ampath.net:Ampath1:40, vlan: 100)"
+            )
+            self.assertTrue(expected_error in str(ctx.exception))