[IMP] mail: add mixin to compute many2one duration

This commit implements a mixin, `MailTrackingDurationMixin`, that can be added
to a model with a `many2one` field. It computes the time a record spends in each
value the many2one field takes and stores it in a JSON field
(`duration_tracking`).

The primary use is with the StatusBarDurationField
(`widget='statusbar_duration'`), to compute and display the time a record has
spent in each stage in the form view statusbar.

To specify on what field the computation has to be done, the model that inherits
from this mixin has to specify  `_track_duration_field`.  (e.g.
_track_duration_field = 'stage_id')

Computation is based on `mail.tracking.value` messages, so tracking has to be
activated for that field.

Task-3032773

Part-of: odoo/odoo#108554

addons/mail/__manifest__.py

@@ -139,6 +139,7 @@ For more specific needs, you may also assign custom-defined actions
             'mail/static/src/discuss/**/common/**/*',
             'mail/static/src/discuss/**/web/**/*',
             ('remove', 'mail/static/src/discuss/**/*.dark.scss'),
+            'mail/static/src/views/fields/**/*',
         ],
         "web.dark_mode_assets_backend": [
             'mail/static/src/**/*.dark.scss',

addons/mail/models/__init__.py

@@ -14,6 +14,7 @@ from . import mail_thread
 from . import mail_thread_blacklist
 from . import mail_thread_cc
 from . import mail_thread_main_attachment
+from . import mail_tracking_duration_mixin
 from . import template_reset_mixin
 
 # mail models

addons/mail/models/mail_tracking_duration_mixin.py

@@ -0,0 +1,95 @@
+from collections import defaultdict
+
+from odoo import _, fields, models
+
+
+class MailTrackingDurationMixin(models.AbstractModel):
+    _name = "mail.tracking.duration.mixin"
+    _description = "Mixin to compute the time a record has spent in each value a many2one field can take"
+
+    duration_tracking = fields.Json(
+        string="Status time", compute="_compute_duration_tracking",
+        help="JSON that maps ids from a many2one field to seconds spent")
+
+    def _compute_duration_tracking(self):
+        """
+        Computes duration_tracking, a Json field stored as { <many2one_id (str)>: <duration_spent_in_seconds (int)> }
+
+            e.g. {"1": 1230, "2": 2220, "5": 14}
+
+        `_track_duration_field` must be present in the model that uses the mixin to specify on what
+        field to compute time spent. Besides, tracking must be activated for that field.
+
+            e.g.
+            class MyModel(models.Model):
+                _track_duration_field = "tracked_field"
+
+                tracked_field = fields.Many2one('tracked.model', tracking=True)
+        """
+
+        field = self.env['ir.model.fields'].sudo().search_fetch([
+            ('model', '=', self._name),
+            ('name', '=', self._track_duration_field),
+        ], ['id'], limit=1)
+
+        if (
+            self._track_duration_field not in self._track_get_fields()
+            or self._fields[self._track_duration_field].type != 'many2one'
+        ):
+            self.duration_tracking = False
+            raise ValueError(
+                _('Field "%(field)s" on model "%(model)s" must be of type Many2one '
+                'and have tracking=True for the computation of duration.') % {
+                    'field': self._track_duration_field, 'model': self._name
+                }
+            )
+
+        self.env['mail.tracking.value'].flush_model()
+        self.env['mail.message'].flush_model()
+        query = """
+               SELECT m.res_id,
+                      v.create_date,
+                      v.old_value_integer
+                 FROM mail_tracking_value v
+            LEFT JOIN mail_message m
+                   ON m.id = v.mail_message_id
+                  AND v.field = %(field_id)s
+                WHERE m.model = %(model_name)s
+                  AND m.res_id IN %(record_ids)s
+             ORDER BY v.id
+        """
+        self.env.cr.execute(query, {"field_id": field.id, "model_name": self._name, "record_ids": tuple(self.ids)})
+        trackings = self.env.cr.dictfetchall()
+
+        for record in self:
+            record_trackings = [tracking for tracking in trackings if tracking['res_id'] == record._origin.id]
+            record.duration_tracking = record._get_duration_from_tracking(record_trackings)
+
+    def _get_duration_from_tracking(self, trackings):
+        """
+        Calculates the duration spent in each value based on the provided list of trackings.
+        It adds a "fake" tracking at the end of the trackings list to account for the time spent in the current value.
+
+        Args:
+            trackings (list): A list of dictionaries representing the trackings with:
+                - 'create_date': The date and time of the tracking.
+                - 'old_value_integer': The ID of the previous value.
+
+        Returns:
+            dict: A dictionary where the keys are the IDs of the values, and the values are the durations in seconds
+        """
+        self.ensure_one()
+        json = defaultdict(lambda: 0)
+        previous_date = self.create_date
+
+        # add "fake" tracking for time spent in the current value
+        trackings.append({
+            'create_date': self.env.cr.now(),
+            'old_value_integer': self[self._track_duration_field].id,
+        })
+
+        for tracking in trackings:
+            json[tracking['old_value_integer']] += int((tracking['create_date'] - previous_date).total_seconds())
+            previous_date = tracking['create_date']
+
+        return json

addons/mail/tests/mail_tracking_duration_testing.py

@@ -0,0 +1,196 @@
+from collections import defaultdict
+from datetime import datetime, timedelta
+from unittest.mock import patch
+
+from odoo.addons.mail.tests.common import MailCommon
+
+
+class TestMailTrackingDurationMixin(MailCommon):
+
+    @classmethod
+    def setUpClass(cls, tested_model, model_fields=None):
+
+        super().setUpClass()
+        if model_fields:
+            for field in model_fields:
+                if model_fields[field] == 'create':
+                    model_fields[field] = cls.env[tested_model][field].create({'name': 'test'}).id
+
+        stage_1_values = {'name': 'Stage 1'}
+        stage_2_values = {'name': 'Stage 2'}
+        stage_3_values = {'name': 'Stage 3'}
+        stage_4_values = {'name': 'Stage 4'}
+        cls.track_duration_field = cls.env[tested_model]._track_duration_field
+
+        stage_model = cls.env[tested_model][cls.track_duration_field]
+        cls.stage_1 = stage_model.create(stage_1_values)
+        cls.stage_2 = stage_model.create(stage_2_values)
+        cls.stage_3 = stage_model.create(stage_3_values)
+        cls.stage_4 = stage_model.create(stage_4_values)
+
+        record_values = {'name': 'test record', cls.track_duration_field: cls.stage_1.id}
+        if model_fields:
+            record_values.update(model_fields)
+
+        cls.mock_start_time = datetime(2023, 2, 15, 12, 0, 0)
+
+        with patch.object(cls.env.cr, 'now', return_value=cls.mock_start_time):
+            cls.rec_1, cls.rec_2, cls.rec_3, cls.rec_4 = cls.env[tested_model].create(
+                [record_values for i in range(4)])
+        cls.flush_tracking(cls)
+
+    def _update_duration_tracking(self, record_to_tracking_dic, minutes, new_stage=False):
+        """
+        Updates the mock duration_tracking field for multiple records based on the provided minutes.
+        If new_stage is defined, the stage of the records is updated as well.
+
+        Args:
+            record_to_tracking_dic (list): A list of tuples mapping records to their respective tracking dictionaries.
+            minutes (int): The number of minutes to be added to the duration tracking, which will be converted to seconds.
+            new_stage (int, optional): Indicated the new stage to be set for the records. Defaults to False.
+        """
+        for record, tracking_dic in record_to_tracking_dic:
+            tracking_dic[str(record[self.track_duration_field].id)] += minutes * 60
+            if new_stage:
+                tracking_dic[str(new_stage.id)] += 0
+                record[self.track_duration_field] = new_stage
+                self.flush_tracking()
+
+    def assertTrackingDuration(self, records, record_to_tracking_dic):
+        """
+        Asserts whether for multiple records their duration_tracking is equal to a dictionary
+
+        Args:
+            records (recordset): all the records that need to be asserted
+            record_to_tracking_dic (list): A list of tuples mapping records to their respective tracking dictionaries.
+        """
+        records._compute_duration_tracking()
+        for record, tracking_dic in record_to_tracking_dic:
+            self.assertDictEqual(dict(tracking_dic), record.duration_tracking)
+
+    def _test_record_duration_tracking(self):
+        """
+        Moves a record's many2one field through several values and asserts the duration spent in that value each time
+        """
+
+        with patch.object(self.env.cr, 'now', return_value=self.mock_start_time) as now:
+
+            track_duration_tracking = defaultdict(lambda: 0)
+            record = self.rec_1
+
+            minutes = 5
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking([(record, track_duration_tracking)], minutes, self.stage_2)
+            self.assertTrackingDuration(record, [(record, track_duration_tracking)])
+
+            minutes = 100
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking([(record, track_duration_tracking)], minutes)
+            self.assertTrackingDuration(record, [(record, track_duration_tracking)])
+
+            minutes = 5000
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking([(record, track_duration_tracking)], minutes, self.stage_3)
+            self.assertTrackingDuration(record, [(record, track_duration_tracking)])
+
+            minutes = 5000
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking([(record, track_duration_tracking)], minutes, self.stage_4)
+            self.assertTrackingDuration(record, [(record, track_duration_tracking)])
+
+            minutes = 20
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking([(record, track_duration_tracking)], minutes, self.stage_2)
+            self.assertTrackingDuration(record, [(record, track_duration_tracking)])
+
+            minutes = 55
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking([(record, track_duration_tracking)], minutes, self.stage_4)
+            self.assertTrackingDuration(record, [(record, track_duration_tracking)])
+
+            minutes = 200
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking([(record, track_duration_tracking)], minutes)
+            self.assertTrackingDuration(record, [(record, track_duration_tracking)])
+
+            minutes = 300
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking([(record, track_duration_tracking)], minutes, self.stage_3)
+            self.assertTrackingDuration(record, [(record, track_duration_tracking)])
+
+    def _test_record_duration_tracking_batch(self):
+        """
+        Moves for a batch of records many2one field through several values and asserts the duration
+        spent in that value each time.
+        """
+
+        with patch.object(self.env.cr, 'now', return_value=self.mock_start_time) as now:
+
+            track_duration_tracking1 = defaultdict(lambda: 0)
+            track_duration_tracking2 = defaultdict(lambda: 0)
+            track_duration_tracking3 = defaultdict(lambda: 0)
+            batch = self.rec_1 | self.rec_2 | self.rec_3
+            record_to_tracking_dic = [
+                (self.rec_1, track_duration_tracking1),
+                (self.rec_2, track_duration_tracking2),
+                (self.rec_3, track_duration_tracking3)
+            ]
+
+            minutes = 5
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking(record_to_tracking_dic, minutes, self.stage_2)
+            self.assertTrackingDuration(batch, record_to_tracking_dic)
+
+            minutes = 100
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking(record_to_tracking_dic, minutes)
+            self.assertTrackingDuration(batch, record_to_tracking_dic)
+
+            minutes = 5000
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking(record_to_tracking_dic, minutes, self.stage_3)
+            self.assertTrackingDuration(batch, record_to_tracking_dic)
+
+            minutes = 5000
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking(record_to_tracking_dic, minutes, self.stage_4)
+            self.assertTrackingDuration(batch, record_to_tracking_dic)
+
+            minutes = 20
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking(record_to_tracking_dic, minutes, self.stage_2)
+            self.assertTrackingDuration(batch, record_to_tracking_dic)
+
+            minutes = 55
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking(record_to_tracking_dic, minutes, self.stage_4)
+            self.assertTrackingDuration(batch, record_to_tracking_dic)
+
+            minutes = 200
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking(record_to_tracking_dic, minutes)
+            self.assertTrackingDuration(batch, record_to_tracking_dic)
+
+            minutes = 300
+            now.return_value += timedelta(minutes=minutes)
+            self._update_duration_tracking(record_to_tracking_dic, minutes, self.stage_3)
+            self.assertTrackingDuration(batch, record_to_tracking_dic)
+
+    def _test_queries_batch_duration_tracking(self):
+        """
+        The MailTrackingDuration mixin is only supposed to add 3 queries
+        """
+
+        batch = self.rec_1 | self.rec_2 | self.rec_3 | self.rec_4
+        batch[self.track_duration_field] = self.stage_2.id
+        self.flush_tracking()
+        batch[self.track_duration_field] = self.stage_4.id
+        self.flush_tracking()
+        batch[self.track_duration_field] = self.stage_1.id
+        self.flush_tracking()
+        batch[self.track_duration_field] = self.stage_3.id
+        self.flush_tracking()
+        batch[self.track_duration_field] = self.stage_2.id
+
+        with self.assertQueryCount(3):
+            batch._compute_duration_tracking()

addons/test_mail/models/test_mail_models.py

@@ -397,3 +397,13 @@ class MailTestComposerSource(models.Model):
 
     def _mail_get_partner_fields(self):
         return ['customer_id']
+
+
+class MailTestMailTrackingDuration(models.Model):
+    _description = 'Fake model to test the mixin mail.tracking.duration.mixin'
+    _name = 'mail.test.mail.tracking.duration'
+    _track_duration_field = 'customer_id'
+    _inherit = ['mail.thread', 'mail.tracking.duration.mixin']
+
+    name = fields.Char()
+    customer_id = fields.Many2one('res.partner', 'Customer', tracking=True)

addons/test_mail/security/ir.model.access.csv

@@ -49,3 +49,4 @@ access_mail_test_track_monetary,mail.test.track.monetary,model_mail_test_track_m
 access_mail_test_track_selection_portal,mail.test.track.selection.portal,model_mail_test_track_selection,base.group_portal,0,0,0,0
 access_mail_test_track_selection_user,mail.test.track.selection.user,model_mail_test_track_selection,base.group_user,1,1,1,1
 access_mail_test_properties_user,mail.test.properties.user,model_mail_test_properties,base.group_user,1,1,1,1
+access_mail_test_mail_tracking_duration,mail.test.mail.tracking.duration,model_mail_test_mail_tracking_duration,base.group_user,1,1,1,1

addons/test_mail/tests/__init__.py

@@ -6,17 +6,18 @@ from . import test_mail_activity
 from . import test_mail_composer
 from . import test_mail_composer_mixin
 from . import test_mail_followers
-from . import test_mail_message
-from . import test_mail_mail
 from . import test_mail_gateway
+from . import test_mail_mail
+from . import test_mail_management
+from . import test_mail_message
 from . import test_mail_multicompany
-from . import test_mail_thread_internals
+from . import test_mail_security
 from . import test_mail_template
 from . import test_mail_template_preview
+from . import test_mail_thread_internals
+from . import test_mail_tracking_duration_mixin
 from . import test_message_management
 from . import test_message_post
 from . import test_message_track
 from . import test_performance
 from . import test_ui
-from . import test_mail_management
-from . import test_mail_security

addons/test_mail/tests/test_mail_tracking_duration_mixin.py

@@ -0,0 +1,17 @@
+from odoo.addons.mail.tests.mail_tracking_duration_testing import TestMailTrackingDurationMixin
+
+
+class TestMailTrackingDuration(TestMailTrackingDurationMixin):
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass('mail.test.mail.tracking.duration')
+
+    def test_mail_tracking_duration(self):
+        self._test_record_duration_tracking()
+
+    def test_mail_tracking_duration_batch(self):
+        self._test_record_duration_tracking_batch()
+
+    def test_queries_batch_mail_tracking_duration(self):
+        self._test_queries_batch_duration_tracking()