Job scheduling (#1163)

* First RQScheduler prototype

* WIP job scheduling

* Fixed Python 2.7 tests

* Added ScheduledJobRegistry.get_scheduled_time(job)

* WIP on scheduler's threading mechanism

* Fixed test errors

* Changed scheduler.acquire_locks() to instance method

* Added scheduler.prepare_registries()

* Somewhat working implementation of RQ scheduler

* Only call stop_scheduler if there's a scheduler present

* Use OSError rather than ProcessLookupError for PyPy compatibility

* Added `auto_start` argument to scheduler.acquire_locks()

* Make RQScheduler play better with timezone

* Fixed test error

* Added --with-scheduler flag to rq worker CLI

* Fix tests on Python 2.x

* More Python 2 fixes

* Only call `scheduler.start` if worker is run in non burst mode

* Fixed an issue where running worker with scheduler would fail sometimes

* Make `worker.stop_scheduler()` more resilient to errors

* worker.dequeue_job_and_maintain_ttl() should also periodically run maintenance tasks

* Scheduler can now work with worker in both burst and non burst mode

* Fixed scheduler logging message

* Always log scheduler errors when running

* Improve scheduler error logging message

* Removed testing code

* Scheduler should periodically try to acquire locks for other queues it doesn't have

* Added tests for scheduler.should_reacquire_locks

* Added queue.enqueue_in()

* Fixes queue.enqueue_in() in Python 2.7

* First stab at documenting job scheduling

* Remove unused methods

* Remove Python 2.6 logging compatibility code

* Remove more unused imports

* Added convenience methods to access job registries from queue

* Added test for worker.run_maintenance_tasks()

* Simplify worker.queue_names() and worker.queue_keys()

* Updated changelog to mention RQ's new job scheduling mechanism.

1 files changed, 20 insertions, 2 deletions

diff --git a/tests/test_cli.py b/tests/test_cli.py
index 343e286..1dea1e3 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -2,14 +2,17 @@
 from __future__ import (absolute_import, division, print_function,
                         unicode_literals)
 
+from datetime import datetime
+
 from click.testing import CliRunner
 from redis import Redis
 
 from rq import Queue
+from rq.compat import utc
 from rq.cli import main
 from rq.cli.helpers import read_config_file, CliConfig
 from rq.job import Job
-from rq.registry import FailedJobRegistry
+from rq.registry import FailedJobRegistry, ScheduledJobRegistry
 from rq.worker import Worker, WorkerStatus
 
 import pytest
@@ -244,6 +247,21 @@ class TestRQCli(RQTestCase):
         self.assertTrue(len(pid.read()) > 0)
         self.assert_normal_execution(result)
 
+    def test_worker_with_scheduler(self):
+        """rq worker -u <url> --with-scheduler"""
+        queue = Queue(connection=self.connection)
+        queue.enqueue_at(datetime(2019, 1, 1, tzinfo=utc), say_hello)
+        registry = ScheduledJobRegistry(queue=queue)
+
+        runner = CliRunner()
+        result = runner.invoke(main, ['worker', '-u', self.redis_url, '-b'])
+        self.assert_normal_execution(result)
+        self.assertEqual(len(registry), 1)  # 1 job still scheduled
+
+        result = runner.invoke(main, ['worker', '-u', self.redis_url, '-b', '--with-scheduler'])
+        self.assert_normal_execution(result)
+        self.assertEqual(len(registry), 0)  # Job has been enqueued
+
     def test_worker_logging_options(self):
         """--quiet and --verbose logging options are supported"""
         runner = CliRunner()
@@ -256,7 +274,7 @@ class TestRQCli(RQTestCase):
         # --quiet and --verbose are mutually exclusive
         result = runner.invoke(main, args + ['--quiet', '--verbose'])
         self.assertNotEqual(result.exit_code, 0)
-    
+
     def test_exception_handlers(self):
         """rq worker -u <url> -b --exception-handler <handler>"""
         connection = Redis.from_url(self.redis_url)