import math from datetime import timedelta from unittest import mock from unittest.mock import ANY import pytest from rq.defaults import DEFAULT_FAILURE_TTL from rq.exceptions import AbandonedJobError, InvalidJobOperation from rq.executions import Execution from rq.job import Dependency, Job, JobStatus, requeue_job from rq.queue import Queue from rq.registry import ( BaseRegistry, CanceledJobRegistry, DeferredJobRegistry, FailedJobRegistry, FinishedJobRegistry, StartedJobRegistry, clean_registries, ) from rq.serializers import JSONSerializer from rq.utils import as_text, current_timestamp, now from rq.worker import Worker from tests import RQTestCase from tests.fixtures import div_by_zero, say_hello class CustomJob(Job): """A custom job class just to test it""" class TestRegistry(RQTestCase): """Test all the BaseRegistry functionality""" def setUp(self): super().setUp() self.registry = BaseRegistry(connection=self.connection) def test_init(self): """Registry can be instantiated with queue or name/Redis connection""" queue = Queue('foo', connection=self.connection) registry = BaseRegistry(queue=queue) self.assertEqual(registry.name, queue.name) self.assertEqual(registry.connection, queue.connection) self.assertEqual(registry.serializer, queue.serializer) registry = BaseRegistry('bar', self.connection, serializer=JSONSerializer) self.assertEqual(registry.name, 'bar') self.assertEqual(registry.connection, self.connection) self.assertEqual(registry.serializer, JSONSerializer) def test_key(self): self.assertEqual(self.registry.key, 'rq:registry:default') def test_custom_job_class(self): registry = BaseRegistry(job_class=CustomJob) self.assertIsNot(registry.job_class, self.registry.job_class) def test_contains(self): queue = Queue(connection=self.connection) job = queue.enqueue(say_hello) self.assertNotIn(job, self.registry) self.assertNotIn(job.id, self.registry) self.registry.add(job, 5) self.assertIn(job, self.registry) self.assertIn(job.id, self.registry) def test_get_expiration_time(self): """registry.get_expiration_time() returns correct datetime objects""" queue = Queue(connection=self.connection) job = queue.enqueue(say_hello) self.registry.add(job, 5) time = self.registry.get_expiration_time(job) expected_time = (now() + timedelta(seconds=5)).replace(microsecond=0) self.assertGreaterEqual(time, expected_time - timedelta(seconds=2)) self.assertLessEqual(time, expected_time + timedelta(seconds=2)) def test_add_and_remove(self): """Adding and removing job from BaseRegistry.""" timestamp = current_timestamp() queue = Queue(connection=self.connection) job = queue.enqueue(say_hello) # Test that job is added with the right score self.registry.add(job, 1000) self.assertLess(self.connection.zscore(self.registry.key, job.id), timestamp + 1002) # Ensure that a timeout of -1 results in a score of inf self.registry.add(job, -1) self.assertEqual(self.connection.zscore(self.registry.key, job.id), float('inf')) # Ensure that job is removed from sorted set, but job key is not deleted self.registry.remove(job) self.assertIsNone(self.connection.zscore(self.registry.key, job.id)) self.assertTrue(self.connection.exists(job.key)) self.registry.add(job, -1) # registry.remove() also accepts job.id self.registry.remove(job.id) self.assertIsNone(self.connection.zscore(self.registry.key, job.id)) self.registry.add(job, -1) # delete_job = True deletes job key self.registry.remove(job, delete_job=True) self.assertIsNone(self.connection.zscore(self.registry.key, job.id)) self.assertFalse(self.connection.exists(job.key)) job = queue.enqueue(say_hello) self.registry.add(job, -1) # delete_job = True also works with job.id self.registry.remove(job.id, delete_job=True) self.assertIsNone(self.connection.zscore(self.registry.key, job.id)) self.assertFalse(self.connection.exists(job.key)) def test_add_and_remove_with_serializer(self): """Adding and removing job from BaseRegistry (with serializer).""" # delete_job = True also works with job.id and custom serializer queue = Queue(connection=self.connection, serializer=JSONSerializer) registry = BaseRegistry(connection=self.connection, serializer=JSONSerializer) job = queue.enqueue(say_hello) registry.add(job, -1) registry.remove(job.id, delete_job=True) self.assertIsNone(self.connection.zscore(registry.key, job.id)) self.assertFalse(self.connection.exists(job.key)) def test_get_job_ids(self): """Getting job ids from BaseRegistry.""" timestamp = current_timestamp() self.connection.zadd(self.registry.key, {'will-be-cleaned-up': 1}) self.connection.zadd(self.registry.key, {'foo': timestamp + 10}) self.connection.zadd(self.registry.key, {'bar': timestamp + 20}) self.assertEqual(self.registry.get_job_ids(), ['will-be-cleaned-up', 'foo', 'bar']) def test_get_expired_job_ids(self): """Getting expired job ids form BaseRegistry.""" timestamp = current_timestamp() self.connection.zadd(self.registry.key, {'foo': 1}) self.connection.zadd(self.registry.key, {'bar': timestamp + 10}) self.connection.zadd(self.registry.key, {'baz': timestamp + 30}) self.assertEqual(self.registry.get_expired_job_ids(), ['foo']) self.assertEqual(self.registry.get_expired_job_ids(timestamp + 20), ['foo', 'bar']) # CanceledJobRegistry does not implement get_expired_job_ids() registry = CanceledJobRegistry(connection=self.connection) self.assertRaises(NotImplementedError, registry.get_expired_job_ids) def test_count(self): """BaseRegistry returns the right number of job count.""" timestamp = current_timestamp() + 10 self.connection.zadd(self.registry.key, {'will-be-cleaned-up': 1}) self.connection.zadd(self.registry.key, {'foo': timestamp}) self.connection.zadd(self.registry.key, {'bar': timestamp}) self.assertEqual(self.registry.count, 3) self.assertEqual(len(self.registry), 3) def test_get_job_count(self): """Ensure cleanup is not called and does not affect the reported number of jobs. Note, the original motivation to stop calling cleanup was to make the count operation O(1) to allow usage of monitoring tools and avoid side effects of failure callbacks that cleanup triggers. """ timestamp = current_timestamp() + 10 self.connection.zadd(self.registry.key, {'will-be-counted-despite-outdated': 1}) self.connection.zadd(self.registry.key, {'foo': timestamp}) self.connection.zadd(self.registry.key, {'bar': timestamp}) with mock.patch.object(self.registry, 'cleanup') as mock_cleanup: self.assertEqual(self.registry.get_job_count(cleanup=False), 3) mock_cleanup.assert_not_called() def test_clean_registries(self): """clean_registries() cleans Started and Finished job registries.""" queue = Queue(connection=self.connection) finished_job_registry = FinishedJobRegistry(connection=self.connection) self.connection.zadd(finished_job_registry.key, {'foo': 1}) started_job_registry = StartedJobRegistry(connection=self.connection) self.connection.zadd(started_job_registry.key, {'foo:execution_id': 1}) failed_job_registry = FailedJobRegistry(connection=self.connection) self.connection.zadd(failed_job_registry.key, {'foo': 1}) clean_registries(queue) self.assertEqual(self.connection.zcard(finished_job_registry.key), 0) self.assertEqual(self.connection.zcard(started_job_registry.key), 0) self.assertEqual(self.connection.zcard(failed_job_registry.key), 0) def test_clean_registries_with_serializer(self): """clean_registries() cleans Started and Finished job registries (with serializer).""" queue = Queue(connection=self.connection, serializer=JSONSerializer) finished_job_registry = FinishedJobRegistry(connection=self.connection, serializer=JSONSerializer) self.connection.zadd(finished_job_registry.key, {'foo': 1}) started_job_registry = StartedJobRegistry(connection=self.connection, serializer=JSONSerializer) self.connection.zadd(started_job_registry.key, {'foo:execution_id': 1}) failed_job_registry = FailedJobRegistry(connection=self.connection, serializer=JSONSerializer) self.connection.zadd(failed_job_registry.key, {'foo': 1}) clean_registries(queue) self.assertEqual(self.connection.zcard(finished_job_registry.key), 0) self.assertEqual(self.connection.zcard(started_job_registry.key), 0) self.assertEqual(self.connection.zcard(failed_job_registry.key), 0) def test_get_queue(self): """registry.get_queue() returns the right Queue object.""" registry = BaseRegistry(connection=self.connection) self.assertEqual(registry.get_queue(), Queue(connection=self.connection)) registry = BaseRegistry('foo', connection=self.connection, serializer=JSONSerializer) self.assertEqual(registry.get_queue(), Queue('foo', connection=self.connection, serializer=JSONSerializer)) class TestFinishedJobRegistry(RQTestCase): def setUp(self): super().setUp() self.registry = FinishedJobRegistry(connection=self.connection) def test_key(self): self.assertEqual(self.registry.key, 'rq:finished:default') def test_cleanup(self): """Finished job registry removes expired jobs.""" timestamp = current_timestamp() self.connection.zadd(self.registry.key, {'foo': 1}) self.connection.zadd(self.registry.key, {'bar': timestamp + 10}) self.connection.zadd(self.registry.key, {'baz': timestamp + 30}) self.registry.cleanup() self.assertEqual(self.registry.get_job_ids(), ['bar', 'baz']) self.registry.cleanup(timestamp + 20) self.assertEqual(self.registry.get_job_ids(), ['baz']) # CanceledJobRegistry now implements noop cleanup, should not raise exception registry = CanceledJobRegistry(connection=self.connection) registry.cleanup() def test_jobs_are_put_in_registry(self): """Completed jobs are added to FinishedJobRegistry.""" self.assertEqual(self.registry.get_job_ids(), []) queue = Queue(connection=self.connection) worker = Worker([queue], connection=self.connection) # Completed jobs are put in FinishedJobRegistry job = queue.enqueue(say_hello) worker.perform_job(job, queue) self.assertEqual(self.registry.get_job_ids(), [job.id]) # When job is deleted, it should be removed from FinishedJobRegistry self.assertEqual(job.get_status(), JobStatus.FINISHED) job.delete() self.assertEqual(self.registry.get_job_ids(), []) # Failed jobs are not put in FinishedJobRegistry failed_job = queue.enqueue(div_by_zero) worker.perform_job(failed_job, queue) self.assertEqual(self.registry.get_job_ids(), []) class TestDeferredRegistry(RQTestCase): def setUp(self): super().setUp() self.registry = DeferredJobRegistry(connection=self.connection) def test_key(self): self.assertEqual(self.registry.key, 'rq:deferred:default') def test_add(self): """Adding a job to DeferredJobsRegistry.""" job = Job(connection=self.connection) self.registry.add(job) job_ids = [as_text(job_id) for job_id in self.connection.zrange(self.registry.key, 0, -1)] self.assertEqual(job_ids, [job.id]) def test_add_with_deferred_ttl(self): """Job TTL defaults to +inf""" queue = Queue(connection=self.connection) job = queue.enqueue(say_hello) key = self.registry.key self.registry.add(job) score = self.connection.zscore(key, job.id) self.assertEqual(score, float('inf')) timestamp = current_timestamp() ttl = 5 self.registry.add(job, ttl=ttl) score = self.connection.zscore(key, job.id) self.assertLess(score, timestamp + ttl + 2) self.assertGreater(score, timestamp + ttl - 2) def test_register_dependency(self): """Ensure job creation and deletion works with DeferredJobRegistry.""" queue = Queue(connection=self.connection) job = queue.enqueue(say_hello) job2 = queue.enqueue(say_hello, depends_on=job) registry = DeferredJobRegistry(connection=self.connection) self.assertEqual(registry.get_job_ids(), [job2.id]) # When deleted, job removes itself from DeferredJobRegistry job2.delete() self.assertEqual(registry.get_job_ids(), []) def test_cleanup_supports_deleted_jobs(self): queue = Queue(connection=self.connection) job = queue.enqueue(say_hello) self.registry.add(job, ttl=10) self.assertEqual(self.registry.count, 1) job.delete(remove_from_queue=False) self.assertEqual(self.registry.count, 1) self.registry.cleanup(current_timestamp() + 100) self.assertEqual(self.registry.count, 0) def test_cleanup_moves_jobs_to_failed_job_registry(self): """Moving expired jobs to FailedJobRegistry.""" queue = Queue(connection=self.connection) failed_job_registry = FailedJobRegistry(connection=self.connection) job = queue.enqueue(say_hello) self.connection.zadd(self.registry.key, {job.id: 2}) # Job has not been moved to FailedJobRegistry self.registry.cleanup(1) self.assertNotIn(job, failed_job_registry) self.assertIn(job, self.registry) self.registry.cleanup() self.assertIn(job.id, failed_job_registry) self.assertNotIn(job, self.registry) job.refresh() self.assertEqual(job.get_status(), JobStatus.FAILED) self.assertTrue(job.exc_info) # explanation is written to exc_info class TestFailedJobRegistry(RQTestCase): def test_default_failure_ttl(self): """Job TTL defaults to DEFAULT_FAILURE_TTL""" queue = Queue(connection=self.connection) job = queue.enqueue(say_hello) registry = FailedJobRegistry(connection=self.connection) key = registry.key timestamp = current_timestamp() registry.add(job) score = self.connection.zscore(key, job.id) self.assertLess(score, timestamp + DEFAULT_FAILURE_TTL + 2) self.assertGreater(score, timestamp + DEFAULT_FAILURE_TTL - 2) # Job key will also expire job_ttl = self.connection.ttl(job.key) self.assertLess(job_ttl, DEFAULT_FAILURE_TTL + 2) self.assertGreater(job_ttl, DEFAULT_FAILURE_TTL - 2) timestamp = current_timestamp() ttl = 5 registry.add(job, ttl=ttl) score = self.connection.zscore(key, job.id) self.assertLess(score, timestamp + ttl + 2) self.assertGreater(score, timestamp + ttl - 2) job_ttl = self.connection.ttl(job.key) self.assertLess(job_ttl, ttl + 2) self.assertGreater(job_ttl, ttl - 2) def test_requeue(self): """FailedJobRegistry.requeue works properly""" queue = Queue(connection=self.connection) job = queue.enqueue(div_by_zero, failure_ttl=5) worker = Worker([queue], connection=self.connection) worker.work(burst=True) registry = FailedJobRegistry(connection=worker.connection) self.assertIn(job, registry) registry.requeue(job.id) self.assertNotIn(job, registry) self.assertIn(job.id, queue.get_job_ids()) job.refresh() self.assertEqual(job.get_status(), JobStatus.QUEUED) self.assertEqual(job.started_at, None) self.assertEqual(job.ended_at, None) worker.work(burst=True) self.assertIn(job, registry) # Should also work with job instance registry.requeue(job) self.assertNotIn(job, registry) self.assertIn(job.id, queue.get_job_ids()) job.refresh() self.assertEqual(job.get_status(), JobStatus.QUEUED) worker.work(burst=True) self.assertIn(job, registry) # requeue_job should work the same way requeue_job(job.id, connection=self.connection) self.assertNotIn(job, registry) self.assertIn(job.id, queue.get_job_ids()) job.refresh() self.assertEqual(job.get_status(), JobStatus.QUEUED) worker.work(burst=True) self.assertIn(job, registry) # And so does job.requeue() job.requeue() self.assertNotIn(job, registry) self.assertIn(job.id, queue.get_job_ids()) job.refresh() self.assertEqual(job.get_status(), JobStatus.QUEUED) def test_requeue_with_serializer(self): """FailedJobRegistry.requeue works properly (with serializer)""" queue = Queue(connection=self.connection, serializer=JSONSerializer) job = queue.enqueue(div_by_zero, failure_ttl=5) worker = Worker([queue], serializer=JSONSerializer, connection=self.connection) worker.work(burst=True) registry = FailedJobRegistry(connection=worker.connection, serializer=JSONSerializer) self.assertIn(job, registry) registry.requeue(job.id) self.assertNotIn(job, registry) self.assertIn(job.id, queue.get_job_ids()) job.refresh() self.assertEqual(job.get_status(), JobStatus.QUEUED) self.assertEqual(job.started_at, None) self.assertEqual(job.ended_at, None) worker.work(burst=True) self.assertIn(job, registry) # Should also work with job instance registry.requeue(job) self.assertNotIn(job, registry) self.assertIn(job.id, queue.get_job_ids()) job.refresh() self.assertEqual(job.get_status(), JobStatus.QUEUED) worker.work(burst=True) self.assertIn(job, registry) # requeue_job should work the same way requeue_job(job.id, connection=self.connection, serializer=JSONSerializer) self.assertNotIn(job, registry) self.assertIn(job.id, queue.get_job_ids()) job.refresh() self.assertEqual(job.get_status(), JobStatus.QUEUED) worker.work(burst=True) self.assertIn(job, registry) # And so does job.requeue() job.requeue() self.assertNotIn(job, registry) self.assertIn(job.id, queue.get_job_ids()) job.refresh() self.assertEqual(job.get_status(), JobStatus.QUEUED) def test_invalid_job(self): """Requeuing a job that's not in FailedJobRegistry raises an error.""" queue = Queue(connection=self.connection) job = queue.enqueue(say_hello) registry = FailedJobRegistry(connection=self.connection) with self.assertRaises(InvalidJobOperation): registry.requeue(job) def test_worker_handle_job_failure(self): """Failed jobs are added to FailedJobRegistry""" q = Queue(connection=self.connection) w = Worker([q], connection=self.connection) registry = FailedJobRegistry(connection=w.connection) timestamp = current_timestamp() job = q.enqueue(div_by_zero, failure_ttl=5) w.handle_job_failure(job, q) # job is added to FailedJobRegistry with default failure ttl self.assertIn(job.id, registry.get_job_ids()) self.assertLess(self.connection.zscore(registry.key, job.id), timestamp + DEFAULT_FAILURE_TTL + 5) # job is added to FailedJobRegistry with specified ttl job = q.enqueue(div_by_zero, failure_ttl=5) w.handle_job_failure(job, q) self.assertLess(self.connection.zscore(registry.key, job.id), timestamp + 7) class TestStartedJobRegistry(RQTestCase): def setUp(self): super().setUp() self.registry = StartedJobRegistry(connection=self.connection) self.queue = Queue(connection=self.connection) def test_job_deletion(self): """Ensure job is removed from StartedJobRegistry when deleted.""" worker = Worker([self.queue], connection=self.connection) job = self.queue.enqueue(say_hello) self.assertTrue(job.is_queued) execution = worker.prepare_execution(job) worker.prepare_job_execution(job) self.assertIn(execution.job_id, self.registry.get_job_ids()) pipeline = self.connection.pipeline() job.delete(pipeline=pipeline) pipeline.execute() self.assertNotIn(execution.job_id, self.registry.get_job_ids()) def test_contains(self): """Test the StartedJobRegistry __contains__ method. It is slightly different because the entries in the registry are {job_id}:{execution_id} format.""" job = self.queue.enqueue(say_hello) self.assertNotIn(job, self.registry) self.assertNotIn(job.id, self.registry) with self.connection.pipeline() as pipe: self.registry.add_execution( Execution(id='execution', job_id=job.id, connection=self.connection), pipeline=pipe, ttl=5 ) pipe.execute() self.assertIn(job, self.registry) self.assertIn(job.id, self.registry) def test_infinite_score(self): """Test the StartedJobRegistry __contains__ method. It is slightly different because the entries in the registry are {job_id}:{execution_id} format.""" job = self.queue.enqueue(say_hello) self.assertNotIn(job, self.registry) self.assertNotIn(job.id, self.registry) with self.connection.pipeline() as pipe: execution = Execution(id='execution', job_id=job.id, connection=self.connection) self.registry.add_execution(execution=execution, pipeline=pipe, ttl=-1) pipe.execute() self.assertIn(job, self.registry) self.assertIn(job.id, self.registry) self.assertEqual(self.connection.zscore(self.registry.key, execution.composite_key), math.inf) def test_remove_executions(self): """Ensure all executions for a job are removed from registry.""" worker = Worker([self.queue], connection=self.connection) job = self.queue.enqueue(say_hello) execution_1 = worker.prepare_execution(job) execution_2 = worker.prepare_execution(job) self.assertIn((job.id, execution_1.id), self.registry.get_job_and_execution_ids()) self.assertIn((job.id, execution_2.id), self.registry.get_job_and_execution_ids()) self.registry.remove_executions(job) self.assertNotIn((job.id, execution_1.id), self.registry.get_job_and_execution_ids()) self.assertNotIn((job.id, execution_2.id), self.registry.get_job_and_execution_ids()) job.delete() def test_job_execution(self): """Job is removed from StartedJobRegistry after execution.""" worker = Worker([self.queue], connection=self.connection) job = self.queue.enqueue(say_hello) self.assertTrue(job.is_queued) execution = worker.prepare_execution(job) worker.prepare_job_execution(job) self.assertIn(job.id, self.registry.get_job_ids()) self.assertIn((job.id, execution.id), self.registry.get_job_and_execution_ids()) self.assertTrue(job.is_started) worker.perform_job(job, self.queue) self.assertNotIn(job.id, self.registry.get_job_ids()) self.assertNotIn((job.id, execution.id), self.registry.get_job_and_execution_ids()) self.assertTrue(job.is_finished) # Job that fails job = self.queue.enqueue(div_by_zero) execution = worker.prepare_execution(job) worker.prepare_job_execution(job) self.assertIn(job.id, self.registry.get_job_ids()) self.assertIn((job.id, execution.id), self.registry.get_job_and_execution_ids()) worker.perform_job(job, self.queue) self.assertNotIn(job.id, self.registry.get_job_ids()) self.assertNotIn((job.id, execution.id), self.registry.get_job_and_execution_ids()) def test_get_job_ids(self): """Getting job ids with cleanup.""" timestamp = current_timestamp() self.connection.zadd(self.registry.key, {'will-be-cleaned-up:execution_id1': 1}) self.connection.zadd(self.registry.key, {'foo:execution_id2': timestamp + 10}) self.connection.zadd(self.registry.key, {'bar:execution_id3': timestamp + 20}) self.assertEqual(self.registry.get_job_ids(), ['foo', 'bar']) def test_get_job_ids_does_not_cleanup(self): """Getting job ids without a cleanup.""" timestamp = current_timestamp() self.connection.zadd(self.registry.key, {'will-be-returned-despite-outdated:execution_id1': 1}) self.connection.zadd(self.registry.key, {'foo:execution_id2': timestamp + 10}) self.connection.zadd(self.registry.key, {'bar:execution_id3': timestamp + 20}) self.assertEqual(self.registry.get_job_ids(cleanup=False), ['will-be-returned-despite-outdated', 'foo', 'bar']) def test_count(self): """Return the right number of job count (cleanup should be performed)""" timestamp = current_timestamp() + 10 self.connection.zadd(self.registry.key, {'will-be-cleaned-up': 1}) self.connection.zadd(self.registry.key, {'foo': timestamp}) self.connection.zadd(self.registry.key, {'bar': timestamp}) self.assertEqual(self.registry.count, 2) self.assertEqual(len(self.registry), 2) def test_cleanup_moves_jobs_to_failed_job_registry(self): """Moving expired jobs to FailedJobRegistry.""" failed_job_registry = FailedJobRegistry(connection=self.connection) job = self.queue.enqueue(say_hello) self.connection.zadd(self.registry.key, {f'{job.id}:execution_id': 100}) # Job has not been moved to FailedJobRegistry self.registry.cleanup(1) self.assertNotIn(job, failed_job_registry) self.assertIn(job, self.registry) with mock.patch.object(Job, 'execute_failure_callback') as mocked: mock_handler = mock.MagicMock() mock_handler.return_value = False mock_handler_no_return = mock.MagicMock() mock_handler_no_return.return_value = None self.registry.cleanup(exception_handlers=[mock_handler_no_return, mock_handler]) mocked.assert_called_once_with(self.queue.death_penalty_class, AbandonedJobError, ANY, ANY) mock_handler.assert_called_once_with(job, AbandonedJobError, ANY, ANY) mock_handler_no_return.assert_called_once_with(job, AbandonedJobError, ANY, ANY) self.assertIn(job.id, failed_job_registry) self.assertNotIn(job, self.registry) job.refresh() self.assertEqual(job.get_status(), JobStatus.FAILED) latest_result = job.latest_result() self.assertIsNotNone(latest_result) self.assertTrue(latest_result.exc_string) # explanation is written to exc_info def test_enqueue_dependents_when_parent_job_is_abandoned(self): """Enqueuing parent job's dependencies after moving it to FailedJobRegistry due to AbandonedJobError.""" queue = Queue(connection=self.connection) worker = Worker([queue]) failed_job_registry = FailedJobRegistry(connection=self.connection) finished_job_registry = FinishedJobRegistry(connection=self.connection) deferred_job_registry = DeferredJobRegistry(connection=self.connection) parent_job = queue.enqueue(say_hello) job_to_be_executed = queue.enqueue_call(say_hello, depends_on=Dependency(jobs=[parent_job], allow_failure=True)) job_not_to_be_executed = queue.enqueue_call( say_hello, depends_on=Dependency(jobs=[parent_job], allow_failure=False) ) self.assertIn(job_to_be_executed, deferred_job_registry) self.assertIn(job_not_to_be_executed, deferred_job_registry) self.connection.zadd(self.registry.key, {f'{parent_job.id}:execution': 2}) queue.remove(parent_job.id) with mock.patch.object(Job, 'execute_failure_callback') as mocked: self.registry.cleanup() mocked.assert_called_once_with(queue.death_penalty_class, AbandonedJobError, ANY, ANY) # check that parent job was moved to FailedJobRegistry and has correct status self.assertIn(parent_job, failed_job_registry) self.assertNotIn(parent_job, self.registry) self.assertTrue(parent_job.is_failed) # check that only job_to_be_executed has been queued and executed self.assertEqual(len(queue.get_job_ids()), 1) self.assertTrue(job_to_be_executed.is_queued) self.assertFalse(job_not_to_be_executed.is_queued) worker.work(burst=True) self.assertTrue(job_to_be_executed.is_finished) self.assertNotIn(job_to_be_executed, deferred_job_registry) self.assertIn(job_to_be_executed, finished_job_registry) self.assertFalse(job_not_to_be_executed.is_finished) self.assertNotIn(job_not_to_be_executed, finished_job_registry) def test_warnings_on_add_remove_and_exception(self): """Test backwards compatibility of the .add and .remove methods for the StartedJobRegistry.""" with pytest.raises(NotImplementedError): self.registry.add('job_id') with pytest.raises(NotImplementedError): self.registry.remove('job_id')