Make test more readable by using constants.

main
Vincent Driessen 11 years ago
parent b89d6c8990
commit 01f9a98ad9

@ -307,24 +307,36 @@ class TestQueue(RQTestCase):
self.assertEqual(q.job_ids, [job_1.id, job_2.id]) self.assertEqual(q.job_ids, [job_1.id, job_2.id])
self.assertFalse(self.testconn.exists(parent_job.waitlist_key)) self.assertFalse(self.testconn.exists(parent_job.waitlist_key))
def test_enqueue_job_with_dependency(self, timeout=None): def test_enqueue_job_with_dependency(self):
"""Jobs are enqueued only when their dependencies are finished""" """Jobs are enqueued only when their dependencies are finished."""
# Job with unfinished dependency is not immediately enqueued # Job with unfinished dependency is not immediately enqueued
parent_job = Job.create(func=say_hello) parent_job = Job.create(func=say_hello)
q = Queue() q = Queue()
job = q.enqueue_call(say_hello, after=parent_job, timeout=timeout) q.enqueue_call(say_hello, after=parent_job)
self.assertEqual(q.job_ids, []) self.assertEqual(q.job_ids, [])
self.assertEqual(job.timeout, timeout)
# Jobs dependent on finished jobs are immediately enqueued # Jobs dependent on finished jobs are immediately enqueued
parent_job.status = 'finished' parent_job.status = 'finished'
parent_job.save() parent_job.save()
job = q.enqueue_call(say_hello, after=parent_job, timeout=timeout) job = q.enqueue_call(say_hello, after=parent_job)
self.assertEqual(q.job_ids, [job.id]) self.assertEqual(q.job_ids, [job.id])
self.assertEqual(job.timeout, Queue.DEFAULT_TIMEOUT if timeout is None else timeout) self.assertEqual(job.timeout, Queue.DEFAULT_TIMEOUT)
def test_enqueue_job_with_dependency_and_timeout(self): def test_enqueue_job_with_dependency_and_timeout(self):
self.test_enqueue_job_with_dependency(123) """Jobs still know their specified timeout after being scheduled as a dependency."""
# Job with unfinished dependency is not immediately enqueued
parent_job = Job.create(func=say_hello)
q = Queue()
job = q.enqueue_call(say_hello, after=parent_job, timeout=123)
self.assertEqual(q.job_ids, [])
self.assertEqual(job.timeout, 123)
# Jobs dependent on finished jobs are immediately enqueued
parent_job.status = 'finished'
parent_job.save()
job = q.enqueue_call(say_hello, after=parent_job, timeout=123)
self.assertEqual(q.job_ids, [job.id])
self.assertEqual(job.timeout, 123)
class TestFailedQueue(RQTestCase): class TestFailedQueue(RQTestCase):

Loading…
Cancel
Save