This addon adds an integrated Job Queue to Odoo.
@@ -375,19 +375,19 @@Jobs are executed in the background by a Jobrunner, in their own transaction.
Example:
-from odoo import models, fields, api +from odoo import models, fields, api -class MyModel(models.Model): +class MyModel(models.Model): _name = 'my.model' - def my_method(self, a, k=None): + def my_method(self, a, k=None): _logger.info('executed with a: %s and k: %s', a, k) -class MyOtherModel(models.Model): +class MyOtherModel(models.Model): _name = 'my.other.model' - def button_do_stuff(self): + def button_do_stuff(self): self.env['my.model'].with_delay().my_method('a', k=2)
In the snippet of code above, when we call button_do_stuff, a job capturing
@@ -506,7 +506,7 @@ The fast way to enqueue a job for a method is to use with_delay() on a record
or model:Delaying jobs
-def button_done(self):
+def button_done(self):
self.with_delay().print_confirmation_document(self.state)
self.write({"state": "done"})
return True
@@ -522,7 +522,7 @@
Delaying jobs
on a record or model. The following is the equivalent of with_delay() but using the
long form:
-def button_done(self): +def button_done(self): delayable = self.delayable() delayable.print_confirmation_document(self.state) delayable.delay() @@ -532,7 +532,7 @@Delaying jobs
Methods of Delayable objects return itself, so it can be used as a builder pattern, which in some cases allow to build the jobs dynamically:
-def button_generate_simple_with_delayable(self): +def button_generate_simple_with_delayable(self): self.ensure_one() # Introduction of a delayable object, using a builder pattern # allowing to chain jobs or set properties. The delay() method @@ -548,7 +548,7 @@Delaying jobs
The simplest way to define a dependency is to use .on_done(job) on a Delayable:
-def button_chain_done(self): +def button_chain_done(self): self.ensure_one() job1 = self.browse(1).delayable().generate_thumbnail((50, 50)) job2 = self.browse(1).delayable().generate_thumbnail((50, 50)) @@ -565,9 +565,9 @@Delaying jobs
[B] of jobs. When and only when all the jobs of the group [A] are executed, the jobs of the group [B] are executed. The code would look like:-from odoo.addons.queue_job.delay import group, chain +from odoo.addons.queue_job.delay import group, chain -def button_done(self): +def button_done(self): group_a = group(self.delayable().method_foo(), self.delayable().method_bar()) group_b = group(self.delayable().method_baz(1), self.delayable().method_baz(2)) chain(group_a, group_b).delay() @@ -588,7 +588,7 @@Delaying jobs
work. This can be useful to avoid very long jobs, parallelize some task and get more specific errors. Usage is as follows:-def button_split_delayable(self): +def button_split_delayable(self): ( self # Can be a big recordset, let's say 1000 records .delayable() @@ -603,7 +603,7 @@Delaying jobs
True, the jobs will be chained, meaning that the next job will only start when the previous one is done:-def button_increment_var(self): +def button_increment_var(self): ( self .delayable() @@ -687,10 +687,10 @@Configure default options for job
Example of related action code:
-class QueueJob(models.Model): +class QueueJob(models.Model): _inherit = 'queue.job' - def related_action_partner(self, name): + def related_action_partner(self, name): self.ensure_one() model = self.model_name partner = self.records @@ -732,12 +732,12 @@Configure default options for job be customized in Base._job_prepare_context_before_enqueue_keys.
Example:
-class Base(models.AbstractModel): +class Base(models.AbstractModel): _inherit = "base" @api.model - def _job_prepare_context_before_enqueue_keys(self): + def _job_prepare_context_before_enqueue_keys(self): """Keys to keep in context of stored jobs Empty by default for backward compatibility. @@ -755,7 +755,7 @@Configure default options for job
Tip: you can do this at test case level like this
@classmethod -def setUpClass(cls): +def setUpClass(cls): super().setUpClass() cls.env = cls.env(context=dict( cls.env.context, @@ -795,20 +795,20 @@Testing
A very small example (more details in tests/common.py):
# code -def my_job_method(self, name, count): +def my_job_method(self, name, count): self.write({"name": " ".join([name] * count) -def method_to_test(self): +def method_to_test(self): count = self.env["other.model"].search_count([]) self.with_delay(priority=15).my_job_method("Hi!", count=count) return count # tests -from odoo.addons.queue_job.tests.common import trap_jobs +from odoo.addons.queue_job.tests.common import trap_jobs # first test only check the expected behavior of the method and the proper # enqueuing of jobs -def test_method_to_test(self): +def test_method_to_test(self): with trap_jobs() as trap: result = self.env["model"].method_to_test() expected_count = 12 @@ -824,7 +824,7 @@Testing
# second test to validate the behavior of the job unitarily - def test_my_job_method(self): + def test_my_job_method(self): record = self.env["model"].browse(1) record.my_job_method("Hi!", count=12) self.assertEqual(record.name, "Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi! Hi!") @@ -832,7 +832,7 @@Testing
If you prefer, you can still test the whole thing in a single test, by calling jobs_tester.perform_enqueued_jobs() in your test.
-def test_method_to_test(self): +def test_method_to_test(self): with trap_jobs() as trap: result = self.env["model"].method_to_test() expected_count = 12 @@ -867,7 +867,7 @@Testing
Tip: you can do this at test case level like this
@classmethod -def setUpClass(cls): +def setUpClass(cls): super().setUpClass() cls.env = cls.env(context=dict( cls.env.context, diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index c6486e27ef..131ce6322d 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -3,8 +3,57 @@ # pylint: disable=odoo-addons-relative-import # we are testing, we want to test as we were an external consumer of the API +import os + +from odoo.tests import BaseCase, tagged + from odoo.addons.queue_job.jobrunner import runner from .common import load_doctests load_tests = load_doctests(runner) + + +@tagged("-at_install", "post_install") +class TestRunner(BaseCase): + @classmethod + def _is_open_file_descriptor(cls, fd): + try: + os.fstat(fd) + return True + except OSError: + return False + + def test_runner_file_descriptor(self): + a_runner = runner.QueueJobRunner.from_environ_or_config() + + read_fd, write_fd = a_runner._stop_pipe + self.assertTrue(self._is_open_file_descriptor(read_fd)) + self.assertTrue(self._is_open_file_descriptor(write_fd)) + + del a_runner + + self.assertFalse(self._is_open_file_descriptor(read_fd)) + self.assertFalse(self._is_open_file_descriptor(write_fd)) + + def test_runner_file_closed_read_descriptor(self): + a_runner = runner.QueueJobRunner.from_environ_or_config() + + read_fd, write_fd = a_runner._stop_pipe + os.close(read_fd) + + del a_runner + + self.assertFalse(self._is_open_file_descriptor(read_fd)) + self.assertFalse(self._is_open_file_descriptor(write_fd)) + + def test_runner_file_closed_write_descriptor(self): + a_runner = runner.QueueJobRunner.from_environ_or_config() + + read_fd, write_fd = a_runner._stop_pipe + os.close(write_fd) + + del a_runner + + self.assertFalse(self._is_open_file_descriptor(read_fd)) + self.assertFalse(self._is_open_file_descriptor(write_fd))