openstack/neutron
Code Issues Proposed changes

Add UnionModel support to filter query generator

Browse Source 
The _apply_filters_to_query method did not handle UnionModels
so objects leveraging it (i.e. RBAC policies) did not have
queries applied to them.

This patch corrects it by iterating through the component models
of the UnionModel and applying the filters to each component model.
It also adds an API test on RBAC that exercises the filtering.

Change-Id: I449acf359dd61189bbdacd200d7c41a4a88d3de8
Closes-Bug: #1517818
(cherry-picked from 35beca5327)
This commit is contained in:
Kevin Benton 2015-11-19 21:16:42 -08:00 committed by Kevin Benton
parent 17a9f8b67f
commit 8c09277728
2 changed files with 26 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
neutron/db/common_db_mixin.py
View File

@ -206,6 +206,13 @@ class CommonDbMixin(object):
model.rbac_entries.remote_attr.class_.id)
def _apply_filters_to_query(self, query, model, filters, context=None):
if isinstance(model, UnionModel):
# NOTE(kevinbenton): a unionmodel is made up of multiple tables so
# we apply the filter to each table
for component_model in model.model_map.values():
query = self._apply_filters_to_query(query, component_model,
filters, context)
return query
if filters:
for key, value in six.iteritems(filters):
column = getattr(model, key, None)

19
neutron/tests/api/admin/test_shared_network_extension.py
View File

@ -329,6 +329,25 @@ class RBACSharedNetworksTest(base.BaseAdminNetworkTest):
self.assertEqual(
p2, self.admin_client.show_rbac_policy(p2['id'])['rbac_policy'])
@test.attr(type='smoke')
@test.idempotent_id('e7bcb1ea-4877-4266-87bb-76f68b421f31')
def test_filter_policies(self):
net = self.create_network()
pol1 = self.client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_shared',
target_tenant=self.client2.tenant_id)['rbac_policy']
pol2 = self.client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_shared',
target_tenant=self.client.tenant_id)['rbac_policy']
res1 = self.client.list_rbac_policies(id=pol1['id'])['rbac_policies']
res2 = self.client.list_rbac_policies(id=pol2['id'])['rbac_policies']
self.assertEqual(1, len(res1))
self.assertEqual(1, len(res2))
self.assertEqual(pol1['id'], res1[0]['id'])
self.assertEqual(pol2['id'], res2[0]['id'])
@test.attr(type='smoke')
@test.idempotent_id('86c3529b-1231-40de-803c-afffffff6fff')
def test_regular_client_blocked_from_sharing_anothers_network(self):