From 92aa83d9cc927febaef4bc7502658aab209a6d02 Mon Sep 17 00:00:00 2001 From: Tom Date: Thu, 8 Jan 2026 15:57:09 +0100 Subject: [PATCH 01/13] [FIX] test_queue_job: also add an _unregister_hook function that reverts patches Otherwise monkey-patches could be added twice when instantiating new test classes. --- test_queue_job/models/test_models.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/test_queue_job/models/test_models.py b/test_queue_job/models/test_models.py index fc9e43edf8..4b626d99ee 100644 --- a/test_queue_job/models/test_models.py +++ b/test_queue_job/models/test_models.py @@ -158,6 +158,13 @@ def _register_hook(self): ) return super()._register_hook() + def _unregister_hook(self): + """Remove the patches installed by _register_hook()""" + self._revert_method("delay_me") + self._revert_method("delay_me_options") + self._revert_method("delay_me_context_key") + return super()._unregister_hook() + def _job_store_values(self, job): value = "JUST_TESTING" if job.state == "failed": From 2bf16433e59366dc1c73dfcfdef3e2f732541ed5 Mon Sep 17 00:00:00 2001 From: Adam Heinz Date: Thu, 30 Apr 2026 15:26:39 -0400 Subject: [PATCH 02/13] [FIX] queue_job: Restore doctesting. --- queue_job/tests/common.py | 51 ------------------------- queue_job/tests/test_runner_channels.py | 9 ++++- queue_job/tests/test_runner_runner.py | 12 +++--- 3 files changed, 14 insertions(+), 58 deletions(-) diff --git a/queue_job/tests/common.py b/queue_job/tests/common.py index f74e4ad651..14d7cabba3 100644 --- a/queue_job/tests/common.py +++ b/queue_job/tests/common.py @@ -1,13 +1,10 @@ # Copyright 2019 Camptocamp # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). -import doctest -import logging import typing from contextlib import contextmanager from operator import attrgetter from unittest import TestCase, mock -from odoo.tests.case import TestCase as _TestCase from odoo.tools import groupby from odoo.addons.queue_job.delay import Graph @@ -417,51 +414,3 @@ def test_export(self): delayable = mock.MagicMock(name="DelayableBinding") delayable_cls.return_value = delayable yield delayable_cls, delayable - - -class OdooDocTestCase(doctest.DocTestCase, _TestCase): - """ - We need a custom DocTestCase class in order to: - - define test_tags to run as part of standard tests - - output a more meaningful test name than default "DocTestCase.runTest" - """ - - def __init__( - self, doctest, optionflags=0, setUp=None, tearDown=None, checker=None, seq=0 - ): - super().__init__( - doctest._dt_test, - optionflags=optionflags, - setUp=setUp, - tearDown=tearDown, - checker=checker, - ) - self.test_sequence = seq - - def setUp(self): - """Log an extra statement which test is started.""" - super().setUp() - logging.getLogger(__name__).info("Running tests for %s", self._dt_test.name) - - -def load_doctests(module): - """ - Generates a tests loading method for the doctests of the given module - https://docs.python.org/3/library/unittest.html#load-tests-protocol - """ - - def load_tests(loader, tests, ignore): - """ - Apply the 'test_tags' attribute to each DocTestCase found by the DocTestSuite. - Also extend the DocTestCase class trivially to fit the class teardown - that Odoo backported for its own test classes from Python 3.8. - """ - - for idx, test in enumerate(doctest.DocTestSuite(module)): - odoo_test = OdooDocTestCase(test, seq=idx) - odoo_test.test_tags = {"standard", "at_install", "queue_job", "doctest"} - tests.addTest(odoo_test) - - return tests - - return load_tests diff --git a/queue_job/tests/test_runner_channels.py b/queue_job/tests/test_runner_channels.py index d323d00683..e7124abce8 100644 --- a/queue_job/tests/test_runner_channels.py +++ b/queue_job/tests/test_runner_channels.py @@ -1,10 +1,15 @@ # Copyright 2015-2016 Camptocamp SA # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) +import doctest + +from odoo.tests import BaseCase, tagged # pylint: disable=odoo-addons-relative-import # we are testing, we want to test as we were an external consumer of the API from odoo.addons.queue_job.jobrunner import channels -from .common import load_doctests -load_tests = load_doctests(channels) +@tagged("doctest") +class TestDoctest(BaseCase): + def test_doctest(self): + doctest.testmod(channels, exclude_empty=True, raise_on_error=True) diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index 2adbf610db..2cb388faf7 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -1,19 +1,21 @@ # Copyright 2015-2016 Camptocamp SA # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html) - -# pylint: disable=odoo-addons-relative-import -# we are testing, we want to test as we were an external consumer of the API +import doctest import os import threading from unittest.mock import patch from odoo.tests import BaseCase, tagged +# pylint: disable=odoo-addons-relative-import +# we are testing, we want to test as we were an external consumer of the API from odoo.addons.queue_job.jobrunner import START_DELAY, QueueJobRunnerThread, runner -from .common import load_doctests -load_tests = load_doctests(runner) +@tagged("doctest") +class TestDoctest(BaseCase): + def test_doctest(self): + doctest.testmod(runner, exclude_empty=True, raise_on_error=True) @tagged("-at_install", "post_install") From 4ce93bfe694cb5fcf8f2d5f37a9a1cfef13e263a Mon Sep 17 00:00:00 2001 From: Adam Heinz Date: Tue, 14 Jul 2026 10:30:43 -0400 Subject: [PATCH 03/13] [IMP] queue_job: Set doctest.REPORT_ONLY_FIRST_FAILURE option. --- queue_job/tests/test_runner_channels.py | 4 +++- queue_job/tests/test_runner_runner.py | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/queue_job/tests/test_runner_channels.py b/queue_job/tests/test_runner_channels.py index e7124abce8..b00dd26261 100644 --- a/queue_job/tests/test_runner_channels.py +++ b/queue_job/tests/test_runner_channels.py @@ -12,4 +12,6 @@ @tagged("doctest") class TestDoctest(BaseCase): def test_doctest(self): - doctest.testmod(channels, exclude_empty=True, raise_on_error=True) + doctest.testmod( + channels, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE + ) diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index 2cb388faf7..d65e9fa21a 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -15,7 +15,9 @@ @tagged("doctest") class TestDoctest(BaseCase): def test_doctest(self): - doctest.testmod(runner, exclude_empty=True, raise_on_error=True) + doctest.testmod( + runner, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE + ) @tagged("-at_install", "post_install") From 785183df058d769cf9a89bb96cb804d13d4c478d Mon Sep 17 00:00:00 2001 From: Adam Heinz Date: Wed, 26 Feb 2025 09:46:48 -0500 Subject: [PATCH 04/13] [IMP] queue_job: Default subchannel capacity and sequential. This adds `capacity_default` and `sequential_default` options that set `capacity` and `sequential` for autocreated subchannels. It also allows non-root channel configurations to omit capacity. --- queue_job/jobrunner/channels.py | 88 ++++++++++++++++++++----- queue_job/tests/test_runner_channels.py | 3 +- queue_job/tests/test_runner_runner.py | 3 +- 3 files changed, 77 insertions(+), 17 deletions(-) diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py index b273b9cf47..10ee56fda5 100644 --- a/queue_job/jobrunner/channels.py +++ b/queue_job/jobrunner/channels.py @@ -408,7 +408,15 @@ class Channel: """ def __init__( - self, name, parent, capacity=None, sequential=False, throttle=0, paused=False + self, + name, + parent, + capacity=None, + sequential=False, + throttle=0, + paused=False, + capacity_default=None, + sequential_default=False, ): self.name = name self.parent = parent @@ -419,9 +427,13 @@ def __init__( self._running = set() self._failed = set() self._pause_until = 0 # utc seconds since the epoch - self.capacity = capacity + self.capacity = ( + capacity if (capacity is not None) else (parent and parent.capacity_default) + ) + self.capacity_default = capacity_default self.throttle = throttle # seconds - self.sequential = sequential + self.sequential = sequential or (parent and parent.sequential_default) + self.sequential_default = sequential_default self.paused = paused @property @@ -438,13 +450,17 @@ def configure(self, config): Supported keys are: * capacity + * capacity_default (default for sub channels) * sequential + * sequential_default (default for sub channels) * throttle * paused """ assert self.fullname.endswith(config["name"]) self.capacity = config.get("capacity", None) + self.capacity_default = config.get("capacity_default", None) self.sequential = bool(config.get("sequential", False)) + self.sequential_default = config.get("sequential_default", False) self.throttle = int(config.get("throttle", 0)) self.paused = bool(config.get("paused", False)) if self.sequential and self.capacity != 1: @@ -899,22 +915,23 @@ def parse_simple_config(cls, config_string): continue config = {} config_items = split_strip(channel_config_string, ":") - name = config_items[0] - if not name: + if not (name := config_items.pop(0)): raise ValueError( f"Invalid channel config {config_string}: missing channel name" ) config["name"] = name - if len(config_items) > 1: - capacity = config_items[1] + if len(config_items) > 0: try: - config["capacity"] = int(capacity) + config["capacity"] = int(config_items[0]) + config_items.pop(0) except Exception as ex: - raise ValueError( - f"Invalid channel config {config_string}: " - f"invalid capacity {capacity}" - ) from ex - for config_item in config_items[2:]: + if name == "root": + raise ValueError( + f"Invalid channel config {config_string}: " + f"invalid capacity {config_items[0]}" + ) from ex + + for config_item in config_items: kv = split_strip(config_item, "=") if len(kv) == 1: k, v = kv[0], True @@ -929,7 +946,16 @@ def parse_simple_config(cls, config_string): raise ValueError( f"Invalid channel config {config_string}: duplicate key {k}" ) - config[k] = v + if k == "capacity_default": + try: + config[k] = int(v) + except Exception as ex: + raise ValueError( + f"Invalid channel config {config_string}: " + f"invalid capacity_default {v}" + ) from ex + else: + config[k] = v else: config["capacity"] = 1 res.append(config) @@ -942,6 +968,17 @@ def simple_configure(self, config_string): >>> c = cm.get_channel_by_name('root') >>> c.capacity 1 + + >>> cm.simple_configure('root:bogus') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:bogus: invalid capacity bogus + + >>> cm.simple_configure('root:4,:2') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:4,:2: missing channel name + >>> cm.simple_configure('root:4,autosub.sub:2,seq:1:sequential') >>> cm.get_channel_by_name('root').capacity 4 @@ -958,7 +995,28 @@ def simple_configure(self, config_string): 1 >>> cm.get_channel_by_name('seq').sequential True - """ + + >>> cm.simple_configure('root:4:capacity_default=bogus') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:4:capacity_default=bogus: invalid capacity_default bogus + + >>> cm.simple_configure('root:4,sub:3:capacity_default=2') + >>> cm.get_channel_by_name('root.sub').capacity + 3 + >>> cm.get_channel_by_name('root.sub.auto', autocreate=True).capacity + 2 + + >>> cm.simple_configure('root:4,seq:2:sequential') + Traceback (most recent call last): + ... + ValueError: A sequential channel must have a capacity of 1 + + >>> cm.simple_configure('root:4,seq:sequential_default') + >>> cm.get_channel_by_name('root.seq.auto', autocreate=True).sequential + True + + """ # noqa: E501 for config in ChannelManager.parse_simple_config(config_string): self.get_channel_from_config(config) diff --git a/queue_job/tests/test_runner_channels.py b/queue_job/tests/test_runner_channels.py index b00dd26261..313e350bd3 100644 --- a/queue_job/tests/test_runner_channels.py +++ b/queue_job/tests/test_runner_channels.py @@ -12,6 +12,7 @@ @tagged("doctest") class TestDoctest(BaseCase): def test_doctest(self): - doctest.testmod( + results = doctest.testmod( channels, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE ) + self.assertEqual(results.failed, 0, "doctest failed") diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index d65e9fa21a..0c1b8006d2 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -15,9 +15,10 @@ @tagged("doctest") class TestDoctest(BaseCase): def test_doctest(self): - doctest.testmod( + results = doctest.testmod( runner, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE ) + self.assertEqual(results.failed, 0, "doctest failed") @tagged("-at_install", "post_install") From b241cef4512c94eb98b4fffc124c7493a45ac88a Mon Sep 17 00:00:00 2001 From: Guewen Baconnier Date: Mon, 17 Aug 2026 17:59:36 +0200 Subject: [PATCH 05/13] [ADD] queue_job: add jobrunner config on channels Not used at this point by the jobrunner, but the changes on channels trigger a notify to the jobrunner. --- queue_job/jobrunner/channels.py | 1 + queue_job/models/queue_job_channel.py | 102 +++++++++++++++- queue_job/tests/test_model_job_channel.py | 127 ++++++++++++++++++++ queue_job/views/queue_job_channel_views.xml | 80 +++++++++--- 4 files changed, 293 insertions(+), 17 deletions(-) diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py index 10ee56fda5..7ce0d0b75e 100644 --- a/queue_job/jobrunner/channels.py +++ b/queue_job/jobrunner/channels.py @@ -10,6 +10,7 @@ from ..exception import ChannelNotFound from ..job import CANCELLED, DONE, ENQUEUED, FAILED, PENDING, STARTED, WAIT_DEPENDENCIES +RELOAD_PAYLOAD = "reload" NOT_DONE = (WAIT_DEPENDENCIES, PENDING, ENQUEUED, STARTED, FAILED) JobSortingKey = namedtuple("SortingKey", "eta priority date_created seq") diff --git a/queue_job/models/queue_job_channel.py b/queue_job/models/queue_job_channel.py index 0a5c1c917b..fe63366d82 100644 --- a/queue_job/models/queue_job_channel.py +++ b/queue_job/models/queue_job_channel.py @@ -4,12 +4,28 @@ from odoo import api, exceptions, fields, models +from ..jobrunner.channels import RELOAD_PAYLOAD + class QueueJobChannel(models.Model): _name = "queue.job.channel" _description = "Job Channels" _rec_name = "complete_name" + # fields that trigger a reload of the jobrunner for this database when changed + _JOBRUNNER_CONFIG_FIELDS = frozenset( + ( + "capacity", + "sequential", + "throttle", + "paused", + "name", + "parent_id", + "capacity_default", + "sequential_default", + ) + ) + name = fields.Char() complete_name = fields.Char( compute="_compute_complete_name", store=True, readonly=True, recursive=True @@ -23,7 +39,33 @@ class QueueJobChannel(models.Model): string="Job Functions", ) removal_interval = fields.Integer( - default=lambda self: self.env["queue.job"]._removal_interval, required=True + default=lambda self: self.env["queue.job"]._removal_interval, + required=True, + help="Number of days after which done jobs are deleted.", + ) + capacity = fields.Integer( + help="Maximum number of jobs running at the same time in this channel. " + "0 means no limit, but they are still limited by the capacity of the parent " + "channel. On the root channel, 0 is limited by the global server-side " + "configuration." + ) + sequential = fields.Boolean( + help="Jobs are executed one after the other and failed jobs block the channel. " + "Requires a capacity of 1." + ) + throttle = fields.Integer( + help="Minimum delay in seconds between the start of two jobs in this channel." + ) + paused = fields.Boolean( + help="A paused channel (an its sub-channels) do not execute any jobs until " + "resumed." + ) + capacity_default = fields.Integer( + help="Default capacity for unconfigured sub-channels. " + "0 means they would have the same capacity as the current channel." + ) + sequential_default = fields.Boolean( + help="If sequential is enabled for unconfigured sub-channels." ) _name_uniq = models.Constraint( @@ -31,6 +73,35 @@ class QueueJobChannel(models.Model): "Channel complete name must be unique", ) + @api.constrains( + "capacity", "sequential", "throttle", "capacity_default", "sequential_default" + ) + def _check_jobrunner_configuration(self): + for record in self: + if record.capacity < 0: + raise exceptions.ValidationError( + self.env._("The capacity of a channel cannot be negative.") + ) + if record.throttle < 0: + raise exceptions.ValidationError( + self.env._("The throttle of a channel cannot be negative.") + ) + if record.sequential and record.capacity != 1: + raise exceptions.ValidationError( + self.env._("A sequential channel must have a capacity of 1.") + ) + if record.capacity_default < 0: + raise exceptions.ValidationError( + self.env._("The default capacity of a channel cannot be negative.") + ) + if record.sequential_default and record.capacity_default != 1: + raise exceptions.ValidationError( + self.env._( + "A channel with a sequential default must have a " + "default capacity of 1." + ) + ) + @api.depends("name", "parent_id.complete_name") def _compute_complete_name(self): for record in self: @@ -72,8 +143,17 @@ def create(self, vals_list): new_vals_list.append(vals) vals_list = new_vals_list records |= super().create(vals_list) + records._notify_channel_config_changed() return records + @api.onchange("capacity") + def _onchange_capacity(self): + self.capacity_default = self.capacity + + @api.onchange("sequential") + def _onchange_sequential(self): + self.sequential_default = self.sequential + def write(self, values): for channel in self: if ( @@ -83,7 +163,10 @@ def write(self, values): ): msg = self.env._("Cannot change the root channel") raise exceptions.UserError(msg) - return super().write(values) + res = super().write(values) + if self._JOBRUNNER_CONFIG_FIELDS.intersection(values): + self._notify_channel_config_changed() + return res @api.ondelete(at_uninstall=False) def _check_not_root_ondelete(self): @@ -91,3 +174,18 @@ def _check_not_root_ondelete(self): if channel.name == "root": msg = self.env._("Cannot remove the root channel") raise exceptions.UserError(msg) + + def unlink(self): + res = super().unlink() + self._notify_channel_config_changed() + return res + + def action_pause(self): + self.write({"paused": True}) + + def action_resume(self): + self.write({"paused": False}) + + def _notify_channel_config_changed(self): + """Notify the jobrunner to reload its configuration""" + self.env.cr.execute("SELECT pg_notify('queue_job', %s)", (RELOAD_PAYLOAD,)) diff --git a/queue_job/tests/test_model_job_channel.py b/queue_job/tests/test_model_job_channel.py index 20ebbc0bfe..37a1f89485 100644 --- a/queue_job/tests/test_model_job_channel.py +++ b/queue_job/tests/test_model_job_channel.py @@ -1,9 +1,12 @@ # copyright 2018 Camptocamp # license lgpl-3.0 or later (http://www.gnu.org/licenses/lgpl.html) +from unittest import mock + from psycopg2 import IntegrityError import odoo +from odoo import exceptions from odoo.tests import common @@ -57,3 +60,127 @@ def test_channel_display_name(self): {"name": "test", "parent_id": self.root_channel.id} ) self.assertEqual(channel.display_name, channel.complete_name) + + def test_capacity_should_not_be_negative(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "The capacity of a channel cannot be negative.", + ): + self.Channel.create( + { + "name": "test_capacity", + "parent_id": self.root_channel.id, + "capacity": -1, + } + ) + + def test_capacity_default_should_not_be_negative(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "The default capacity of a channel cannot be negative.", + ): + self.Channel.create( + { + "name": "test_capacity_default", + "parent_id": self.root_channel.id, + "capacity_default": -1, + } + ) + + def test_throttle_should_not_be_negative(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "The throttle of a channel cannot be negative.", + ): + self.Channel.create( + { + "name": "test_throttle", + "parent_id": self.root_channel.id, + "throttle": -1, + } + ) + + def test_sequential_should_have_capacity_one(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "A sequential channel must have a capacity of 1.", + ): + self.Channel.create( + { + "name": "test_sequential", + "parent_id": self.root_channel.id, + "sequential": True, + "capacity": 2, + } + ) + + def test_sequential_default_should_have_capacity_default_one(self): + with self.assertRaisesRegex( + exceptions.ValidationError, + "A channel with a sequential default must have a default capacity of 1.", + ): + self.Channel.create( + { + "name": "test_sequential_default", + "parent_id": self.root_channel.id, + "sequential_default": True, + "capacity_default": 2, + } + ) + + def test_action_pause(self): + self.root_channel.action_pause() + self.assertTrue(self.root_channel.paused) + + def test_action_resume(self): + self.root_channel.paused = True + self.root_channel.action_resume() + self.assertFalse(self.root_channel.paused) + + def _patch_notify(self): + return mock.patch.object( + type(self.Channel), "_notify_channel_config_changed", autospec=True + ) + + def test_notify_create_channel(self): + with self._patch_notify() as notify: + self.Channel.create( + { + "name": "create_notify", + "parent_id": self.root_channel.id, + "capacity": 2, + } + ) + notify.assert_called_once() + + def test_notify_write_jobrunner_config(self): + channel = self.Channel.create( + {"name": "write_notify", "parent_id": self.root_channel.id} + ) + with self._patch_notify() as notify: + channel.capacity = 3 + notify.assert_called_once() + + with self._patch_notify() as notify: + channel.paused = True + notify.assert_called_once() + + with self._patch_notify() as notify: + channel.removal_interval = 60 + notify.assert_not_called() + + with self._patch_notify() as notify: + channel.capacity_default = 1 + notify.assert_called_once() + + with self._patch_notify() as notify: + channel.sequential_default = True + notify.assert_called_once() + + def test_notify_unlink_channel(self): + channel = self.Channel.create( + {"name": "unlink_notify", "parent_id": self.root_channel.id} + ) + with self._patch_notify() as notify: + channel.unlink() + notify.assert_called_once() diff --git a/queue_job/views/queue_job_channel_views.xml b/queue_job/views/queue_job_channel_views.xml index 50c245716b..70e309531f 100644 --- a/queue_job/views/queue_job_channel_views.xml +++ b/queue_job/views/queue_job_channel_views.xml @@ -5,23 +5,66 @@ queue.job.channel
- - +