Handle VLAN ranges in connection requests

pyproject.toml

@@ -67,12 +67,6 @@ filterwarnings = [
     "ignore:Type has been deprecated in version 0.1.5 and will be removed:DeprecationWarning",
 ]
 
-
-[tool.setuptools_scm]
-# Write version info collected from git to a file.  This happens when
-# we run `python -m build`.
-write_to = "src/sdx_pce/_version.py"
-
 [tool.isort]
 profile = "black"
 src_paths = ["src", "tests", "scripts"]

src/sdx_pce/topology/temanager.py

@@ -1,4 +1,5 @@
 import logging
+import re
 import threading
 from itertools import chain
 from typing import List, Optional
@@ -927,6 +928,18 @@ def _find_common_vlan_on_link(
         )
         return None
 
+    def _tag_is_vlan_range(self, tag: str) -> bool:
+        """
+        Return True if tag is of the form `n:m`
+        """
+        if isinstance(tag, str):
+            return bool(re.match(r"\d+:\d+", tag))
+        else:
+            return False
+
+    def _handle_vlan_range(self, tag):
+        print("HERE HERE HERE")
+
     def _reserve_vlan(
         self,
         domain: str,
@@ -995,6 +1008,29 @@ def _reserve_vlan(
             for vlan_tag, vlan_usage in vlan_table.items():
                 if vlan_usage is UNUSED_VLAN:
                     available_tag = vlan_tag
+        elif self._tag_is_vlan_range(tag):
+            # expand the range.
+            start, end = map(int, tag.split(":"))
+            vlans = list(range(start, end + 1))
+
+            self._logger.debug(f"Attempting to reseve vlan range {vlans}")
+
+            # Check if all VLANs in the range are available.
+            for vlan in vlans:
+                if vlan_table[vlan] is not UNUSED_VLAN:
+                    raise Exception(f"VLAN {vlan} is in use; can't reserve {tag} range")
+
+            # Mark range in use.
+            for vlan in vlans:
+                vlan_table[vlan] = request_id
+
+            self._logger.debug(
+                f"reserve_vlan domain {domain}, after reservation: "
+                f"vlan_table: {vlan_table}, requested range: {tag}"
+            )
+
+            # Return the whole list to indicate success.
+            return vlans
         else:
             if tag in vlan_table:
                 if vlan_table[tag] is UNUSED_VLAN:

tests/__init__.py

@@ -1,12 +1,6 @@
 import pathlib
 import tempfile
-
-try:
-    # Use stdlib modules with Python > 3.8.
-    from importlib.resources import files
-except ImportError:
-    # Use compatibility library with Python 3.8.
-    from importlib_resources import files
+from importlib.resources import files
 
 
 class TestData:

tests/test_te_manager.py

@@ -1519,11 +1519,67 @@ def test_disallowed_vlan(self):
 
         # Although we have a solution (in terms of connectivity
         # between ports), we should not have a breakdown at this
-        # point, because we asked for a port that is not present on
-        # the port.
+        # point, because we asked for a VLAN tag that is not present
+        # on the port.
         with self.assertRaises(TEError):
             self.temanager.generate_connection_breakdown(solution, connection_request)
 
+    def test_vlan_range(self):
+        """
+        Test when requests are for a range like [n:m]
+        """
+
+        connection_request = json.loads(
+            """
+            {
+                "name": "new-connection",            
+                "id": "test-connection-id",
+                "endpoints": [
+                    {
+                        "port_id": "urn:sdx:port:amlight.net:A1:1",
+                        "vlan": "100:200"
+                    },
+                    {
+                        "port_id": "urn:sdx:port:amlight:B1:1",
+                        "vlan": "100:200"
+                    }
+                ]
+            }
+            """
+        )
+
+        temanager = TEManager(topology_data=None)
+
+        temanager.add_topology(
+            json.loads(TestData.TOPOLOGY_FILE_AMLIGHT_USER_PORT.read_text())
+        )
+
+        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)
+
+        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)
+
+        breakdown = self.temanager.generate_connection_breakdown(
+            solution, connection_request
+        )
+        pprint.pprint(f"Breakdown: {breakdown}")
+
+        self.assertIsNotNone(breakdown)
+        self.assertIsInstance(breakdown, dict)
+        self.assertEqual(len(breakdown), 1)
+
     def _vlan_meets_request(self, requested_vlan: str, assigned_vlan: int) -> bool:
         """
         A helper to compare requested VLAN against the VLAN assignment