|
|
|
@ -5,8 +5,6 @@ from __future__ import (absolute_import, division, print_function,
|
|
|
|
|
from datetime import datetime
|
|
|
|
|
|
|
|
|
|
from tests import RQTestCase
|
|
|
|
|
from tests.fixtures import (CallableObject, Number, access_self,
|
|
|
|
|
long_running_job, say_hello, some_calculation)
|
|
|
|
|
from tests.helpers import strip_microseconds
|
|
|
|
|
|
|
|
|
|
from rq.compat import PY2, as_text
|
|
|
|
@ -17,6 +15,8 @@ from rq.registry import DeferredJobRegistry
|
|
|
|
|
from rq.utils import utcformat
|
|
|
|
|
from rq.worker import Worker
|
|
|
|
|
|
|
|
|
|
from . import fixtures
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
from cPickle import loads, dumps
|
|
|
|
|
except ImportError:
|
|
|
|
@ -70,7 +70,7 @@ class TestJob(RQTestCase):
|
|
|
|
|
|
|
|
|
|
def test_create_typical_job(self):
|
|
|
|
|
"""Creation of jobs for function calls."""
|
|
|
|
|
job = Job.create(func=some_calculation, args=(3, 4), kwargs=dict(z=2))
|
|
|
|
|
job = Job.create(func=fixtures.some_calculation, args=(3, 4), kwargs=dict(z=2))
|
|
|
|
|
|
|
|
|
|
# Jobs have a random UUID
|
|
|
|
|
self.assertIsNotNone(job.id)
|
|
|
|
@ -79,7 +79,7 @@ class TestJob(RQTestCase):
|
|
|
|
|
self.assertIsNone(job.instance)
|
|
|
|
|
|
|
|
|
|
# Job data is set...
|
|
|
|
|
self.assertEquals(job.func, some_calculation)
|
|
|
|
|
self.assertEquals(job.func, fixtures.some_calculation)
|
|
|
|
|
self.assertEquals(job.args, (3, 4))
|
|
|
|
|
self.assertEquals(job.kwargs, {'z': 2})
|
|
|
|
|
|
|
|
|
@ -90,7 +90,7 @@ class TestJob(RQTestCase):
|
|
|
|
|
|
|
|
|
|
def test_create_instance_method_job(self):
|
|
|
|
|
"""Creation of jobs for instance methods."""
|
|
|
|
|
n = Number(2)
|
|
|
|
|
n = fixtures.Number(2)
|
|
|
|
|
job = Job.create(func=n.div, args=(4,))
|
|
|
|
|
|
|
|
|
|
# Job data is set
|
|
|
|
@ -103,13 +103,13 @@ class TestJob(RQTestCase):
|
|
|
|
|
job = Job.create(func='tests.fixtures.say_hello', args=('World',))
|
|
|
|
|
|
|
|
|
|
# Job data is set
|
|
|
|
|
self.assertEquals(job.func, say_hello)
|
|
|
|
|
self.assertEquals(job.func, fixtures.say_hello)
|
|
|
|
|
self.assertIsNone(job.instance)
|
|
|
|
|
self.assertEquals(job.args, ('World',))
|
|
|
|
|
|
|
|
|
|
def test_create_job_from_callable_class(self):
|
|
|
|
|
"""Creation of jobs using a callable class specifier."""
|
|
|
|
|
kallable = CallableObject()
|
|
|
|
|
kallable = fixtures.CallableObject()
|
|
|
|
|
job = Job.create(func=kallable)
|
|
|
|
|
|
|
|
|
|
self.assertEquals(job.func, kallable.__call__)
|
|
|
|
@ -138,7 +138,7 @@ class TestJob(RQTestCase):
|
|
|
|
|
|
|
|
|
|
def test_save(self): # noqa
|
|
|
|
|
"""Storing jobs."""
|
|
|
|
|
job = Job.create(func=some_calculation, args=(3, 4), kwargs=dict(z=2))
|
|
|
|
|
job = Job.create(func=fixtures.some_calculation, args=(3, 4), kwargs=dict(z=2))
|
|
|
|
|
|
|
|
|
|
# Saving creates a Redis hash
|
|
|
|
|
self.assertEquals(self.testconn.exists(job.key), False)
|
|
|
|
@ -174,7 +174,7 @@ class TestJob(RQTestCase):
|
|
|
|
|
|
|
|
|
|
def test_persistence_of_typical_jobs(self):
|
|
|
|
|
"""Storing typical jobs."""
|
|
|
|
|
job = Job.create(func=some_calculation, args=(3, 4), kwargs=dict(z=2))
|
|
|
|
|
job = Job.create(func=fixtures.some_calculation, args=(3, 4), kwargs=dict(z=2))
|
|
|
|
|
job.save()
|
|
|
|
|
|
|
|
|
|
expected_date = strip_microseconds(job.created_at)
|
|
|
|
@ -190,15 +190,15 @@ class TestJob(RQTestCase):
|
|
|
|
|
|
|
|
|
|
def test_persistence_of_parent_job(self):
|
|
|
|
|
"""Storing jobs with parent job, either instance or key."""
|
|
|
|
|
parent_job = Job.create(func=some_calculation)
|
|
|
|
|
parent_job = Job.create(func=fixtures.some_calculation)
|
|
|
|
|
parent_job.save()
|
|
|
|
|
job = Job.create(func=some_calculation, depends_on=parent_job)
|
|
|
|
|
job = Job.create(func=fixtures.some_calculation, depends_on=parent_job)
|
|
|
|
|
job.save()
|
|
|
|
|
stored_job = Job.fetch(job.id)
|
|
|
|
|
self.assertEqual(stored_job._dependency_id, parent_job.id)
|
|
|
|
|
self.assertEqual(stored_job.dependency, parent_job)
|
|
|
|
|
|
|
|
|
|
job = Job.create(func=some_calculation, depends_on=parent_job.id)
|
|
|
|
|
job = Job.create(func=fixtures.some_calculation, depends_on=parent_job.id)
|
|
|
|
|
job.save()
|
|
|
|
|
stored_job = Job.fetch(job.id)
|
|
|
|
|
self.assertEqual(stored_job._dependency_id, parent_job.id)
|
|
|
|
@ -206,7 +206,7 @@ class TestJob(RQTestCase):
|
|
|
|
|
|
|
|
|
|
def test_store_then_fetch(self):
|
|
|
|
|
"""Store, then fetch."""
|
|
|
|
|
job = Job.create(func=some_calculation, args=(3, 4), kwargs=dict(z=2))
|
|
|
|
|
job = Job.create(func=fixtures.some_calculation, args=(3, 4), kwargs=dict(z=2))
|
|
|
|
|
job.save()
|
|
|
|
|
|
|
|
|
|
job2 = Job.fetch(job.id)
|
|
|
|
@ -225,7 +225,7 @@ class TestJob(RQTestCase):
|
|
|
|
|
def test_fetching_unreadable_data(self):
|
|
|
|
|
"""Fetching succeeds on unreadable data, but lazy props fail."""
|
|
|
|
|
# Set up
|
|
|
|
|
job = Job.create(func=some_calculation, args=(3, 4), kwargs=dict(z=2))
|
|
|
|
|
job = Job.create(func=fixtures.some_calculation, args=(3, 4), kwargs=dict(z=2))
|
|
|
|
|
job.save()
|
|
|
|
|
|
|
|
|
|
# Just replace the data hkey with some random noise
|
|
|
|
@ -238,7 +238,7 @@ class TestJob(RQTestCase):
|
|
|
|
|
|
|
|
|
|
def test_job_is_unimportable(self):
|
|
|
|
|
"""Jobs that cannot be imported throw exception on access."""
|
|
|
|
|
job = Job.create(func=say_hello, args=('Lionel',))
|
|
|
|
|
job = Job.create(func=fixtures.say_hello, args=('Lionel',))
|
|
|
|
|
job.save()
|
|
|
|
|
|
|
|
|
|
# Now slightly modify the job to make it unimportable (this is
|
|
|
|
@ -254,7 +254,7 @@ class TestJob(RQTestCase):
|
|
|
|
|
|
|
|
|
|
def test_custom_meta_is_persisted(self):
|
|
|
|
|
"""Additional meta data on jobs are stored persisted correctly."""
|
|
|
|
|
job = Job.create(func=say_hello, args=('Lionel',))
|
|
|
|
|
job = Job.create(func=fixtures.say_hello, args=('Lionel',))
|
|
|
|
|
job.meta['foo'] = 'bar'
|
|
|
|
|
job.save()
|
|
|
|
|
|
|
|
|
@ -266,25 +266,25 @@ class TestJob(RQTestCase):
|
|
|
|
|
|
|
|
|
|
def test_result_ttl_is_persisted(self):
|
|
|
|
|
"""Ensure that job's result_ttl is set properly"""
|
|
|
|
|
job = Job.create(func=say_hello, args=('Lionel',), result_ttl=10)
|
|
|
|
|
job = Job.create(func=fixtures.say_hello, args=('Lionel',), result_ttl=10)
|
|
|
|
|
job.save()
|
|
|
|
|
Job.fetch(job.id, connection=self.testconn)
|
|
|
|
|
self.assertEqual(job.result_ttl, 10)
|
|
|
|
|
|
|
|
|
|
job = Job.create(func=say_hello, args=('Lionel',))
|
|
|
|
|
job = Job.create(func=fixtures.say_hello, args=('Lionel',))
|
|
|
|
|
job.save()
|
|
|
|
|
Job.fetch(job.id, connection=self.testconn)
|
|
|
|
|
self.assertEqual(job.result_ttl, None)
|
|
|
|
|
|
|
|
|
|
def test_description_is_persisted(self):
|
|
|
|
|
"""Ensure that job's custom description is set properly"""
|
|
|
|
|
job = Job.create(func=say_hello, args=('Lionel',), description='Say hello!')
|
|
|
|
|
job = Job.create(func=fixtures.say_hello, args=('Lionel',), description='Say hello!')
|
|
|
|
|
job.save()
|
|
|
|
|
Job.fetch(job.id, connection=self.testconn)
|
|
|
|
|
self.assertEqual(job.description, 'Say hello!')
|
|
|
|
|
|
|
|
|
|
# Ensure job description is constructed from function call string
|
|
|
|
|
job = Job.create(func=say_hello, args=('Lionel',))
|
|
|
|
|
job = Job.create(func=fixtures.say_hello, args=('Lionel',))
|
|
|
|
|
job.save()
|
|
|
|
|
Job.fetch(job.id, connection=self.testconn)
|
|
|
|
|
if PY2:
|
|
|
|
@ -299,23 +299,23 @@ class TestJob(RQTestCase):
|
|
|
|
|
def test_job_access_within_job_function(self):
|
|
|
|
|
"""The current job is accessible within the job function."""
|
|
|
|
|
q = Queue()
|
|
|
|
|
q.enqueue(access_self) # access_self calls get_current_job() and asserts
|
|
|
|
|
q.enqueue(fixtures.access_self) # access_self calls get_current_job() and asserts
|
|
|
|
|
w = Worker([q])
|
|
|
|
|
w.work(burst=True)
|
|
|
|
|
|
|
|
|
|
def test_job_access_within_synchronous_job_function(self):
|
|
|
|
|
queue = Queue(async=False)
|
|
|
|
|
queue.enqueue(access_self)
|
|
|
|
|
queue.enqueue(fixtures.access_self)
|
|
|
|
|
|
|
|
|
|
def test_get_result_ttl(self):
|
|
|
|
|
"""Getting job result TTL."""
|
|
|
|
|
job_result_ttl = 1
|
|
|
|
|
default_ttl = 2
|
|
|
|
|
job = Job.create(func=say_hello, result_ttl=job_result_ttl)
|
|
|
|
|
job = Job.create(func=fixtures.say_hello, result_ttl=job_result_ttl)
|
|
|
|
|
job.save()
|
|
|
|
|
self.assertEqual(job.get_result_ttl(default_ttl=default_ttl), job_result_ttl)
|
|
|
|
|
self.assertEqual(job.get_result_ttl(), job_result_ttl)
|
|
|
|
|
job = Job.create(func=say_hello)
|
|
|
|
|
job = Job.create(func=fixtures.say_hello)
|
|
|
|
|
job.save()
|
|
|
|
|
self.assertEqual(job.get_result_ttl(default_ttl=default_ttl), default_ttl)
|
|
|
|
|
self.assertEqual(job.get_result_ttl(), None)
|
|
|
|
@ -323,24 +323,24 @@ class TestJob(RQTestCase):
|
|
|
|
|
def test_get_job_ttl(self):
|
|
|
|
|
"""Getting job TTL."""
|
|
|
|
|
ttl = 1
|
|
|
|
|
job = Job.create(func=say_hello, ttl=ttl)
|
|
|
|
|
job = Job.create(func=fixtures.say_hello, ttl=ttl)
|
|
|
|
|
job.save()
|
|
|
|
|
self.assertEqual(job.get_ttl(), ttl)
|
|
|
|
|
job = Job.create(func=say_hello)
|
|
|
|
|
job = Job.create(func=fixtures.say_hello)
|
|
|
|
|
job.save()
|
|
|
|
|
self.assertEqual(job.get_ttl(), None)
|
|
|
|
|
|
|
|
|
|
def test_ttl_via_enqueue(self):
|
|
|
|
|
ttl = 1
|
|
|
|
|
queue = Queue(connection=self.testconn)
|
|
|
|
|
job = queue.enqueue(say_hello, ttl=ttl)
|
|
|
|
|
job = queue.enqueue(fixtures.say_hello, ttl=ttl)
|
|
|
|
|
self.assertEqual(job.get_ttl(), ttl)
|
|
|
|
|
|
|
|
|
|
def test_never_expire_during_execution(self):
|
|
|
|
|
"""Test what happens when job expires during execution"""
|
|
|
|
|
ttl = 1
|
|
|
|
|
queue = Queue(connection=self.testconn)
|
|
|
|
|
job = queue.enqueue(long_running_job, args=(2,), ttl=ttl)
|
|
|
|
|
job = queue.enqueue(fixtures.long_running_job, args=(2,), ttl=ttl)
|
|
|
|
|
self.assertEqual(job.get_ttl(), ttl)
|
|
|
|
|
job.save()
|
|
|
|
|
job.perform()
|
|
|
|
@ -350,7 +350,7 @@ class TestJob(RQTestCase):
|
|
|
|
|
|
|
|
|
|
def test_cleanup(self):
|
|
|
|
|
"""Test that jobs and results are expired properly."""
|
|
|
|
|
job = Job.create(func=say_hello)
|
|
|
|
|
job = Job.create(func=fixtures.say_hello)
|
|
|
|
|
job.save()
|
|
|
|
|
|
|
|
|
|
# Jobs with negative TTLs don't expire
|
|
|
|
@ -370,7 +370,7 @@ class TestJob(RQTestCase):
|
|
|
|
|
origin = 'some_queue'
|
|
|
|
|
registry = DeferredJobRegistry(origin, self.testconn)
|
|
|
|
|
|
|
|
|
|
job = Job.create(func=say_hello, origin=origin)
|
|
|
|
|
job = Job.create(func=fixtures.say_hello, origin=origin)
|
|
|
|
|
job._dependency_id = 'id'
|
|
|
|
|
job.save()
|
|
|
|
|
|
|
|
|
@ -382,8 +382,8 @@ class TestJob(RQTestCase):
|
|
|
|
|
def test_cancel(self):
|
|
|
|
|
"""job.cancel() deletes itself & dependents mapping from Redis."""
|
|
|
|
|
queue = Queue(connection=self.testconn)
|
|
|
|
|
job = queue.enqueue(say_hello)
|
|
|
|
|
job2 = Job.create(func=say_hello, depends_on=job)
|
|
|
|
|
job = queue.enqueue(fixtures.say_hello)
|
|
|
|
|
job2 = Job.create(func=fixtures.say_hello, depends_on=job)
|
|
|
|
|
job2.register_dependency()
|
|
|
|
|
job.cancel()
|
|
|
|
|
self.assertFalse(self.testconn.exists(job.key))
|
|
|
|
@ -394,8 +394,8 @@ class TestJob(RQTestCase):
|
|
|
|
|
def test_create_job_with_id(self):
|
|
|
|
|
"""test creating jobs with a custom ID"""
|
|
|
|
|
queue = Queue(connection=self.testconn)
|
|
|
|
|
job = queue.enqueue(say_hello, job_id="1234")
|
|
|
|
|
job = queue.enqueue(fixtures.say_hello, job_id="1234")
|
|
|
|
|
self.assertEqual(job.id, "1234")
|
|
|
|
|
job.perform()
|
|
|
|
|
|
|
|
|
|
self.assertRaises(TypeError, queue.enqueue, say_hello, job_id=1234)
|
|
|
|
|
self.assertRaises(TypeError, queue.enqueue, fixtures.say_hello, job_id=1234)
|
|
|
|
|