# -*- coding: utf-8 -*- from __future__ import (absolute_import, division, print_function, unicode_literals) from tests import RQTestCase from tests.fixtures import (div_by_zero, echo, Number, say_hello, some_calculation) from rq import get_failed_queue, Queue from rq.exceptions import InvalidJobDependency, InvalidJobOperationError from rq.job import Job, JobStatus from rq.registry import DeferredJobRegistry from rq.worker import Worker class CustomJob(Job): pass class TestQueue(RQTestCase): def test_create_queue(self): """Creating queues.""" q = Queue('my-queue') self.assertEqual(q.name, 'my-queue') self.assertEqual(str(q), '') def test_create_default_queue(self): """Instantiating the default queue.""" q = Queue() self.assertEqual(q.name, 'default') def test_equality(self): """Mathematical equality of queues.""" q1 = Queue('foo') q2 = Queue('foo') q3 = Queue('bar') self.assertEqual(q1, q2) self.assertEqual(q2, q1) self.assertNotEqual(q1, q3) self.assertNotEqual(q2, q3) self.assertGreater(q1, q3) self.assertRaises(TypeError, lambda: q1 == 'some string') self.assertRaises(TypeError, lambda: q1 < 'some string') def test_empty_queue(self): """Emptying queues.""" q = Queue('example') self.testconn.rpush('rq:queue:example', 'foo') self.testconn.rpush('rq:queue:example', 'bar') self.assertEqual(q.is_empty(), False) q.empty() self.assertEqual(q.is_empty(), True) self.assertIsNone(self.testconn.lpop('rq:queue:example')) def test_empty_removes_jobs(self): """Emptying a queue deletes the associated job objects""" q = Queue('example') job = q.enqueue(say_hello) self.assertTrue(Job.exists(job.id)) q.empty() self.assertFalse(Job.exists(job.id)) def test_queue_is_empty(self): """Detecting empty queues.""" q = Queue('example') self.assertEqual(q.is_empty(), True) self.testconn.rpush('rq:queue:example', 'sentinel message') self.assertEqual(q.is_empty(), False) def test_queue_delete(self): """Test queue.delete properly removes queue""" q = Queue('example') job = q.enqueue(say_hello) job2 = q.enqueue(say_hello) self.assertEqual(2, len(q.get_job_ids())) q.delete() self.assertEqual(0, len(q.get_job_ids())) self.assertEqual(False, self.testconn.exists(job.key)) self.assertEqual(False, self.testconn.exists(job2.key)) self.assertEqual(0, len(self.testconn.smembers(Queue.redis_queues_keys))) self.assertEqual(False, self.testconn.exists(q.key)) def test_queue_delete_but_keep_jobs(self): """Test queue.delete properly removes queue but keeps the job keys in the redis store""" q = Queue('example') job = q.enqueue(say_hello) job2 = q.enqueue(say_hello) self.assertEqual(2, len(q.get_job_ids())) q.delete(delete_jobs=False) self.assertEqual(0, len(q.get_job_ids())) self.assertEqual(True, self.testconn.exists(job.key)) self.assertEqual(True, self.testconn.exists(job2.key)) self.assertEqual(0, len(self.testconn.smembers(Queue.redis_queues_keys))) self.assertEqual(False, self.testconn.exists(q.key)) def test_remove(self): """Ensure queue.remove properly removes Job from queue.""" q = Queue('example') job = q.enqueue(say_hello) self.assertIn(job.id, q.job_ids) q.remove(job) self.assertNotIn(job.id, q.job_ids) job = q.enqueue(say_hello) self.assertIn(job.id, q.job_ids) q.remove(job.id) self.assertNotIn(job.id, q.job_ids) def test_jobs(self): """Getting jobs out of a queue.""" q = Queue('example') self.assertEqual(q.jobs, []) job = q.enqueue(say_hello) self.assertEqual(q.jobs, [job]) # Deleting job removes it from queue job.delete() self.assertEqual(q.job_ids, []) def test_compact(self): """Queue.compact() removes non-existing jobs.""" q = Queue() q.enqueue(say_hello, 'Alice') q.enqueue(say_hello, 'Charlie') self.testconn.lpush(q.key, '1', '2') self.assertEqual(q.count, 4) self.assertEqual(len(q), 4) q.compact() self.assertEqual(q.count, 2) self.assertEqual(len(q), 2) def test_enqueue(self): """Enqueueing job onto queues.""" q = Queue() self.assertEqual(q.is_empty(), True) # say_hello spec holds which queue this is sent to job = q.enqueue(say_hello, 'Nick', foo='bar') job_id = job.id self.assertEqual(job.origin, q.name) # Inspect data inside Redis q_key = 'rq:queue:default' self.assertEqual(self.testconn.llen(q_key), 1) self.assertEqual( self.testconn.lrange(q_key, 0, -1)[0].decode('ascii'), job_id) def test_enqueue_sets_metadata(self): """Enqueueing job onto queues modifies meta data.""" q = Queue() job = Job.create(func=say_hello, args=('Nick',), kwargs=dict(foo='bar')) # Preconditions self.assertIsNone(job.enqueued_at) # Action q.enqueue_job(job) # Postconditions self.assertIsNotNone(job.enqueued_at) def test_pop_job_id(self): """Popping job IDs from queues.""" # Set up q = Queue() uuid = '112188ae-4e9d-4a5b-a5b3-f26f2cb054da' q.push_job_id(uuid) # Pop it off the queue... self.assertEqual(q.count, 1) self.assertEqual(q.pop_job_id(), uuid) # ...and assert the queue count when down self.assertEqual(q.count, 0) def test_dequeue(self): """Dequeueing jobs from queues.""" # Set up q = Queue() result = q.enqueue(say_hello, 'Rick', foo='bar') # Dequeue a job (not a job ID) off the queue self.assertEqual(q.count, 1) job = q.dequeue() self.assertEqual(job.id, result.id) self.assertEqual(job.func, say_hello) self.assertEqual(job.origin, q.name) self.assertEqual(job.args[0], 'Rick') self.assertEqual(job.kwargs['foo'], 'bar') # ...and assert the queue count when down self.assertEqual(q.count, 0) def test_dequeue_deleted_jobs(self): """Dequeueing deleted jobs from queues don't blow the stack.""" q = Queue() for _ in range(1, 1000): job = q.enqueue(say_hello) job.delete() q.dequeue() def test_dequeue_instance_method(self): """Dequeueing instance method jobs from queues.""" q = Queue() n = Number(2) q.enqueue(n.div, 4) job = q.dequeue() # The instance has been pickled and unpickled, so it is now a separate # object. Test for equality using each object's __dict__ instead. self.assertEqual(job.instance.__dict__, n.__dict__) self.assertEqual(job.func.__name__, 'div') self.assertEqual(job.args, (4,)) def test_dequeue_class_method(self): """Dequeueing class method jobs from queues.""" q = Queue() q.enqueue(Number.divide, 3, 4) job = q.dequeue() self.assertEqual(job.instance.__dict__, Number.__dict__) self.assertEqual(job.func.__name__, 'divide') self.assertEqual(job.args, (3, 4)) def test_dequeue_ignores_nonexisting_jobs(self): """Dequeuing silently ignores non-existing jobs.""" q = Queue() uuid = '49f205ab-8ea3-47dd-a1b5-bfa186870fc8' q.push_job_id(uuid) q.push_job_id(uuid) result = q.enqueue(say_hello, 'Nick', foo='bar') q.push_job_id(uuid) # Dequeue simply ignores the missing job and returns None self.assertEqual(q.count, 4) self.assertEqual(q.dequeue().id, result.id) self.assertIsNone(q.dequeue()) self.assertEqual(q.count, 0) def test_dequeue_any(self): """Fetching work from any given queue.""" fooq = Queue('foo') barq = Queue('bar') self.assertEqual(Queue.dequeue_any([fooq, barq], None), None) # Enqueue a single item barq.enqueue(say_hello) job, queue = Queue.dequeue_any([fooq, barq], None) self.assertEqual(job.func, say_hello) self.assertEqual(queue, barq) # Enqueue items on both queues barq.enqueue(say_hello, 'for Bar') fooq.enqueue(say_hello, 'for Foo') job, queue = Queue.dequeue_any([fooq, barq], None) self.assertEqual(queue, fooq) self.assertEqual(job.func, say_hello) self.assertEqual(job.origin, fooq.name) self.assertEqual( job.args[0], 'for Foo', 'Foo should be dequeued first.' ) job, queue = Queue.dequeue_any([fooq, barq], None) self.assertEqual(queue, barq) self.assertEqual(job.func, say_hello) self.assertEqual(job.origin, barq.name) self.assertEqual( job.args[0], 'for Bar', 'Bar should be dequeued second.' ) def test_dequeue_any_ignores_nonexisting_jobs(self): """Dequeuing (from any queue) silently ignores non-existing jobs.""" q = Queue('low') uuid = '49f205ab-8ea3-47dd-a1b5-bfa186870fc8' q.push_job_id(uuid) # Dequeue simply ignores the missing job and returns None self.assertEqual(q.count, 1) self.assertEqual( Queue.dequeue_any([Queue(), Queue('low')], None), # noqa None ) self.assertEqual(q.count, 0) def test_enqueue_sets_status(self): """Enqueueing a job sets its status to "queued".""" q = Queue() job = q.enqueue(say_hello) self.assertEqual(job.get_status(), JobStatus.QUEUED) def test_enqueue_meta_arg(self): """enqueue() can set the job.meta contents.""" q = Queue() job = q.enqueue(say_hello, meta={'foo': 'bar', 'baz': 42}) self.assertEqual(job.meta['foo'], 'bar') self.assertEqual(job.meta['baz'], 42) def test_enqueue_explicit_args(self): """enqueue() works for both implicit/explicit args.""" q = Queue() # Implicit args/kwargs mode job = q.enqueue(echo, 1, timeout=1, result_ttl=1, bar='baz') self.assertEqual(job.timeout, 1) self.assertEqual(job.result_ttl, 1) self.assertEqual( job.perform(), ((1,), {'bar': 'baz'}) ) # Explicit kwargs mode kwargs = { 'timeout': 1, 'result_ttl': 1, } job = q.enqueue(echo, timeout=2, result_ttl=2, args=[1], kwargs=kwargs) self.assertEqual(job.timeout, 2) self.assertEqual(job.result_ttl, 2) self.assertEqual( job.perform(), ((1,), {'timeout': 1, 'result_ttl': 1}) ) def test_all_queues(self): """All queues""" q1 = Queue('first-queue') q2 = Queue('second-queue') q3 = Queue('third-queue') # Ensure a queue is added only once a job is enqueued self.assertEqual(len(Queue.all()), 0) q1.enqueue(say_hello) self.assertEqual(len(Queue.all()), 1) # Ensure this holds true for multiple queues q2.enqueue(say_hello) q3.enqueue(say_hello) names = [q.name for q in Queue.all()] self.assertEqual(len(Queue.all()), 3) # Verify names self.assertTrue('first-queue' in names) self.assertTrue('second-queue' in names) self.assertTrue('third-queue' in names) # Now empty two queues w = Worker([q2, q3]) w.work(burst=True) # Queue.all() should still report the empty queues self.assertEqual(len(Queue.all()), 3) def test_all_custom_job(self): class CustomJob(Job): pass q = Queue('all-queue') q.enqueue(say_hello) queues = Queue.all(job_class=CustomJob) self.assertEqual(len(queues), 1) self.assertIs(queues[0].job_class, CustomJob) def test_from_queue_key(self): """Ensure being able to get a Queue instance manually from Redis""" q = Queue() key = Queue.redis_queue_namespace_prefix + 'default' reverse_q = Queue.from_queue_key(key) self.assertEqual(q, reverse_q) def test_from_queue_key_error(self): """Ensure that an exception is raised if the queue prefix is wrong""" key = 'some:weird:prefix:' + 'default' self.assertRaises(ValueError, Queue.from_queue_key, key) def test_enqueue_dependents(self): """Enqueueing dependent jobs pushes all jobs in the depends set to the queue and removes them from DeferredJobQueue.""" q = Queue() parent_job = Job.create(func=say_hello) parent_job.save() job_1 = q.enqueue(say_hello, depends_on=parent_job) job_2 = q.enqueue(say_hello, depends_on=parent_job) registry = DeferredJobRegistry(q.name, connection=self.testconn) self.assertEqual( set(registry.get_job_ids()), set([job_1.id, job_2.id]) ) # After dependents is enqueued, job_1 and job_2 should be in queue self.assertEqual(q.job_ids, []) q.enqueue_dependents(parent_job) self.assertEqual(set(q.job_ids), set([job_2.id, job_1.id])) self.assertFalse(self.testconn.exists(parent_job.dependents_key)) # DeferredJobRegistry should also be empty self.assertEqual(registry.get_job_ids(), []) def test_enqueue_dependents_on_multiple_queues(self): """Enqueueing dependent jobs on multiple queues pushes jobs in the queues and removes them from DeferredJobRegistry for each different queue.""" q_1 = Queue("queue_1") q_2 = Queue("queue_2") parent_job = Job.create(func=say_hello) parent_job.save() job_1 = q_1.enqueue(say_hello, depends_on=parent_job) job_2 = q_2.enqueue(say_hello, depends_on=parent_job) # Each queue has its own DeferredJobRegistry registry_1 = DeferredJobRegistry(q_1.name, connection=self.testconn) self.assertEqual( set(registry_1.get_job_ids()), set([job_1.id]) ) registry_2 = DeferredJobRegistry(q_2.name, connection=self.testconn) self.assertEqual( set(registry_2.get_job_ids()), set([job_2.id]) ) # After dependents is enqueued, job_1 on queue_1 and # job_2 should be in queue_2 self.assertEqual(q_1.job_ids, []) self.assertEqual(q_2.job_ids, []) q_1.enqueue_dependents(parent_job) q_2.enqueue_dependents(parent_job) self.assertEqual(set(q_1.job_ids), set([job_1.id])) self.assertEqual(set(q_2.job_ids), set([job_2.id])) self.assertFalse(self.testconn.exists(parent_job.dependents_key)) # DeferredJobRegistry should also be empty self.assertEqual(registry_1.get_job_ids(), []) self.assertEqual(registry_2.get_job_ids(), []) def test_enqueue_job_with_dependency(self): """Jobs are enqueued only when their dependencies are finished.""" # Job with unfinished dependency is not immediately enqueued parent_job = Job.create(func=say_hello) parent_job.save() q = Queue() job = q.enqueue_call(say_hello, depends_on=parent_job) self.assertEqual(q.job_ids, []) self.assertEqual(job.get_status(), JobStatus.DEFERRED) # Jobs dependent on finished jobs are immediately enqueued parent_job.set_status(JobStatus.FINISHED) parent_job.save() job = q.enqueue_call(say_hello, depends_on=parent_job) self.assertEqual(q.job_ids, [job.id]) self.assertEqual(job.timeout, Queue.DEFAULT_TIMEOUT) self.assertEqual(job.get_status(), JobStatus.QUEUED) def test_enqueue_job_with_dependency_by_id(self): """Can specify job dependency with job object or job id.""" parent_job = Job.create(func=say_hello) parent_job.save() q = Queue() q.enqueue_call(say_hello, depends_on=parent_job.id) self.assertEqual(q.job_ids, []) # Jobs dependent on finished jobs are immediately enqueued parent_job.set_status(JobStatus.FINISHED) parent_job.save() job = q.enqueue_call(say_hello, depends_on=parent_job.id) self.assertEqual(q.job_ids, [job.id]) self.assertEqual(job.timeout, Queue.DEFAULT_TIMEOUT) def test_enqueue_job_with_dependency_and_timeout(self): """Jobs remember their timeout when enqueued as a dependency.""" # Job with unfinished dependency is not immediately enqueued parent_job = Job.create(func=say_hello) parent_job.save() q = Queue() job = q.enqueue_call(say_hello, depends_on=parent_job, timeout=123) self.assertEqual(q.job_ids, []) self.assertEqual(job.timeout, 123) # Jobs dependent on finished jobs are immediately enqueued parent_job.set_status(JobStatus.FINISHED) parent_job.save() job = q.enqueue_call(say_hello, depends_on=parent_job, timeout=123) self.assertEqual(q.job_ids, [job.id]) self.assertEqual(job.timeout, 123) def test_enqueue_job_with_invalid_dependency(self): """Enqueuing a job fails, if the dependency does not exist at all.""" parent_job = Job.create(func=say_hello) # without save() the job is not visible to others q = Queue() with self.assertRaises(InvalidJobDependency): q.enqueue_call(say_hello, depends_on=parent_job) with self.assertRaises(InvalidJobDependency): q.enqueue_call(say_hello, depends_on=parent_job.id) self.assertEqual(q.job_ids, []) def test_fetch_job_successful(self): """Fetch a job from a queue.""" q = Queue('example') job_orig = q.enqueue(say_hello) job_fetch = q.fetch_job(job_orig.id) self.assertIsNotNone(job_fetch) self.assertEqual(job_orig.id, job_fetch.id) self.assertEqual(job_orig.description, job_fetch.description) def test_fetch_job_missing(self): """Fetch a job from a queue which doesn't exist.""" q = Queue('example') job = q.fetch_job('123') self.assertIsNone(job) def test_fetch_job_different_queue(self): """Fetch a job from a queue which is in a different queue.""" q1 = Queue('example1') q2 = Queue('example2') job_orig = q1.enqueue(say_hello) job_fetch = q2.fetch_job(job_orig.id) self.assertIsNone(job_fetch) job_fetch = q1.fetch_job(job_orig.id) self.assertIsNotNone(job_fetch) class TestFailedQueue(RQTestCase): def test_get_failed_queue(self): """Use custom job class""" class CustomJob(Job): pass failed_queue = get_failed_queue(job_class=CustomJob) self.assertIs(failed_queue.job_class, CustomJob) failed_queue = get_failed_queue(job_class='rq.job.Job') self.assertIsNot(failed_queue.job_class, CustomJob) def test_requeue_job(self): """Requeueing existing jobs.""" job = Job.create(func=div_by_zero, args=(1, 2, 3)) job.origin = 'fake' job.save() get_failed_queue().quarantine(job, Exception('Some fake error')) # noqa self.assertEqual(Queue.all(), [get_failed_queue()]) # noqa self.assertEqual(get_failed_queue().count, 1) requeued_job = get_failed_queue().requeue(job.id) self.assertEqual(get_failed_queue().count, 0) self.assertEqual(Queue('fake').count, 1) self.assertEqual(requeued_job.origin, job.origin) def test_get_job_on_failed_queue(self): default_queue = Queue() failed_queue = get_failed_queue() job = default_queue.enqueue(div_by_zero, args=(1, 2, 3)) job_on_default_queue = default_queue.fetch_job(job.id) job_on_failed_queue = failed_queue.fetch_job(job.id) self.assertIsNotNone(job_on_default_queue) self.assertIsNone(job_on_failed_queue) job.set_status(JobStatus.FAILED) job_on_default_queue = default_queue.fetch_job(job.id) job_on_failed_queue = failed_queue.fetch_job(job.id) self.assertIsNotNone(job_on_default_queue) self.assertIsNotNone(job_on_failed_queue) self.assertTrue(job_on_default_queue.is_failed) def test_requeue_nonfailed_job_fails(self): """Requeueing non-failed jobs raises error.""" q = Queue() job = q.enqueue(say_hello, 'Nick', foo='bar') # Assert that we cannot requeue a job that's not on the failed queue with self.assertRaises(InvalidJobOperationError): get_failed_queue().requeue(job.id) def test_quarantine_preserves_timeout(self): """Quarantine preserves job timeout.""" job = Job.create(func=div_by_zero, args=(1, 2, 3)) job.origin = 'fake' job.timeout = 200 job.save() get_failed_queue().quarantine(job, Exception('Some fake error')) self.assertEqual(job.timeout, 200) def test_requeueing_preserves_timeout(self): """Requeueing preserves job timeout.""" job = Job.create(func=div_by_zero, args=(1, 2, 3)) job.origin = 'fake' job.timeout = 200 job.save() get_failed_queue().quarantine(job, Exception('Some fake error')) get_failed_queue().requeue(job.id) job = Job.fetch(job.id) self.assertEqual(job.timeout, 200) def test_requeue_sets_status_to_queued(self): """Requeueing a job should set its status back to QUEUED.""" job = Job.create(func=div_by_zero, args=(1, 2, 3)) job.save() get_failed_queue().quarantine(job, Exception('Some fake error')) get_failed_queue().requeue(job.id) job = Job.fetch(job.id) self.assertEqual(job.get_status(), JobStatus.QUEUED) def test_enqueue_preserves_result_ttl(self): """Enqueueing persists result_ttl.""" q = Queue() job = q.enqueue(div_by_zero, args=(1, 2, 3), result_ttl=10) self.assertEqual(job.result_ttl, 10) job_from_queue = Job.fetch(job.id, connection=self.testconn) self.assertEqual(int(job_from_queue.result_ttl), 10) def test_async_false(self): """Job executes and cleaned up immediately if async=False.""" q = Queue(async=False) job = q.enqueue(some_calculation, args=(2, 3)) self.assertEqual(job.return_value, 6) self.assertNotEqual(self.testconn.ttl(job.key), -1) def test_custom_job_class(self): """Ensure custom job class assignment works as expected.""" q = Queue(job_class=CustomJob) self.assertEqual(q.job_class, CustomJob) def test_skip_queue(self): """Ensure the skip_queue option functions""" q = Queue('foo') job1 = q.enqueue(say_hello) job2 = q.enqueue(say_hello) assert q.dequeue() == job1 skip_job = q.enqueue(say_hello, at_front=True) assert q.dequeue() == skip_job assert q.dequeue() == job2 def test_job_deletion(self): """Ensure job.delete() removes itself from FailedQueue.""" job = Job.create(func=div_by_zero, args=(1, 2, 3)) job.origin = 'fake' job.timeout = 200 job.save() job.set_status(JobStatus.FAILED) failed_queue = get_failed_queue() failed_queue.quarantine(job, Exception('Some fake error')) self.assertTrue(job.id in failed_queue.get_job_ids()) job.delete() self.assertFalse(job.id in failed_queue.get_job_ids()) def test_job_in_failed_queue_persists(self): """Make sure failed job key does not expire""" q = Queue('foo') job = q.enqueue(div_by_zero, args=(1,), ttl=5) self.assertEqual(self.testconn.ttl(job.key), 5) self.assertRaises(ZeroDivisionError, job.perform) job.set_status(JobStatus.FAILED) failed_queue = get_failed_queue() failed_queue.quarantine(job, Exception('Some fake error')) self.assertEqual(self.testconn.ttl(job.key), -1)