[IMP] core: New 'capture_trigger' test helper

When testing cron triggers, it is common to get the newly created
triggers in order to validate they exist and are scheduled are the right
moment.

This new helper is a context manager that capture all triggers or
triggers created for a specific cron. The created triggers are
accessible via the context's object `records` attribute.

The various tests have been updated so they use that new helper.

closes odoo/odoo#68163

Signed-off-by: Julien Castiaux <Julien00859@users.noreply.github.com>
Co-authored-by: Raphaël Collet <rco@odoo.com>

addons/calendar/tests/test_event_notifications.py

@@ -7,10 +7,11 @@ from dateutil.relativedelta import relativedelta
 
 from odoo import fields
 from odoo.tests.common import TransactionCase, new_test_user
+from odoo.addons.base.tests.test_ir_cron import CronMixinCase
 from odoo.addons.mail.tests.common import MailCase
 
 
-class TestEventNotifications(TransactionCase, MailCase):
+class TestEventNotifications(TransactionCase, MailCase, CronMixinCase):
 
     @classmethod
     def setUpClass(cls):
@@ -142,27 +143,23 @@ class TestEventNotifications(TransactionCase, MailCase):
             self.assertEqual(notif, bus_message)
 
     def test_email_alarm(self):
-        cron_id = self.env.ref('calendar.ir_cron_scheduler_alarm').id
-        triggers_before = self.env['ir.cron.trigger'].search([('cron_id', '=', cron_id)])
-
-        alarm = self.env['calendar.alarm'].create({
-            'name': 'Alarm',
-            'alarm_type': 'email',
-            'interval': 'minutes',
-            'duration': 20,
-        })
         now = fields.Datetime.now()
-        self.event.write({
-            'start': now + relativedelta(minutes=15),
-            'stop': now + relativedelta(minutes=18),
-            'partner_ids': [fields.Command.link(self.partner.id)],
-            'alarm_ids': [fields.Command.link(alarm.id)],
-        })
+        with self.capture_triggers('calendar.ir_cron_scheduler_alarm') as capt:
+            alarm = self.env['calendar.alarm'].create({
+                'name': 'Alarm',
+                'alarm_type': 'email',
+                'interval': 'minutes',
+                'duration': 20,
+            })
+            self.event.write({
+                'start': now + relativedelta(minutes=15),
+                'stop': now + relativedelta(minutes=18),
+                'partner_ids': [fields.Command.link(self.partner.id)],
+                'alarm_ids': [fields.Command.link(alarm.id)],
+            })
 
-        triggers_after = self.env['ir.cron.trigger'].search([('cron_id', '=', cron_id)])
-        new_triggers = triggers_after - triggers_before
-        new_triggers.ensure_one()
-        self.assertLessEqual(new_triggers.call_at, now)
+        capt.records.ensure_one()
+        self.assertLessEqual(capt.records.call_at, now)
 
         with patch.object(fields.Datetime, 'now', lambda: now):
             with self.assertSinglePostNotifications([{'partner': self.partner, 'type': 'inbox'}], {

addons/mass_mailing/tests/test_mailing_internals.py

@@ -6,6 +6,7 @@ from datetime import datetime
 
 from freezegun import freeze_time
 
+from odoo.addons.base.tests.test_ir_cron import CronMixinCase
 from odoo.addons.mass_mailing.tests.common import MassMailCommon
 from odoo.tests.common import users, Form
 from odoo.tools import formataddr, mute_logger
@@ -125,7 +126,7 @@ class TestMassMailValues(MassMailCommon):
         self.assertEqual(mailing_form.mailing_model_real, 'res.partner')
 
 
-class TestMassMailFeatures(MassMailCommon):
+class TestMassMailFeatures(MassMailCommon, CronMixinCase):
 
     @classmethod
     def setUpClass(cls):
@@ -176,7 +177,7 @@ class TestMassMailFeatures(MassMailCommon):
         """ Technical test to ensure the cron is triggered at the correct
         time """
 
-        ir_cron_triggers = self.env['ir.cron.trigger'].sudo()
+        cron_id = self.env.ref('mass_mailing.ir_cron_mass_mailing_queue').id
         partner = self.env['res.partner'].create({
             'name': 'Jean-Alphonce',
             'email': 'jeanalph@example.com',
@@ -196,17 +197,14 @@ class TestMassMailFeatures(MassMailCommon):
         with freeze_time(now):
             for (test, truth) in [(False, now), (then, then)]:
                 with self.subTest(schedule_date=test):
-                    triggers_before = ir_cron_triggers.search([])
-                    mailing = self.env['mailing.mailing'].create({
-                        **common_mailing_values,
-                        'schedule_date': test,
-                    })
-                    mailing.action_put_in_queue()
-
-                    triggers_after = ir_cron_triggers.search([])
-                    new_triggers = triggers_after - triggers_before
-                    new_triggers.ensure_one()
-                    self.assertLessEqual(new_triggers.call_at, truth)
+                    with self.capture_triggers(cron_id) as capt:
+                        mailing = self.env['mailing.mailing'].create({
+                            **common_mailing_values,
+                            'schedule_date': test,
+                        })
+                        mailing.action_put_in_queue()
+                    capt.records.ensure_one()
+                    self.assertLessEqual(capt.records.call_at, truth)
 
     @users('user_marketing')
     @mute_logger('odoo.addons.mail.models.mail_mail')

odoo/addons/base/tests/test_ir_cron.py

@@ -4,10 +4,33 @@
 from unittest.mock import patch
 
 from odoo import fields
-from odoo.tests.common import TransactionCase
+from odoo.tests.common import TransactionCase, RecordCapturer
 
 
-class TestIrCron(TransactionCase):
+class CronMixinCase:
+    def capture_triggers(self, cron_id=None):
+        """
+        Get a context manager to get all cron triggers created during
+        the context lifetime. While in the context, it exposes the
+        triggers created so far from the beginning of the context. When
+        the context exits, it doesn't capture new triggers anymore.
+
+        The triggers are accessible on the `records` attribute of the
+        returned object.
+
+        :param cron_id: An optional cron record id (int) or xmlid (str)
+                        to only capture triggers for that cron.
+        """
+        if isinstance(cron_id, str):  # xmlid case
+            cron_id = self.env.ref(cron_id).id
+
+        return RecordCapturer(
+            model=self.env['ir.cron.trigger'].sudo(),
+            domain=[('cron_id', '=', cron_id)] if cron_id else []
+        )
+
+
+class TestIrCron(TransactionCase, CronMixinCase):
 
     def setUp(self):
         super(TestIrCron, self).setUp()

odoo/tests/common.py

@@ -148,6 +148,27 @@ def new_test_user(env, login='', groups='base.group_user', context=None, **kwarg
 
     return env['res.users'].with_context(**context).create(create_values)
 
+
+class RecordCapturer:
+    def __init__(self, model, domain):
+        self._model = model
+        self._domain = domain
+
+    def __enter__(self):
+        self._before = self._model.search(self._domain)
+        self._after = None
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_traceback):
+        if exc_type is None:
+            self._after = self._model.search(self._domain) - self._before
+
+    @property
+    def records(self):
+        if self._after is None:
+            return self._model.search(self._domain) - self._before
+        return self._after
+
 # ------------------------------------------------------------
 # Main classes
 # ------------------------------------------------------------