add tests for utils and trunk

Author: Milan Fencik

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