openstack/neutron
Code Issues Proposed changes

Merge "Update network external attribute for RBAC change"

Browse Source
This commit is contained in:
Zuul 2017-12-13 08:55:05 +00:00 committed by Gerrit Code Review
commit 59e2c40f14
3 changed files with 62 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
neutron/db/external_net_db.py
View File

@ -169,6 +169,23 @@ class External_net_db_mixin(object):
{extnet_apidef.EXTERNAL: True},
allow_all=False)
@registry.receives('rbac-policy', [events.AFTER_DELETE])
def _process_ext_policy_delete(self, resource, event, trigger, context,
object_type, policy, **kwargs):
if (object_type != 'network' or
policy['action'] != 'access_as_external'):
return
net_as_external = context.session.query(rbac_db.NetworkRBAC).filter(
rbac_db.NetworkRBAC.object_id == policy['object_id'],
rbac_db.NetworkRBAC.action == 'access_as_external').count()
# If the network still have rbac policies, we should not
# update external attribute.
if net_as_external:
return
net = self.get_network(context, policy['object_id'])
self._process_l3_update(context, net,
{extnet_apidef.EXTERNAL: False})
@registry.receives('rbac-policy', (events.BEFORE_UPDATE,
events.BEFORE_DELETE))
def _validate_ext_not_in_use_by_tenant(self, resource, event, trigger,

3
neutron/db/rbac_db_mixin.py
View File

@ -94,6 +94,9 @@ class RbacPluginMixin(common_db_mixin.CommonDbMixin):
details=ex)
with context.session.begin(subtransactions=True):
context.session.delete(entry)
registry.notify(RBAC_POLICY, events.AFTER_DELETE, self,
context=context, object_type=object_type,
policy=entry)
self.object_type_cache.pop(id, None)
def _get_rbac_policy(self, context, id):

42
neutron/tests/unit/db/test_rbac_db_mixin.py
View File

@ -87,6 +87,48 @@ class NetworkRbacTestcase(test_plugin.NeutronDbPluginV2TestCase):
for k, v in policy['rbac_policy'].items():
self.assertEqual(netrbac2[k], v)
def test_delete_network_rbac_external(self):
with self.network() as ext_net:
net_id = ext_net['network']['id']
self._assert_external_net_state(net_id, is_external=False)
policy = self._make_networkrbac(ext_net,
'*',
'access_as_external')
net_rbac = self.plugin.create_rbac_policy(self.context, policy)
self._assert_external_net_state(net_id, is_external=True)
self.plugin.delete_rbac_policy(self.context, net_rbac['id'])
self._assert_external_net_state(net_id, is_external=False)
def test_delete_network_rbac_external_with_multi_rbac_policy(self):
with self.network() as ext_net:
net_id = ext_net['network']['id']
self._assert_external_net_state(net_id, is_external=False)
policy1 = self._make_networkrbac(ext_net,
'test-tenant-1',
'access_as_external')
net_rbac1 = self.plugin.create_rbac_policy(self.context, policy1)
self._assert_external_net_state(net_id, is_external=True)
policy2 = self._make_networkrbac(ext_net,
'test-tenant-2',
'access_as_external')
self.plugin.create_rbac_policy(self.context, policy2)
self._assert_external_net_state(net_id, is_external=True)
self.plugin.delete_rbac_policy(self.context, net_rbac1['id'])
self._assert_external_net_state(net_id, is_external=True)
def test_delete_external_network_shared_rbac(self):
with self.network() as ext_net:
net_id = ext_net['network']['id']
self.plugin.update_network(
self.context, net_id,
{'network': {'router:external': True}})
self._assert_external_net_state(net_id, is_external=True)
policy = self._make_networkrbac(ext_net, 'test-tenant-2')
net_rbac = self.plugin.create_rbac_policy(self.context, policy)
self.plugin.delete_rbac_policy(self.context, net_rbac['id'])
# Make sure that external attribute not changed.
self._assert_external_net_state(net_id, is_external=True)
def test_update_networkrbac_valid(self):
orig_target = 'test-tenant-2'
new_target = 'test-tenant-3'