[12.0] queue_job: ease testing

queue_job/models/base.py

@@ -2,10 +2,14 @@
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)
 
 import inspect
+import logging
+import os
 
 from odoo import models, api
 from ..job import DelayableRecordset
 
+_logger = logging.getLogger(__name__)
+
 
 class Base(models.AbstractModel):
  """The base model, which is implicitly inherited by all models.
@@ -43,7 +47,6 @@ def with_delay(self, priority=None, eta=None,
  In the line above, in so far ``write`` is allowed to be delayed with
  ``@job``, the write will be executed in an asynchronous job.
 
-
  :param priority: Priority of the job, 0 being the higher priority.
  Default is 10.
  :param eta: Estimated Time of Arrival of the job. It will not be
@@ -61,7 +64,27 @@ def with_delay(self, priority=None, eta=None,
  the new job will not be added.
  :return: instance of a DelayableRecordset
  :rtype: :class:`odoo.addons.queue_job.job.DelayableRecordset`
+
+ Note for developers: if you want to run tests or simply disable
+ jobs queueing for debugging purposes, you can:
+
+ a. set the env var `TEST_QUEUE_JOB_NO_DELAY=1`
+ b. pass a ctx key `test_queue_job_no_delay=1`
+
+ In tests you'll have to mute the logger like:
+
+ @mute_logger('odoo.addons.queue_job.models.base')
  """
+ if os.getenv('TEST_QUEUE_JOB_NO_DELAY'):
+ _logger.warn(
+ '`TEST_QUEUE_JOB_NO_DELAY` env var found. NO JOB scheduled.'
+ )
+ return self
+ if self.env.context.get('test_queue_job_no_delay'):
+ _logger.warn(
+ '`test_queue_job_no_delay` ctx key found. NO JOB scheduled.'
+ )
+ return self
  return DelayableRecordset(self, priority=priority,
  eta=eta,
  max_retries=max_retries,

queue_job/readme/USAGE.rst

@@ -1,3 +1,34 @@
 To use this module, you need to:
 
 #. Go to ``Job Queue`` menu
+
+Developers
+~~~~~~~~~~
+
+**Bypass jobs on running Odoo**
+
+When you are developing (ie: connector modules) you might want
+to bypass the queue job and run your code immediately.
+
+To do so you can set `TEST_QUEUE_JOB_NO_DELAY=1` in your enviroment.
+
+**Bypass jobs in tests**
+
+When writing tests on job-related methods is always tricky to deal with
+delayed recordsets. To make your testing life easier
+you can set `test_queue_job_no_delay=True` in the context.
+
+Tip: you can do this at test case level like this
+
+.. code-block:: python
+
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ cls.env = cls.env(context=dict(
+ cls.env.context,
+ test_queue_job_no_delay=True, # no jobs thanks
+ ))
+
+Then all your tests execute the job methods synchronously
+without delaying any jobs.