Merge "rbac: Catch correct exception for duplicated entry"

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)
 

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'