[MERGE] (test_)mail: improve activities tests and access checks

Overall purpose of this merge is to add some tests on activities notably
about ACLs and access to records.

See sub commits for more details.

Task-2272475

closes odoo/odoo#71131

Signed-off-by: Thibault Delavallee (tde) <tde@openerp.com>

addons/mail/models/mail_activity.py

@@ -408,6 +408,17 @@ class MailActivity(models.Model):
                     {'type': 'activity_updated', 'activity_created': True})
         return activities
 
+    def read(self, fields=None, load='_classic_read'):
+        """ When reading specific fields, read calls _read that manually applies ir rules
+        (_apply_ir_rules), instead of calling check_access_rule.
+
+        Meaning that our custom rules enforcing from '_filter_access_rules' and
+        '_filter_access_rules_python' are bypassed in that case.
+        To make sure we apply our custom security rules, we force a call to 'check_access_rule'. """
+
+        self.check_access_rule('read')
+        return super(MailActivity, self).read(fields=fields, load=load)
+
     def write(self, values):
         if values.get('user_id'):
             user_changes = self.filtered(lambda activity: activity.user_id.id != values.get('user_id'))
@@ -442,6 +453,93 @@ class MailActivity(models.Model):
                     {'type': 'activity_updated', 'activity_deleted': True})
         return super(MailActivity, self).unlink()
 
+    @api.model
+    def _search(self, args, offset=0, limit=None, order=None, count=False, access_rights_uid=None):
+        """ Override that adds specific access rights of mail.activity, to remove
+        ids uid could not see according to our custom rules. Please refer to
+        _filter_access_rules_remaining for more details about those rules.
+
+        The method is inspired by what has been done on mail.message. """
+
+        # Rules do not apply to administrator
+        if self.env.is_superuser():
+            return super(MailActivity, self)._search(
+                args, offset=offset, limit=limit, order=order,
+                count=count, access_rights_uid=access_rights_uid)
+        # Perform a super with count as False, to have the ids, not a counter
+        ids = super(MailActivity, self)._search(
+            args, offset=offset, limit=limit, order=order,
+            count=False, access_rights_uid=access_rights_uid)
+        if not ids and count:
+            return 0
+        elif not ids:
+            return ids
+
+        # check read access rights before checking the actual rules on the given ids
+        super(MailActivity, self.with_user(access_rights_uid or self._uid)).check_access_rights('read')
+
+        self.flush(['res_model', 'res_id'])
+        activities_to_check = []
+        for sub_ids in self._cr.split_for_in_conditions(ids):
+            self._cr.execute("""
+                SELECT DISTINCT activity.id, activity.res_model, activity.res_id
+                FROM "%s" activity
+                WHERE activity.id = ANY (%%(ids)s)""" % self._table, dict(ids=list(sub_ids)))
+            activities_to_check = self._cr.dictfetchall()
+
+        activity_to_documents = {}
+        for activity in activities_to_check:
+            activity_to_documents.setdefault(activity['res_model'], list()).append(activity['res_id'])
+
+        allowed_ids = []
+        for doc_model, doc_ids in activity_to_documents.items():
+            # fall back on related document access right checks. Use the same as defined for mail.thread
+            # if available; otherwise fall back on read
+            if hasattr(self.env[doc_model], '_mail_post_access'):
+                doc_operation = self.env[doc_model]._mail_post_access
+            else:
+                doc_operation = 'read'
+            DocumentModel = self.env[doc_model].with_user(access_rights_uid or self._uid)
+            right = DocumentModel.check_access_rights(doc_operation, raise_exception=False)
+            if right:
+                valid_docs = DocumentModel.browse(doc_ids)._filter_access_rules(doc_operation)
+                allowed_ids += [
+                    activity['id'] for activity in activities_to_check
+                    if activity['res_model'] == doc_model and activity['res_id'] in valid_docs.ids]
+
+        if count:
+            return len(allowed_ids)
+        else:
+            # re-construct a list based on ids, because 'allowed_ids' does not keep the original order
+            id_list = [id for id in ids if id in allowed_ids]
+            return id_list
+
+    @api.model
+    def _read_group_raw(self, domain, fields, groupby, offset=0, limit=None, orderby=False, lazy=True):
+        """ The base _read_group_raw method implementation computes a where based on a given domain
+        (_where_calc) and manually applies ir rules (_apply_ir_rules).
+
+        Meaning that our custom rules enforcing from '_filter_access_rules' and
+        '_filter_access_rules_python' are bypassed in that case.
+
+        This overrides re-uses the _search implementation to force the read group domain to allowed
+        ids only, that are computed based on our custom rules (see _filter_access_rules_remaining
+        for more details). """
+
+        # Rules do not apply to administrator
+        if not self.env.is_superuser():
+            allowed_ids = self._search(domain, count=False)
+            if allowed_ids:
+                domain = expression.AND([domain, [('id', 'in', allowed_ids)]])
+            else:
+                # force void result if no allowed ids found
+                domain = expression.AND([domain, [(0, '=', 1)]])
+
+        return super(MailActivity, self)._read_group_raw(
+            domain=domain, fields=fields, groupby=groupby, offset=offset,
+            limit=limit, orderby=orderby, lazy=lazy,
+        )
+
     def name_get(self):
         res = []
         for record in self:

