diff --git a/neutron/db/rbac_db_mixin.py b/neutron/db/rbac_db_mixin.py index 1db8ed2f98e..21f03f3630f 100644 --- a/neutron/db/rbac_db_mixin.py +++ b/neutron/db/rbac_db_mixin.py @@ -20,7 +20,7 @@ from neutron_lib.callbacks import resources from neutron_lib.db import api as db_api from neutron_lib.db import utils as db_utils from neutron_lib import exceptions as n_exc -from oslo_db import exception as db_exc +from neutron_lib.objects import exceptions as o_exc from neutron.extensions import rbac as ext_rbac from neutron.objects import base as base_obj @@ -52,7 +52,7 @@ class RbacPluginMixin(object): 'target_tenant': e['target_tenant']} _rbac_obj = rbac_class(context, **rbac_args) _rbac_obj.create() - except db_exc.DBDuplicateEntry: + except o_exc.NeutronDbObjectDuplicateEntry: raise ext_rbac.DuplicateRbacPolicy() return self._make_rbac_policy_dict(_rbac_obj) diff --git a/neutron/tests/unit/db/test_rbac_db_mixin.py b/neutron/tests/unit/db/test_rbac_db_mixin.py index cb20220f2ff..47b69392fbd 100644 --- a/neutron/tests/unit/db/test_rbac_db_mixin.py +++ b/neutron/tests/unit/db/test_rbac_db_mixin.py @@ -19,6 +19,7 @@ from neutron_lib.callbacks import events from neutron_lib import constants from neutron_lib import context from oslo_utils import uuidutils +import testtools from neutron.db.db_base_plugin_v2 import NeutronDbPluginV2 as db_plugin_v2 from neutron.db import rbac_db_models @@ -73,6 +74,20 @@ class NetworkRbacTestcase(test_plugin.NeutronDbPluginV2TestCase): self.plugin.create_rbac_policy(self.context, policy) self._assert_external_net_state(net_id, is_external=True) + def test_create_network_rbac_shared_existing(self): + tenant = 'test-tenant' + with self.network() as net: + policy = self._make_networkrbac(net, + tenant, + rbac_db_models.ACCESS_SHARED) + self.plugin.create_rbac_policy(self.context, policy) + # Give server maximum of 10 seconds to make sure we don't hit DB + # retry mechanism when resource already exists + with self.assert_max_execution_time(10): + with testtools.ExpectedException( + ext_rbac.DuplicateRbacPolicy): + self.plugin.create_rbac_policy(self.context, policy) + def test_update_network_rbac_external_valid(self): orig_target = 'test-tenant-2' new_target = 'test-tenant-3'