Fix tunnel network detection to use tunnel type driver ranges

The ``_is_tunnel_project_network`` method in ``DNSExtensionDriverML2``
was only checking the static tunnel ranges from ml2_conf.ini, ignoring
the dynamic ranges provided by the ``network-segment-range service``
plugin. When segment ranges are managed via the API, external DNS
eligibility for tenant tunnel networks could be wrong.

Update the method to use the ML2 tunnel type driver's
``get_network_segment_ranges()``, which handles both static
configuration and DB-based ranges when the network-segment-range plugin
is enabled.

Conflicts:
  neutron/plugins/ml2/extensions/dns_integration.py

Closes-Bug: #2154266
Related-Bug: #2140291
Assisted-By: Cursor Composer 2.5
Signed-off-by: Rodolfo Alonso Hernandez <ralonsoh@redhat.com>
Change-Id: I5d96efa626d18b7589cf2563fdac3a8d399a69a8
(cherry picked from commit 6913303)
(cherry picked from commit 1941988)
(cherry picked from commit d04e845)

Author: ralonsoh

neutron/plugins/ml2/extensions/dns_integration.py

@@ -351,35 +351,42 @@ class DNSExtensionDriverML2(DNSExtensionDriver):
     def __init__(self):
         super().__init__()
         self._vlan_driver = None
+        self._tunnel_drivers = {}
         self._plugin = None
 
     def initialize(self):
         LOG.info("DNSExtensionDriverML2 initialization complete")
 
+    @property
+    def plugin(self):
+        if not self._plugin:
+            self._plugin = directory.get_plugin()
+        return self._plugin
+
     @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(
+            self._vlan_driver = self.plugin.type_manager.drivers.get(
                 lib_const.TYPE_VLAN)
         return self._vlan_driver
 
-    def _is_tunnel_tenant_network(self, provider_net):
-        if provider_net['network_type'] == lib_const.TYPE_GENEVE:
-            tunnel_ranges = cfg.CONF.ml2_type_geneve.vni_ranges
-        elif provider_net['network_type'] == lib_const.TYPE_VXLAN:
-            tunnel_ranges = cfg.CONF.ml2_type_vxlan.vni_ranges
-        else:
-            tunnel_ranges = cfg.CONF.ml2_type_gre.tunnel_id_ranges
+    def get_tunnel_driver(self, network_type):
+        if network_type not in self._tunnel_drivers:
+            self._tunnel_drivers[network_type] = (
+                self.plugin.type_manager.drivers.get(network_type))
+        return self._tunnel_drivers[network_type]
 
+    def _is_tunnel_project_network(self, provider_net):
+        network_type = provider_net['network_type']
+        tunnel_driver = self.get_tunnel_driver(network_type)
+        if not tunnel_driver:
+            return False
+        tunnel_ranges = tunnel_driver.obj.get_network_segment_ranges()
+        if not tunnel_ranges:
+            return False
         segmentation_id = int(provider_net['segmentation_id'])
-        for entry in tunnel_ranges:
-            entry = entry.strip()
-            tun_min, tun_max = entry.split(':')
-            tun_min = tun_min.strip()
-            tun_max = tun_max.strip()
-            return int(tun_min) <= segmentation_id <= int(tun_max)
+        return any(tun_min <= segmentation_id <= tun_max
+                   for tun_min, tun_max in tunnel_ranges)
 
     def _is_vlan_tenant_network(self, provider_net):
         if not self.vlan_driver:
@@ -414,7 +421,7 @@ def external_dns_not_needed(self, context, network, subnets):
         if provider_net['network_type'] in [
                 lib_const.TYPE_GRE, lib_const.TYPE_VXLAN,
                 lib_const.TYPE_GENEVE]:
-            return self._is_tunnel_tenant_network(provider_net)
+            return self._is_tunnel_project_network(provider_net)
         return True
 
 

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

@@ -864,6 +864,86 @@ def test__is_vlan_tenant_network_no_vlan_driver(self):
         }
         self.assertFalse(self.driver._is_vlan_tenant_network(provider_net))
 
+    def _mock_tunnel_driver(self, network_type, ranges):
+        mock_tunnel_driver = mock.Mock()
+        mock_tunnel_driver.obj.get_network_segment_ranges.return_value = ranges
+        return mock.patch.object(
+            self.driver, 'get_tunnel_driver',
+            return_value=mock_tunnel_driver)
+
+    def test__is_tunnel_project_network_with_multiple_ranges(self):
+        with self._mock_tunnel_driver(
+                constants.TYPE_VXLAN, [(100, 200), (300, 400)]):
+            provider_net_in_range = {
+                'network_type': constants.TYPE_VXLAN,
+                'segmentation_id': 150,
+            }
+            self.assertTrue(
+                self.driver._is_tunnel_project_network(provider_net_in_range))
+
+            provider_net_between_ranges = {
+                'network_type': constants.TYPE_VXLAN,
+                'segmentation_id': 250,
+            }
+            self.assertFalse(self.driver._is_tunnel_project_network(
+                provider_net_between_ranges))
+
+            provider_net_second_range = {
+                'network_type': constants.TYPE_VXLAN,
+                'segmentation_id': 350,
+            }
+            self.assertTrue(self.driver._is_tunnel_project_network(
+                provider_net_second_range))
+
+    def test__is_tunnel_project_network_boundary_values(self):
+        with self._mock_tunnel_driver(constants.TYPE_GENEVE, [(100, 200)]):
+            provider_net_min = {
+                'network_type': constants.TYPE_GENEVE,
+                'segmentation_id': 100,
+            }
+            self.assertTrue(
+                self.driver._is_tunnel_project_network(provider_net_min))
+
+            provider_net_max = {
+                'network_type': constants.TYPE_GENEVE,
+                'segmentation_id': 200,
+            }
+            self.assertTrue(
+                self.driver._is_tunnel_project_network(provider_net_max))
+
+            provider_net_below = {
+                'network_type': constants.TYPE_GENEVE,
+                'segmentation_id': 99,
+            }
+            self.assertFalse(
+                self.driver._is_tunnel_project_network(provider_net_below))
+
+            provider_net_above = {
+                'network_type': constants.TYPE_GENEVE,
+                'segmentation_id': 201,
+            }
+            self.assertFalse(
+                self.driver._is_tunnel_project_network(provider_net_above))
+
+    def test__is_tunnel_project_network_empty_ranges(self):
+        with self._mock_tunnel_driver(constants.TYPE_GRE, []):
+            provider_net = {
+                'network_type': constants.TYPE_GRE,
+                'segmentation_id': 100,
+            }
+            self.assertFalse(
+                self.driver._is_tunnel_project_network(provider_net))
+
+    def test__is_tunnel_project_network_no_tunnel_driver(self):
+        with mock.patch.object(
+                self.driver, 'get_tunnel_driver', return_value=None):
+            provider_net = {
+                'network_type': constants.TYPE_VXLAN,
+                'segmentation_id': 100,
+            }
+            self.assertFalse(
+                self.driver._is_tunnel_project_network(provider_net))
+
 
 class TestDesignateClientKeystoneV3(testtools.TestCase):
     """Test case for designate clients """