addons/test_mail/tests/test_mail_activity.py

@@ -69,6 +69,56 @@ class TestActivityRights(TestActivityCommon):
                 raise exceptions.AccessError('Hop hop hop Ernest, please step back.')
             return DEFAULT
 
+        test_activity = self.env['mail.activity'].with_user(self.user_admin).create({
+            'activity_type_id': self.env.ref('test_mail.mail_act_test_todo').id,
+            'res_model_id': self.env.ref('test_mail.model_mail_test_activity').id,
+            'res_id': self.test_record.id,
+            'user_id': self.user_admin.id,
+            'summary': 'Summary',
+        })
+
+        # can _search activities if access to the document
+        self.env['mail.activity'].with_user(self.user_employee)._search(
+            [('id', '=', test_activity.id)], count=False)
+
+        # cannot _search activities if no access to the document
+        with patch.object(MailTestActivity, 'check_access_rights', autospec=True, side_effect=_employee_crash):
+            with self.assertRaises(exceptions.AccessError):
+                searched_activity = self.env['mail.activity'].with_user(self.user_employee)._search(
+                    [('id', '=', test_activity.id)], count=False)
+
+        # can read_group activities if access to the document
+        read_group_result = self.env['mail.activity'].with_user(self.user_employee).read_group(
+            [('id', '=', test_activity.id)],
+            ['summary'],
+            ['summary'],
+        )
+        self.assertEqual(1, read_group_result[0]['summary_count'])
+        self.assertEqual('Summary', read_group_result[0]['summary'])
+
+        # cannot read_group activities if no access to the document
+        with patch.object(MailTestActivity, 'check_access_rights', autospec=True, side_effect=_employee_crash):
+            with self.assertRaises(exceptions.AccessError):
+                self.env['mail.activity'].with_user(self.user_employee).read_group(
+                    [('id', '=', test_activity.id)],
+                    ['summary'],
+                    ['summary'],
+                )
+
+        # cannot read activities if no access to the document
+        with patch.object(MailTestActivity, 'check_access_rights', autospec=True, side_effect=_employee_crash):
+            with self.assertRaises(exceptions.AccessError):
+                searched_activity = self.env['mail.activity'].with_user(self.user_employee).search(
+                    [('id', '=', test_activity.id)])
+                searched_activity.read(['summary'])
+
+        # cannot search_read activities if no access to the document
+        with patch.object(MailTestActivity, 'check_access_rights', autospec=True, side_effect=_employee_crash):
+            with self.assertRaises(exceptions.AccessError):
+                self.env['mail.activity'].with_user(self.user_employee).search_read(
+                    [('id', '=', test_activity.id)],
+                    ['summary'])
+
         # cannot create activities for people that cannot access record
         with patch.object(MailTestActivity, 'check_access_rights', autospec=True, side_effect=_employee_crash):
             with self.assertRaises(exceptions.UserError):

addons/test_mail/tests/test_performance.py

@@ -249,7 +249,7 @@ class TestMailAPIPerformance(BaseMailPerformance):
 
         record.write({'name': 'Dupe write'})
 
-        with self.assertQueryCount(__system__=17, emp=19):
+        with self.assertQueryCount(__system__=17, emp=20):
             record.action_close('Dupe feedback')
 
         self.assertEqual(record.activity_ids, self.env['mail.activity'])