From 8c09277728ab52c0733aaac8e0255e0240b460e7 Mon Sep 17 00:00:00 2001 From: Kevin Benton Date: Thu, 19 Nov 2015 21:16:42 -0800 Subject: [PATCH] Add UnionModel support to filter query generator 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 35beca53277d187af0ade05466e81699c6575b07) --- neutron/db/common_db_mixin.py | 7 +++++++ .../admin/test_shared_network_extension.py | 19 +++++++++++++++++++ 2 files changed, 26 insertions(+) diff --git a/neutron/db/common_db_mixin.py b/neutron/db/common_db_mixin.py index 2a65420e5e4..51c49e9d162 100644 --- a/neutron/db/common_db_mixin.py +++ b/neutron/db/common_db_mixin.py @@ -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) diff --git a/neutron/tests/api/admin/test_shared_network_extension.py b/neutron/tests/api/admin/test_shared_network_extension.py index 99021434e19..e077206dc33 100644 --- a/neutron/tests/api/admin/test_shared_network_extension.py +++ b/neutron/tests/api/admin/test_shared_network_extension.py @@ -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):