From 35beca53277d187af0ade05466e81699c6575b07 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 --- 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 cf319cd37..e2473a5d0 100644 --- a/neutron/db/common_db_mixin.py +++ b/neutron/db/common_db_mixin.py @@ -184,6 +184,13 @@ class CommonDbMixin(object): return query.filter(model.id == id).one() 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 8a30c6a9b..ef3fe2845 100644 --- a/neutron/tests/api/admin/test_shared_network_extension.py +++ b/neutron/tests/api/admin/test_shared_network_extension.py @@ -316,6 +316,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): -- 2.45.2