rbac: Catch correct exception for duplicated entry

RBAC network policy is uniquely identified by network ID. That means
when attempting to create such network policy, we should not retry when
such policy already exists in the database.

Before we switched in rbac to use OVO, we translated DB DBDuplicateEntry
on such ocasions into dedicated RBAC exception to avoid DB retry
mechanism (see bug/1551473). After introducing OVO layer for RBAC, the
exception was not changed to the one coming from OVO. This patch
replaces the exception from DB to the exception from OVO.

Another patch will go to neutron-tempest-plugin to limit time API needs
to reply with failure to user, when attempting to create an existing
policy.

Closes-Bug: #1831647

Change-Id: I7c65376f6fd6fc29d510ea532a684917ed95deb1
This commit is contained in:
Jakub Libosvar 2019-06-06 18:58:20 +00:00
parent fc5235c49c
commit 26b3e6b1c4
2 changed files with 17 additions and 2 deletions

View File

@ -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)

View File

@ -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'