Merge "ovn: support max_tunid option from NB_Global"

neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py

@@ -412,7 +412,15 @@ class OVNMechanismDriver(api.MechanismDriver):
                                  const.TYPE_VXLAN,
                                  const.TYPE_VLAN])
 
+    def _get_max_tunid(self):
+        try:
+            return int(self._nb_ovn.nb_global.options.get('max_tunid'))
+        except (ValueError, TypeError):
+            # max_tunid may be absent in older OVN versions, return None
+            pass
+
     def _validate_network_segments(self, network_segments):
+        max_tunid = self._get_max_tunid()
         for network_segment in network_segments:
             network_type = network_segment['network_type']
             segmentation_id = network_segment['segmentation_id']
@@ -427,6 +435,12 @@ class OVNMechanismDriver(api.MechanismDriver):
             if not self._is_network_type_supported(network_type):
                 msg = _('Network type %s is not supported') % network_type
                 raise n_exc.InvalidInput(error_message=msg)
+            if segmentation_id and max_tunid and segmentation_id > max_tunid:
+                m = (
+                    _('Segmentation ID should be lower or equal to %d') %
+                    max_tunid
+                )
+                raise n_exc.InvalidInput(error_message=m)
 
     def create_segment_provnet_port(self, resource, event, trigger,
                                     context, segment, payload=None):

neutron/tests/unit/fake_resources.py

@@ -130,6 +130,7 @@ class FakeOvsdbNbOvnIdl(object):
         self.pg_del_ports = mock.Mock()
         self.lsp_get_up = mock.Mock()
         self.nb_global = mock.Mock()
+        self.nb_global.options.get.return_value = '100000'
         self.db_list_rows = mock.Mock()
         self.lsp_list = mock.MagicMock()
         self.db_find = mock.Mock()

neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py

@@ -155,6 +155,38 @@ class TestOVNMechanismDriver(test_plugin.Ml2PluginV2TestCase):
                           self.mech_driver._create_neutron_pg_drop)
         self.assertEqual(2, m_ovsdb_api.get_port_group.call_count)
 
+    def test__get_max_tunid_no_key_set(self):
+        self.mech_driver._nb_ovn.nb_global.options.get.return_value = None
+        self.assertIsNone(self.mech_driver._get_max_tunid())
+
+    def test__get_max_tunid_wrong_key_value(self):
+        self.mech_driver._nb_ovn.nb_global.options.get.return_value = '11wrong'
+        self.assertIsNone(self.mech_driver._get_max_tunid())
+
+    def test__get_max_tunid_key_set(self):
+        self.mech_driver._nb_ovn.nb_global.options.get.return_value = '100'
+        self.assertEqual(100, self.mech_driver._get_max_tunid())
+
+    def _test__validate_network_segments_id_succeed(self, val):
+        segment = {
+            "network_type": const.TYPE_VXLAN,
+            "segmentation_id": val,
+            "physical_network": "physnet1",
+        }
+        self.mech_driver._nb_ovn.nb_global.options.get.return_value = '200'
+        self.mech_driver._validate_network_segments([segment])
+
+    def test__validate_network_segments_id_below_max_limit(self):
+        self._test__validate_network_segments_id_succeed(100)
+
+    def test__validate_network_segments_id_eq_max_limit(self):
+        self._test__validate_network_segments_id_succeed(200)
+
+    def test__validate_network_segments_id_above_max_limit(self):
+        self.assertRaises(
+            n_exc.InvalidInput,
+            self._test__validate_network_segments_id_succeed, 300)
+
     @mock.patch.object(ovn_revision_numbers_db, 'bump_revision')
     def test__create_security_group(self, mock_bump):
         self.mech_driver._create_security_group(