Fix VLAN network detection to use VlanTypeDriver ranges

The _is_vlan_project_network method in DNSExtensionDriverML2 was only
checking the static network_vlan_ranges configuration from ml2_conf.ini,
ignoring the dynamic ranges provided by the network-segment-range
service plugin. This caused a KeyError when creating subnets or VMs on
VLAN networks with segment ranges managed via the API.

This fix updates the method to use
VlanTypeDriver.get_network_segment_ranges(), which properly handles both
static configuration and DB-based dynamic ranges when the
network-segment-range plugin is enabled.

Closes-Bug: #2140291
Signed-off-by: Kyuyeong Lee <kyu0.lee@samsung.com>
Change-Id: I8f3a2b1c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a
(cherry picked from commit cd8f88e)

Author: kyu0

neutron/plugins/ml2/extensions/dns_integration.py

@@ -23,7 +23,6 @@
 from neutron_lib.exceptions import dns as dns_exc
 from neutron_lib.plugins import directory
 from neutron_lib.plugins.ml2 import api
-from neutron_lib.plugins import utils as plugin_utils
 from oslo_config import cfg
 from oslo_log import log as logging
 
@@ -349,9 +348,23 @@ def _get_details(self, context, network_id):
 
 class DNSExtensionDriverML2(DNSExtensionDriver):
 
+    def __init__(self):
+        super().__init__()
+        self._vlan_driver = None
+        self._plugin = None
+
     def initialize(self):
         LOG.info("DNSExtensionDriverML2 initialization complete")
 
+    @property
+    def vlan_driver(self):
+        if not self._vlan_driver:
+            if not self._plugin:
+                self._plugin = directory.get_plugin()
+            self._vlan_driver = self._plugin.type_manager.drivers.get(
+                lib_const.TYPE_VLAN)
+        return self._vlan_driver
+
     def _is_tunnel_project_network(self, provider_net):
         if provider_net['network_type'] == lib_const.TYPE_GENEVE:
             tunnel_ranges = cfg.CONF.ml2_type_geneve.vni_ranges
@@ -369,15 +382,15 @@ def _is_tunnel_project_network(self, provider_net):
             return int(tun_min) <= segmentation_id <= int(tun_max)
 
     def _is_vlan_project_network(self, provider_net):
-        network_vlan_ranges = plugin_utils.parse_network_vlan_ranges(
-            cfg.CONF.ml2_type_vlan.network_vlan_ranges)
-        vlan_ranges = network_vlan_ranges[provider_net['physical_network']]
+        if not self.vlan_driver:
+            return False
+        network_vlan_ranges = self.vlan_driver.obj.get_network_segment_ranges()
+        vlan_ranges = network_vlan_ranges.get(provider_net['physical_network'])
         if not vlan_ranges:
             return False
         segmentation_id = int(provider_net['segmentation_id'])
-        for vlan_range in vlan_ranges:
-            if vlan_range[0] <= segmentation_id <= vlan_range[1]:
-                return True
+        return any(vlan_range[0] <= segmentation_id <= vlan_range[1]
+                   for vlan_range in vlan_ranges)
 
     def external_dns_not_needed(self, context, network, subnets):
         dns_driver = _get_dns_driver()

neutron/tests/unit/plugins/ml2/extensions/test_dns_integration.py

@@ -760,6 +760,111 @@ def test_update_fixed_ips_no_effect_after_clearing_dns_domain(self,
             dns_data_db, dns_data_db_1, dns_data_db_2)
 
 
+class DNSExtensionDriverML2TestCase(testtools.TestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.driver = dns_integration.DNSExtensionDriverML2()
+        self.mock_vlan_driver = mock.Mock()
+        self.mock_vlan_driver_property = mock.patch.object(
+            type(self.driver), 'vlan_driver',
+            new_callable=mock.PropertyMock,
+            return_value=self.mock_vlan_driver).start()
+
+    def test__is_vlan_project_network_with_multiple_ranges(self):
+        self.mock_vlan_driver.obj.get_network_segment_ranges.return_value = {
+            'physnet1': [(100, 200), (300, 400)],
+            'physnet2': [(500, 600)]
+        }
+
+        provider_net_1 = {
+            'physical_network': 'physnet1',
+            'segmentation_id': 150
+        }
+        self.assertTrue(self.driver._is_vlan_project_network(provider_net_1))
+
+        provider_net_2 = {
+            'physical_network': 'physnet1',
+            'segmentation_id': 250
+        }
+        self.assertFalse(self.driver._is_vlan_project_network(provider_net_2))
+
+        provider_net_3 = {
+            'physical_network': 'physnet1',
+            'segmentation_id': 350
+        }
+        self.assertTrue(self.driver._is_vlan_project_network(provider_net_3))
+
+        provider_net_4 = {
+            'physical_network': 'physnet2',
+            'segmentation_id': 550
+        }
+        self.assertTrue(self.driver._is_vlan_project_network(provider_net_4))
+
+        provider_net_5 = {
+            'physical_network': 'physnet2',
+            'segmentation_id': 650
+        }
+        self.assertFalse(self.driver._is_vlan_project_network(provider_net_5))
+
+    def test__is_vlan_project_network_boundary_values(self):
+        self.mock_vlan_driver.obj.get_network_segment_ranges.return_value = {
+            'physnet1': [(100, 200)]
+        }
+
+        provider_net_1 = {
+            'physical_network': 'physnet1',
+            'segmentation_id': 100
+        }
+        self.assertTrue(self.driver._is_vlan_project_network(provider_net_1))
+
+        provider_net_2 = {
+            'physical_network': 'physnet1',
+            'segmentation_id': 200
+        }
+        self.assertTrue(self.driver._is_vlan_project_network(provider_net_2))
+
+        provider_net_3 = {
+            'physical_network': 'physnet1',
+            'segmentation_id': 99
+        }
+        self.assertFalse(self.driver._is_vlan_project_network(provider_net_3))
+
+        provider_net_4 = {
+            'physical_network': 'physnet1',
+            'segmentation_id': 201
+        }
+        self.assertFalse(self.driver._is_vlan_project_network(provider_net_4))
+
+    def test__is_vlan_project_network_physnet_not_found(self):
+        self.mock_vlan_driver.obj.get_network_segment_ranges.return_value = {
+            'physnet1': [(100, 200)]
+        }
+
+        provider_net = {
+            'physical_network': 'physnet2',
+            'segmentation_id': 150
+        }
+        self.assertFalse(self.driver._is_vlan_project_network(provider_net))
+
+    def test__is_vlan_project_network_empty_ranges(self):
+        self.mock_vlan_driver.obj.get_network_segment_ranges.return_value = {}
+
+        provider_net = {
+            'physical_network': 'physnet1',
+            'segmentation_id': 100
+        }
+        self.assertFalse(self.driver._is_vlan_project_network(provider_net))
+
+    def test__is_vlan_project_network_no_vlan_driver(self):
+        self.mock_vlan_driver_property.return_value = None
+        provider_net = {
+            'physical_network': 'physnet1',
+            'segmentation_id': 100
+        }
+        self.assertFalse(self.driver._is_vlan_project_network(provider_net))
+
+
 class TestDesignateClientKeystoneV3(testtools.TestCase):
     """Test case for designate clients """