Remove VPN specific exception

This exception is an overkill, and can be safely removed. The tests
affected were not designed to cover any regression as they were asserting
that the mocked Exception was being raised, defeating the very purpose of
catching the regression; they have been revised to ensure that the checks
are not misplaced in future revision of the code, or that they behave the
way they are supposed to.

Partially-Implements: blueprint services-split
Depends-On: https://review.openstack.org/#/c/153543/

Change-Id: I22cc8cd4383259dd4ee20bcbc041d589172d88c8

diff --git a/neutron/common/exceptions.py b/neutron/common/exceptions.py
index b65810d1f5aafd19af19e909ded14819fa3ac972..35beac9a6159eb9382fad31699bdf1d9b1672cb4 100644 (file)
--- a/neutron/common/exceptions.py
+++ b/neutron/common/exceptions.py
@@ -383,7 +383,3 @@ class FirewallInternalDriverError(NeutronException):
     raise this exception to the agent
     """
     message = _("%(driver)s: Internal driver error.")
-
-
-class RouterInUseByVPNService(InUse):
-    message = _("Router %(router_id)s is used by VPNService %(vpnservice_id)s")

diff --git a/neutron/extensions/l3.py b/neutron/extensions/l3.py
index 6776035f7f0bd76ee6c53423ef0c255dfe9bda70..ac9b07d522c3e7e1eba596a5d8bb9bcea17c84c7 100644 (file)
--- a/neutron/extensions/l3.py
+++ b/neutron/extensions/l3.py
@@ -30,7 +30,12 @@ class RouterNotFound(nexception.NotFound):
 
 
 class RouterInUse(nexception.InUse):
-    message = _("Router %(router_id)s still has ports")
+    message = _("Router %(router_id)s %(reason)s")
+
+    def __init__(self, **kwargs):
+        if 'reason' not in kwargs:
+            kwargs['reason'] = "still has ports"
+        super(RouterInUse, self).__init__(**kwargs)
 
 
 class RouterInterfaceNotFound(nexception.NotFound):

diff --git a/neutron/tests/unit/db/test_l3_dvr_db.py b/neutron/tests/unit/db/test_l3_dvr_db.py
index 7a6589c31aec8359bb27aa557b40c3a95e57de9b..943c622935cd86b6526fed320a17900519352930 100644 (file)
--- a/neutron/tests/unit/db/test_l3_dvr_db.py
+++ b/neutron/tests/unit/db/test_l3_dvr_db.py
@@ -17,7 +17,6 @@ import contextlib
 import mock
 
 from neutron.common import constants as l3_const
-from neutron.common import exceptions as nexception
 from neutron import context
 from neutron.db import l3_dvr_db
 from neutron.extensions import l3
@@ -332,77 +331,43 @@ class L3DvrTestCase(testlib_api.SqlTestCase):
                 self.ctx, fip, floatingip, mock.ANY)
             self.assertTrue(vf.called)
 
-    def _router_migration_with_services_setup(
-        self, test_side_effect_vpn=None, test_side_effect_fw=None,
-        no_vpn=None, no_fw=None):
-        '''Helper function to test router migration with services.'''
-        router = {'name': 'foo_router',
-                  'admin_state_up': True}
+    def test__validate_router_migration_prevent_check_advanced_svc(self):
+        router = {'name': 'foo_router', 'admin_state_up': True}
         router_db = self._create_router(router)
-        self.assertFalse(router_db.extra_attributes.distributed)
+        # make sure the check are invoked, whether they pass or
+        # raise, it does not matter in the context of this test
         with contextlib.nested(
-            mock.patch.object(
-                self.mixin, 'check_router_has_no_vpnaas',
-                side_effect=test_side_effect_vpn, return_value=no_vpn),
-            mock.patch.object(
-                self.mixin, 'check_router_has_no_firewall',
-                side_effect=test_side_effect_fw, return_value=no_fw),
-            mock.patch.object(self.mixin, '_get_router',
-                              return_value=router_db)
-                              ) as (vpn_mock, fw_mock, r_mock):
-            router_db['status'] = 'ACTIVE'
-            if no_vpn and no_fw and (
-                test_side_effect_vpn and test_side_effect_fw) is None:
-                self.mixin._validate_router_migration(
-                    self.ctx, router_db, {'distributed': True})
-                return router_db, fw_mock, vpn_mock
-            if not no_vpn and test_side_effect_vpn:
-                self.assertRaises(
-                    nexception.RouterInUseByVPNService,
-                    self.mixin._validate_router_migration,
-                    self.ctx,
-                    router_db,
-                    {'distributed': True})
-                return router_db, fw_mock, vpn_mock
-            if not no_fw and test_side_effect_fw:
-                self.assertRaises(
-                    l3.RouterInUse,
-                    self.mixin._validate_router_migration,
-                    self.ctx,
-                    router_db,
-                    {'distributed': True})
-                return router_db, fw_mock, vpn_mock
-
-    def test__validate_router_migration_fail_with_vpnservice(self):
-        '''Test to check router migration fail with vpn.'''
-        router_db, mock_firewall, mock_vpnaas = (
-            self._router_migration_with_services_setup(
-                test_side_effect_vpn=(
-                    nexception.RouterInUseByVPNService(
-                        router_id='fake_id',
-                        vpnservice_id='fake_vpnaas_id')
-                ),
-                no_vpn=False,
-                no_fw=True))
-        mock_vpnaas.assert_called_once_with(self.ctx, router_db)
-
-    def test__validate_router_migration_fail_with_fwservice(self):
-        '''Test to check router migration with firewall.'''
-        router_db, mock_firewall, mock_vpnaas = (
-            self._router_migration_with_services_setup(
-                test_side_effect_vpn=None,
-                test_side_effect_fw=l3.RouterInUse(
-                    router_id='fake_id'
-                ),
-                no_vpn=True,
-                no_fw=False))
-        mock_firewall.assert_called_once_with(self.ctx, router_db)
-
-    def test__validate_router_migration_with_no_services(self):
-        '''Test to check router migration with no services.'''
-        router_db, mock_firewall, mock_vpnaas = (
-            self._router_migration_with_services_setup(
-                no_vpn=True,
-                no_fw=True))
-        mock_vpnaas.assert_called_once_with(self.ctx, router_db)
-        mock_firewall.assert_called_once_with(self.ctx, router_db)
+            mock.patch.object(self.mixin, 'check_router_has_no_firewall'),
+            mock.patch.object(self.mixin, 'check_router_has_no_vpnaas')
+        ) as (check_fw, check_vpn):
+            self.mixin._validate_router_migration(
+                self.ctx, router_db, {'distributed': True})
+            check_fw.assert_called_once_with(self.ctx, router_db)
+            check_vpn.assert_called_once_with(self.ctx, router_db)
+
+    def test_check_router_has_no_firewall_raises(self):
+        with mock.patch.object(
+            manager.NeutronManager, 'get_service_plugins') as sp:
+            fw_plugin = mock.Mock()
+            sp.return_value = {'FIREWALL': fw_plugin}
+            fw_plugin.get_firewalls.return_value = [mock.ANY]
+            self.assertRaises(
+                l3.RouterInUse,
+                self.mixin.check_router_has_no_firewall,
+                self.ctx, {'id': 'foo_id', 'tenant_id': 'foo_tenant'})
+
+    def test_check_router_has_no_firewall_passes(self):
+        with mock.patch.object(manager.NeutronManager,
+                               'get_service_plugins',
+                               return_value={}):
+            self.assertTrue(
+                self.mixin.check_router_has_no_firewall(mock.ANY, mock.ANY))
+
+    def test_check_router_has_no_vpn(self):
+        with mock.patch.object(
+            manager.NeutronManager, 'get_service_plugins') as sp:
+            vpn_plugin = mock.Mock()
+            sp.return_value = {'VPN': vpn_plugin}
+            self.mixin.check_router_has_no_vpnaas(mock.ANY, {'id': 'foo_id'})
+            vpn_plugin.check_router_in_use.assert_called_once_with(
+                mock.ANY, 'foo_id')