feat: add subport's segmentation id check

python/neutron-understack/neutron_understack/config.py

@@ -82,6 +82,15 @@
             "Nautobot."
         ),
     ),
+    cfg.ListOpt(
+        "default_tenant_vlan_id_range",
+        default=[1, 3799],
+        item_type=cfg.types.Integer(min=1, max=4094),
+        help=(
+            "List of 2 comma separated integers, that represents a VLAN range, that"
+            "will be used for mapped VLANs on the switches."
+        ),
+    ),
 ]
 
 l3_svc_cisco_asa_opts = [

python/neutron-understack/neutron_understack/tests/test_trunk.py

@@ -3,6 +3,7 @@
 from oslo_config import cfg
 
 from neutron_understack import utils
+from neutron_understack.trunk import SubportSegmentationIDError
 
 
 class TestSubportsAdded:
@@ -54,18 +55,22 @@ def test_when_subports_are_not_present(
 @pytest.mark.usefixtures("ironic_baremetal_port_physical_network")
 @pytest.mark.usefixtures("utils_fetch_subport_network_id_patch")
 class Test_HandleTenantVlanIDAndSwitchportConfig:
-    def test_when_ucvni_tenant_vlan_id_is_not_set_yet(
+    def test_that_check_subports_segmentation_id_is_called(
         self, mocker, understack_trunk_driver, trunk, subport, network_id, vlan_num
     ):
         mocker.patch("neutron_understack.utils.fetch_port_object")
         mocker.patch(
             "neutron_understack.utils.parent_port_is_bound", return_value=False
         )
-
+        subport_seg_id_check = mocker.patch.object(
+            understack_trunk_driver, "_check_subports_segmentation_id"
+        )
         understack_trunk_driver._handle_tenant_vlan_id_and_switchport_config(
             [subport], trunk
         )
 
+        subport_seg_id_check.assert_called_once()
+
     def test_when_parent_port_is_bound(
         self,
         mocker,
@@ -76,6 +81,7 @@ def test_when_parent_port_is_bound(
         port_id,
         vlan_network_segment,
     ):
+        mocker.patch.object(understack_trunk_driver, "_check_subports_segmentation_id")
         mocker.patch(
             "neutron_understack.utils.fetch_port_object", return_value=port_object
         )
@@ -87,9 +93,13 @@ def test_when_parent_port_is_bound(
             "neutron_understack.utils.network_segment_by_physnet", return_value=None
         )
         mocker.patch("neutron_understack.utils.create_binding_profile_level")
+        add_subports_networks = mocker.patch.object(
+            understack_trunk_driver, "_add_subports_networks_to_parent_port_switchport"
+        )
         understack_trunk_driver._handle_tenant_vlan_id_and_switchport_config(
             [subport], trunk
         )
+        add_subports_networks.assert_called_once()
 
     def test_subports_add_post(
         self,
@@ -114,20 +124,18 @@ def test_subports_add_post(
     def test_when_parent_port_is_unbound(
         self, mocker, understack_trunk_driver, trunk, subport, port_object
     ):
+        mocker.patch.object(understack_trunk_driver, "_check_subports_segmentation_id")
         port_object.bindings[0].vif_type = portbindings.VIF_TYPE_UNBOUND
         mocker.patch(
             "neutron_understack.utils.fetch_port_object", return_value=port_object
         )
-        mocker.patch.object(
+        add_subports_networks = mocker.patch.object(
             understack_trunk_driver, "_add_subports_networks_to_parent_port_switchport"
         )
         understack_trunk_driver._handle_tenant_vlan_id_and_switchport_config(
             [subport], trunk
         )
-
-        (
-            understack_trunk_driver._add_subports_networks_to_parent_port_switchport.assert_not_called()
-        )
+        add_subports_networks.assert_not_called()
 
 
 class TestSubportsDeleted:
@@ -322,3 +330,51 @@ def test_that_handle_subports_removal_is_called(
             subports=[],
             invoke_undersync=False,
         )
+
+
+class TestCheckSubportsSegmentationId:
+    def test_when_trunk_id_is_network_node_trunk_id(
+        self,
+        mocker,
+        understack_trunk_driver,
+        trunk_id,
+    ):
+        mocker.patch(
+            "oslo_config.cfg.CONF.ml2_understack.network_node_trunk_uuid",
+            trunk_id,
+        )
+        result = understack_trunk_driver._check_subports_segmentation_id([], trunk_id)
+        assert result is None
+
+    def test_when_segmentation_id_is_in_allowed_range(
+        self,
+        mocker,
+        understack_trunk_driver,
+        trunk_id,
+        subport,
+    ):
+        allowed_ranges = mocker.patch(
+            "neutron_understack.utils.allowed_tenant_vlan_id_ranges",
+            return_value=[(1, 1500)],
+        )
+        subport.segmentation_id = 500
+        result = understack_trunk_driver._check_subports_segmentation_id(
+            [subport], trunk_id
+        )
+        allowed_ranges.assert_called_once()
+        assert result is None
+
+    def test_when_segmentation_id_is_not_in_allowed_range(
+        self,
+        mocker,
+        understack_trunk_driver,
+        trunk_id,
+        subport,
+    ):
+        mocker.patch(
+            "neutron_understack.utils.allowed_tenant_vlan_id_ranges",
+            return_value=[(1, 1500)],
+        )
+        subport.segmentation_id = 1600
+        with pytest.raises(SubportSegmentationIDError):
+            understack_trunk_driver._check_subports_segmentation_id([subport], trunk_id)

python/neutron-understack/neutron_understack/tests/test_utils.py

@@ -129,3 +129,107 @@ def test_router_interface_false_device_owner_missing(self):
     def test_router_interface_false_device_owner_none(self):
         context = PortContext(current={"device_owner": None})
         assert not utils.is_router_interface(context)
+
+
+class TestMergeOverlappedRanges:
+    def test_single_range(self):
+        assert utils.merge_overlapped_ranges([(1, 5)]) == [(1, 5)]
+
+    def test_no_overlap(self):
+        ranges = [(1, 3), (5, 7), (9, 10)]
+        expected = [(1, 3), (5, 7), (9, 10)]
+        assert utils.merge_overlapped_ranges(ranges) == expected
+
+    def test_simple_overlap(self):
+        ranges = [(1, 4), (2, 5)]
+        expected = [(1, 5)]
+        assert utils.merge_overlapped_ranges(ranges) == expected
+
+    def test_multiple_ranges(self):
+        ranges = [(1, 3), (2, 6), (8, 10), (15, 18)]
+        expected = [(1, 6), (8, 10), (15, 18)]
+        assert utils.merge_overlapped_ranges(ranges) == expected
+
+    def test_touching_ranges(self):
+        ranges = [(1, 4), (5, 7), (8, 10)]
+        expected = [(1, 10)]
+        assert utils.merge_overlapped_ranges(ranges) == expected
+
+    def test_unsorted_input(self):
+        ranges = [(8, 10), (1, 3), (2, 7), (15, 18)]
+        expected = [(1, 10), (15, 18)]
+        assert utils.merge_overlapped_ranges(ranges) == expected
+
+
+class TestFetchGapsInRanges:
+    def test_full_coverage(self):
+        assert utils.fetch_gaps_in_ranges([(1, 10)], (1, 10)) == []
+
+    def test_gap_at_start(self):
+        assert utils.fetch_gaps_in_ranges([(5, 8)], (1, 10)) == [(1, 4), (9, 10)]
+
+    def test_gap_at_end(self):
+        assert utils.fetch_gaps_in_ranges([(1, 6)], (1, 10)) == [(7, 10)]
+
+    def test_gap_in_middle(self):
+        assert utils.fetch_gaps_in_ranges([(1, 2), (5, 10)], (1, 10)) == [(3, 4)]
+
+    def test_multiple_gaps(self):
+        result = utils.fetch_gaps_in_ranges([(1, 2), (4, 5), (8, 8)], (1, 10))
+        assert result == [(3, 3), (6, 7), (9, 10)]
+
+
+class DummyRange:
+    def __init__(self, minimum, maximum):
+        self.minimum = minimum
+        self.maximum = maximum
+
+
+class TestAllowedTenantVlanIdRanges:
+    def test_multiple_non_overlapping_ranges(
+        self,
+        mocker,
+    ):
+        mocker.patch(
+            "oslo_config.cfg.CONF.ml2_understack.default_tenant_vlan_id_range",
+            [1, 2000],
+        )
+        mocker.patch(
+            "neutron_understack.utils.fetch_vlan_network_segment_ranges",
+            return_value=[DummyRange(500, 700), DummyRange(900, 1200)],
+        )
+        expected_result = [(1, 499), (701, 899), (1201, 2000)]
+        result = utils.allowed_tenant_vlan_id_ranges()
+        assert result == expected_result
+
+    def test_multiple_overlapping_ranges(
+        self,
+        mocker,
+    ):
+        mocker.patch(
+            "oslo_config.cfg.CONF.ml2_understack.default_tenant_vlan_id_range",
+            [1, 2000],
+        )
+        mocker.patch(
+            "neutron_understack.utils.fetch_vlan_network_segment_ranges",
+            return_value=[DummyRange(500, 700), DummyRange(600, 1200)],
+        )
+        expected_result = [(1, 499), (1201, 2000)]
+        result = utils.allowed_tenant_vlan_id_ranges()
+        assert result == expected_result
+
+    def test_single_range(
+        self,
+        mocker,
+    ):
+        mocker.patch(
+            "oslo_config.cfg.CONF.ml2_understack.default_tenant_vlan_id_range",
+            [1, 2000],
+        )
+        mocker.patch(
+            "neutron_understack.utils.fetch_vlan_network_segment_ranges",
+            return_value=[DummyRange(500, 700)],
+        )
+        expected_result = [(1, 499), (701, 2000)]
+        result = utils.allowed_tenant_vlan_id_ranges()
+        assert result == expected_result

python/neutron-understack/neutron_understack/trunk.py

@@ -4,6 +4,7 @@
 from neutron.objects.trunk import SubPort
 from neutron.services.trunk.drivers import base as trunk_base
 from neutron.services.trunk.models import Trunk
+from neutron_lib import exceptions as exc
 from neutron_lib.api.definitions import portbindings
 from neutron_lib.callbacks import events
 from neutron_lib.callbacks import registry
@@ -21,6 +22,14 @@
 SUPPORTED_SEGMENTATION_TYPES = (trunk_consts.SEGMENTATION_TYPE_VLAN,)
 
 
+class SubportSegmentationIDError(exc.NeutronException):
+    message = (
+        "Segmentation ID: %(seg_id)s cannot be set to the Subport: "
+        "%(subport_id)s as it falls outside of allowed ranges: "
+        "%(network_segment_ranges)s. Please use different Segmentation ID."
+    )
+
+
 class UnderStackTrunkDriver(trunk_base.DriverBase):
     def __init__(
         self,
@@ -96,13 +105,43 @@ def register(self, resource, event, trigger, payload=None):
     def _handle_tenant_vlan_id_and_switchport_config(
         self, subports: list[SubPort], trunk: Trunk
     ) -> None:
+        self._check_subports_segmentation_id(subports, trunk.id)
         parent_port_obj = utils.fetch_port_object(trunk.port_id)
 
         if utils.parent_port_is_bound(parent_port_obj):
             self._add_subports_networks_to_parent_port_switchport(
                 parent_port_obj, subports
             )
 
+    def _check_subports_segmentation_id(
+        self, subports: list[SubPort], trunk_id: str
+    ) -> None:
+        """Checks if a subport's segmentation_id is within the allowed range.
+
+        A switchport cannot have a mapped VLAN ID equal to the native VLAN ID.
+        Since the user specifies the VLAN ID (segmentation_id) when adding a
+        subport, an error is raised if it falls within any VLAN network segment
+        range, as these ranges are used to allocate VLAN tags for all VLAN
+        segments, including native VLANs.
+
+        The only case where this check is not required is for a network node
+        trunk, since its subport segmentation_ids are the same as the network
+        segment VLAN tags allocated to the subports. Therefore, there is no
+        possibility of conflict with the native VLAN.
+        """
+        if trunk_id == cfg.CONF.ml2_understack.network_node_trunk_uuid:
+            return
+
+        ns_ranges = utils.allowed_tenant_vlan_id_ranges()
+        for subport in subports:
+            seg_id = subport.segmentation_id
+            if not utils.segmentation_id_in_ranges(seg_id, ns_ranges):
+                raise SubportSegmentationIDError(
+                    seg_id=seg_id,
+                    subport_id=subport.port_id,
+                    network_segment_ranges=utils.printable_ranges(ns_ranges),
+                )
+
     def configure_trunk(self, trunk_details: dict, port_id: str) -> None:
         parent_port_obj = utils.fetch_port_object(port_id)
         subports = trunk_details.get("sub_ports", [])
@@ -146,6 +185,7 @@ def _add_subports_networks_to_parent_port_switchport(
         vlan_group_name = self.ironic_client.baremetal_port_physical_network(
             local_link_info
         )
+
         self._handle_segment_allocation(subports, vlan_group_name, binding_host)
 
     def clean_trunk(