Fix list security groups performance with RBAC

After change [1], if the system has a high number of security groups
with no associated RBAC entries, a non admin user owning only one
security group will experience unacceptable response times when
listing her security groups.

Change [1] added methods get_object and get_objects to class
RbacNeutronDbObjectMixin in neutron.objects.rbac_db, which retrieve with
and admin context all the objects (networks, subnets or security groups)
in the DB and then decide in memory whether the project that made the
query has access to them or not, based on their associated RBAC
policies. This change proposes to remove those methods and revert to
their counterparts in NeutronDbObject (neutron.objects.base), which use
a DB query scoped to the project to retrieve the objects based on their
associated RBAC policies by calling [2]. In this way, the potential
number of objects that are retrieved from the DB and that have to be
converted to OVOs is greatly reduced, improving significantly the
response time to the user.

[1] https://review.opendev.org/#/c/635311
[2] 7a58374fde/neutron_lib/db/model_query.py (L100)

Change-Id: Idd303778d83089da8fbeff40e3dda2bd19008d8e
Closes-Bug: #1830679
(cherry picked from commit a240c68022)

neutron/objects/rbac_db.py

@@ -88,35 +88,6 @@ class RbacNeutronDbObjectMixin(rbac_db_mixin.RbacPluginMixin,
                 cls.is_shared_with_tenant(context, db_obj.id,
                                           context.tenant_id))
 
-    @classmethod
-    def get_object(cls, context, **kwargs):
-        # We want to get the policy regardless of its tenant id. We'll make
-        # sure the tenant has permission to access the policy later on.
-        admin_context = context.elevated()
-        with cls.db_context_reader(admin_context):
-            obj = super(RbacNeutronDbObjectMixin,
-                        cls).get_object(admin_context, **kwargs)
-            if (not obj or not cls.is_accessible(context, obj)):
-                return
-            return obj
-
-    @classmethod
-    def get_objects(cls, context, _pager=None, validate_filters=True,
-                    **kwargs):
-        # We want to get the policy regardless of its tenant id. We'll make
-        # sure the tenant has permission to access the policy later on.
-        admin_context = context.elevated()
-        with cls.db_context_reader(admin_context):
-            objs = super(RbacNeutronDbObjectMixin,
-                         cls).get_objects(admin_context, _pager,
-                                          validate_filters, **kwargs)
-            result = []
-            for obj in objs:
-                if not cls.is_accessible(context, obj):
-                    continue
-                result.append(obj)
-            return result
-
     @classmethod
     def _get_db_obj_rbac_entries(cls, context, rbac_obj_id, rbac_action):
         rbac_db_model = cls.rbac_db_cls.db_model

neutron/tests/unit/objects/qos/test_policy.py

@@ -112,50 +112,33 @@ class QosPolicyObjectTestCase(test_base.BaseObjectIfaceTestCase):
     # TODO(ihrachys): stop overriding those test cases, instead base test cases
     # should be expanded if there are missing bits there to support QoS objects
     def test_get_objects(self):
-        admin_context = self.context.elevated()
-        with mock.patch.object(self.context, 'elevated',
-                               return_value=admin_context) as context_mock:
-            objs = self._test_class.get_objects(self.context)
-        context_mock.assert_called_once_with()
+        objs = self._test_class.get_objects(self.context)
         self.get_objects_mock.assert_any_call(
-            self._test_class, admin_context, _pager=None)
+            self._test_class, self.context, _pager=None)
         self.assertItemsEqual(
             [test_base.get_obj_persistent_fields(obj) for obj in self.objs],
             [test_base.get_obj_persistent_fields(obj) for obj in objs])
 
     def test_get_objects_valid_fields(self):
-        admin_context = self.context.elevated()
-
         with mock.patch.object(
             db_api, 'get_objects',
             return_value=[self.db_objs[0]]) as get_objects_mock:
-
-            with mock.patch.object(
-                    self.context,
-                    'elevated',
-                    return_value=admin_context) as context_mock:
-
-                objs = self._test_class.get_objects(
-                    self.context,
-                    **self.valid_field_filter)
-                context_mock.assert_called_once_with()
+            objs = self._test_class.get_objects(
+                self.context,
+                **self.valid_field_filter)
             get_objects_mock.assert_any_call(
-                self._test_class, admin_context, _pager=None,
+                self._test_class, self.context, _pager=None,
                 **self.valid_field_filter)
         self._check_equal(self.objs[0], objs[0])
 
     def test_get_object(self):
-        admin_context = self.context.elevated()
         with mock.patch.object(db_api, 'get_object',
-                return_value=self.db_objs[0]) as get_object_mock, \
-                mock.patch.object(self.context, 'elevated',
-                return_value=admin_context) as context_mock:
+                return_value=self.db_objs[0]) as get_object_mock:
             obj = self._test_class.get_object(self.context, id='fake_id')
             self.assertTrue(self._is_test_class(obj))
             self._check_equal(self.objs[0], obj)
-            context_mock.assert_called_once_with()
             get_object_mock.assert_called_once_with(
-                self._test_class, admin_context, id='fake_id')
+                self._test_class, self.context, id='fake_id')
 
     def test_to_dict_makes_primitive_field_value(self):
         # is_shared_with_tenant requires DB