diff --git a/CHANGES.md b/CHANGES.md index 7cae844..e305be8 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,9 +1,14 @@ -### 0.3.12 +### 0.4.0 (not released yet) +- Job dependencies! Thanks, Selwin. + - Ability to provide a custom job description (instead of using the default function invocation hint). Thanks, İbrahim. +- Temporary key for the compact queue is now randomly generated, which should + avoid name clashes for concurrent compact actions. + ### 0.3.11 (August 23th, 2013) diff --git a/rq/connections.py b/rq/connections.py index f4a72e4..ee07070 100644 --- a/rq/connections.py +++ b/rq/connections.py @@ -18,8 +18,8 @@ def Connection(connection=None): finally: popped = pop_connection() assert popped == connection, \ - 'Unexpected Redis connection was popped off the stack. ' \ - 'Check your Redis connection setup.' + 'Unexpected Redis connection was popped off the stack. ' \ + 'Check your Redis connection setup.' def push_connection(redis): @@ -37,7 +37,7 @@ def use_connection(redis=None): use of use_connection() and stacked connection contexts. """ assert len(_connection_stack) <= 1, \ - 'You should not mix Connection contexts with use_connection().' + 'You should not mix Connection contexts with use_connection().' release_local(_connection_stack) if redis is None: @@ -61,13 +61,11 @@ def resolve_connection(connection=None): connection = get_current_connection() if connection is None: - raise NoRedisConnectionException( - 'Could not resolve a Redis connection.') + raise NoRedisConnectionException('Could not resolve a Redis connection.') return connection _connection_stack = LocalStack() -__all__ = ['Connection', - 'get_current_connection', 'push_connection', 'pop_connection', - 'use_connection'] +__all__ = ['Connection', 'get_current_connection', 'push_connection', + 'pop_connection', 'use_connection'] diff --git a/rq/decorators.py b/rq/decorators.py index d57b6cc..b433904 100644 --- a/rq/decorators.py +++ b/rq/decorators.py @@ -4,10 +4,10 @@ from .connections import resolve_connection from .worker import DEFAULT_RESULT_TTL from rq.compat import string_types -class job(object): +class job(object): def __init__(self, queue, connection=None, timeout=None, - result_ttl=DEFAULT_RESULT_TTL): + result_ttl=DEFAULT_RESULT_TTL): """A decorator that adds a ``delay`` method to the decorated function, which in turn creates a RQ job when called. Accepts a required ``queue`` argument that can be either a ``Queue`` instance or a string @@ -32,6 +32,6 @@ class job(object): else: queue = self.queue return queue.enqueue_call(f, args=args, kwargs=kwargs, - timeout=self.timeout, result_ttl=self.result_ttl) + timeout=self.timeout, result_ttl=self.result_ttl) f.delay = delay return f diff --git a/rq/exceptions.py b/rq/exceptions.py index 982a580..25e4f0e 100644 --- a/rq/exceptions.py +++ b/rq/exceptions.py @@ -15,5 +15,6 @@ class UnpickleError(Exception): super(UnpickleError, self).__init__(message, inner_exception) self.raw_data = raw_data + class DequeueTimeout(Exception): pass diff --git a/rq/job.py b/rq/job.py index fdc4b57..a10c340 100644 --- a/rq/job.py +++ b/rq/job.py @@ -4,7 +4,7 @@ import times from uuid import uuid4 try: from cPickle import loads, dumps, UnpicklingError -except ImportError: # noqa +except ImportError: # noqa from pickle import loads, dumps, UnpicklingError # noqa from .local import LocalStack from .connections import resolve_connection @@ -16,8 +16,9 @@ def enum(name, *sequential, **named): values = dict(zip(sequential, range(len(sequential))), **named) return type(name, (), values) -Status = enum('Status', QUEUED='queued', FINISHED='finished', FAILED='failed', - STARTED='started') +Status = enum('Status', + QUEUED='queued', FINISHED='finished', FAILED='failed', + STARTED='started') def unpickle(pickled_string): @@ -68,7 +69,7 @@ class Job(object): # Job construction @classmethod def create(cls, func, args=None, kwargs=None, connection=None, - result_ttl=None, status=None, description=None): + result_ttl=None, status=None, description=None, dependency=None): """Creates a new Job instance for the given function, arguments, and keyword arguments. """ @@ -91,6 +92,9 @@ class Job(object): job.description = description or job.get_call_string() job.result_ttl = result_ttl job._status = status + # dependency could be job instance or id + if dependency is not None: + job._dependency_id = dependency.id if isinstance(dependency, Job) else dependency return job @property @@ -123,6 +127,20 @@ class Job(object): def is_started(self): return self.status == Status.STARTED + @property + def dependency(self): + """Returns a job's dependency. To avoid repeated Redis fetches, we cache + job.dependency as job._dependency. + """ + if self._dependency_id is None: + return None + if hasattr(self, '_dependency'): + return self._dependency + job = Job.fetch(self._dependency_id, connection=self.connection) + job.refresh() + self._dependency = job + return job + @property def func(self): func_name = self.func_name @@ -189,6 +207,7 @@ class Job(object): self.timeout = None self.result_ttl = None self._status = None + self._dependency_id = None self.meta = {} @@ -212,11 +231,21 @@ class Job(object): """The Redis key that is used to store job hash under.""" return b'rq:job:' + job_id.encode('utf-8') + @classmethod + def waitlist_key_for(cls, job_id): + """The Redis key that is used to store job hash under.""" + return 'rq:job:%s:waitlist' % (job_id,) + @property def key(self): """The Redis key that is used to store job hash under.""" return self.key_for(self.id) + @property + def waitlist_key(self): + """The Redis key that is used to store job hash under.""" + return self.waitlist_key_for(self.id) + @property # noqa def job_tuple(self): """Returns the job tuple that encodes the actual function call that @@ -289,13 +318,11 @@ class Job(object): self.timeout = int(obj.get('timeout')) if obj.get('timeout') else None self.result_ttl = int(obj.get('result_ttl')) if obj.get('result_ttl') else None # noqa self._status = as_text(obj.get('status') if obj.get('status') else None) + self._dependency_id = as_text(obj.get('dependency_id', None)) self.meta = unpickle(obj.get('meta')) if obj.get('meta') else {} - def save(self, pipeline=None): - """Persists the current job instance to its corresponding Redis key.""" - key = self.key - connection = pipeline if pipeline is not None else self.connection - + def dump(self): + """Returns a serialization of the current job instance""" obj = {} obj['created_at'] = times.format(self.created_at or times.now(), 'UTC') @@ -319,10 +346,19 @@ class Job(object): obj['result_ttl'] = self.result_ttl if self._status is not None: obj['status'] = self._status + if self._dependency_id is not None: + obj['dependency_id'] = self._dependency_id if self.meta: obj['meta'] = dumps(self.meta) - connection.hmset(key, obj) + return obj + + def save(self, pipeline=None): + """Persists the current job instance to its corresponding Redis key.""" + key = self.key + connection = pipeline if pipeline is not None else self.connection + + connection.hmset(key, self.dump()) def cancel(self): """Cancels the given job, which will prevent the job from ever being @@ -350,7 +386,6 @@ class Job(object): assert self.id == _job_stack.pop() return self._result - def get_ttl(self, default_ttl=None): """Returns ttl for a job that determines how long a job and its result will be persisted. In the future, this method will also be responsible @@ -379,13 +414,24 @@ class Job(object): - If it's a positive number, set the job to expire in X seconds. - If result_ttl is negative, don't set an expiry to it (persist forever) - """ + """ if ttl == 0: self.cancel() elif ttl > 0: connection = pipeline if pipeline is not None else self.connection connection.expire(self.key, ttl) + + def register_dependency(self): + """Jobs may have a waitlist. Jobs in this waitlist are enqueued + only if the dependency job is successfully performed. We maintain this + waitlist in Redis, with key that looks something like: + + rq:job:job_id:waitlist = ['job_id_1', 'job_id_2'] + This method puts the job on it's dependency's waitlist. + """ + # TODO: This can probably be pipelined + self.connection.rpush(Job.waitlist_key_for(self._dependency_id), self.id) def __str__(self): return '' % (self.id, self.description) @@ -398,5 +444,4 @@ class Job(object): def __hash__(self): return hash(self.id) - _job_stack = LocalStack() diff --git a/rq/local.py b/rq/local.py index 555a6d1..61f896f 100644 --- a/rq/local.py +++ b/rq/local.py @@ -13,13 +13,13 @@ # current thread ident. try: from greenlet import getcurrent as get_ident -except ImportError: # noqa +except ImportError: # noqa try: from thread import get_ident # noqa - except ImportError: # noqa + except ImportError: # noqa try: from _thread import get_ident # noqa - except ImportError: # noqa + except ImportError: # noqa from dummy_thread import get_ident # noqa @@ -119,6 +119,7 @@ class LocalStack(object): def _get__ident_func__(self): return self._local.__ident_func__ + def _set__ident_func__(self, value): # noqa object.__setattr__(self._local, '__ident_func__', value) __ident_func__ = property(_get__ident_func__, _set__ident_func__) diff --git a/rq/queue.py b/rq/queue.py index e6a8b0e..e0e686e 100644 --- a/rq/queue.py +++ b/rq/queue.py @@ -1,10 +1,15 @@ import times +import uuid + from .connections import resolve_connection from .job import Job, Status -from .exceptions import (NoSuchJobError, UnpickleError, - InvalidJobOperationError, DequeueTimeout) + +from .exceptions import (DequeueTimeout, InvalidJobOperationError, + NoSuchJobError, UnpickleError) from .compat import total_ordering, string_types, as_text +from redis import WatchError + def get_failed_queue(connection=None): """Returns a handle to the special failed queue.""" @@ -115,7 +120,7 @@ class Queue(object): """Removes all "dead" jobs from the queue by cycling through it, while guarantueeing FIFO semantics. """ - COMPACT_QUEUE = 'rq:queue:_compact' + COMPACT_QUEUE = 'rq:queue:_compact:{0}'.format(uuid.uuid4()) self.connection.rename(self.key, COMPACT_QUEUE) while True: @@ -130,8 +135,9 @@ class Queue(object): """Pushes a job ID on the corresponding Redis queue.""" self.connection.rpush(self.key, job_id) + def enqueue_call(self, func, args=None, kwargs=None, timeout=None, - result_ttl=None, description=None): + result_ttl=None, description=None, after=None): """Creates a job to represent the delayed function call and enqueues it. @@ -140,8 +146,29 @@ class Queue(object): contain options for RQ itself. """ timeout = timeout or self._default_timeout - job = Job.create(func, args, kwargs, description=description, connection=self.connection, - result_ttl=result_ttl, status=Status.QUEUED) + + # TODO: job with dependency shouldn't have "queued" as status + job = Job.create(func, args, kwargs, connection=self.connection, + result_ttl=result_ttl, status=Status.QUEUED, + description=description, dependency=after) + + # If job depends on an unfinished job, register itself on it's + # parent's waitlist instead of enqueueing it. + # If WatchError is raised in the process, that means something else is + # modifying the dependency. In this case we simply retry + if after is not None: + with self.connection.pipeline() as pipe: + while True: + try: + pipe.watch(after.key) + if after.status != Status.FINISHED: + job.register_dependency() + job.save() + return job + break + except WatchError: + continue + return self.enqueue_job(job, timeout=timeout) def enqueue(self, f, *args, **kwargs): @@ -167,16 +194,19 @@ class Queue(object): timeout = None description = None result_ttl = None - if 'args' in kwargs or 'kwargs' in kwargs: + after = None + if 'args' in kwargs or 'kwargs' in kwargs or 'after' in kwargs: assert args == (), 'Extra positional arguments cannot be used when using explicit args and kwargs.' # noqa timeout = kwargs.pop('timeout', None) description = kwargs.pop('description', None) args = kwargs.pop('args', None) result_ttl = kwargs.pop('result_ttl', None) + after = kwargs.pop('after', None) kwargs = kwargs.pop('kwargs', None) - return self.enqueue_call(func=f, args=args, kwargs=kwargs, description=description, - timeout=timeout, result_ttl=result_ttl) + return self.enqueue_call(func=f, args=args, kwargs=kwargs, + timeout=timeout, result_ttl=result_ttl, + description=description, after=after) def enqueue_job(self, job, timeout=None, set_meta_data=True): """Enqueues a job for delayed execution. @@ -210,6 +240,16 @@ class Queue(object): job.save() return job + def enqueue_waitlist(self, job): + """Enqueues all jobs in the waitlist and clears it""" + # TODO: can probably be pipelined + while True: + job_id = as_text(self.connection.lpop(job.waitlist_key)) + if job_id is None: + break + waitlisted_job = Job.fetch(job_id, connection=self.connection) + self.enqueue_job(waitlisted_job) + def pop_job_id(self): """Pops a given job ID from this Redis queue.""" return as_text(self.connection.lpop(self.key)) diff --git a/rq/scripts/__init__.py b/rq/scripts/__init__.py index fc31d09..575a8cb 100644 --- a/rq/scripts/__init__.py +++ b/rq/scripts/__init__.py @@ -1,35 +1,45 @@ import importlib import redis +from warnings import warn from rq import use_connection def add_standard_arguments(parser): parser.add_argument('--config', '-c', default=None, - help='Module containing RQ settings.') + help='Module containing RQ settings.') parser.add_argument('--url', '-u', default=None, - help='URL describing Redis connection details. ' - 'Overrides other connection arguments if supplied.') + help='URL describing Redis connection details. ' + 'Overrides other connection arguments if supplied.') parser.add_argument('--host', '-H', default=None, - help='The Redis hostname (default: localhost)') + help='The Redis hostname (default: localhost)') parser.add_argument('--port', '-p', default=None, - help='The Redis portnumber (default: 6379)') + help='The Redis portnumber (default: 6379)') parser.add_argument('--db', '-d', type=int, default=None, - help='The Redis database (default: 0)') + help='The Redis database (default: 0)') parser.add_argument('--password', '-a', default=None, - help='The Redis password (default: None)') + help='The Redis password (default: None)') parser.add_argument('--socket', '-s', default=None, - help='The Redis Unix socket') + help='The Redis Unix socket') + def read_config_file(module): """Reads all UPPERCASE variables defined in the given module file.""" settings = importlib.import_module(module) return dict([(k, v) - for k, v in settings.__dict__.items() - if k.upper() == k]) + for k, v in settings.__dict__.items() + if k.upper() == k]) def setup_default_arguments(args, settings): """ Sets up args from settings or defaults """ + if args.url is None: + args.url = settings.get('REDIS_URL') + + if (args.host or args.port or args.socket or args.db or args.password): + warn('Host, port, db, password options for Redis will not be ' + 'supported in future versions of RQ. ' + 'Please use `REDIS_URL` or `--url` instead.', DeprecationWarning) + if args.host is None: args.host = settings.get('REDIS_HOST', 'localhost') @@ -54,5 +64,5 @@ def setup_redis(args): redis_conn = redis.StrictRedis.from_url(args.url) else: redis_conn = redis.StrictRedis(host=args.host, port=args.port, db=args.db, - password=args.password, unix_socket_path=args.socket) + password=args.password, unix_socket_path=args.socket) use_connection(redis_conn) diff --git a/rq/timeouts.py b/rq/timeouts.py index d26528a..f1e1848 100644 --- a/rq/timeouts.py +++ b/rq/timeouts.py @@ -33,7 +33,7 @@ class death_penalty_after(object): def handle_death_penalty(self, signum, frame): raise JobTimeoutException('Job exceeded maximum timeout ' - 'value (%d seconds).' % self._timeout) + 'value (%d seconds).' % self._timeout) def setup_death_penalty(self): """Sets up an alarm signal and a signal handler that raises diff --git a/rq/utils.py b/rq/utils.py index 44bbe65..8219c9f 100644 --- a/rq/utils.py +++ b/rq/utils.py @@ -16,8 +16,7 @@ def gettermsize(): def ioctl_GWINSZ(fd): try: import fcntl, termios, struct # noqa - cr = struct.unpack('hh', fcntl.ioctl(fd, termios.TIOCGWINSZ, - '1234')) + cr = struct.unpack('hh', fcntl.ioctl(fd, termios.TIOCGWINSZ, '1234')) except: return None return cr @@ -53,7 +52,7 @@ class _Colorizer(object): self.codes["overline"] = esc + "06m" dark_colors = ["black", "darkred", "darkgreen", "brown", "darkblue", - "purple", "teal", "lightgray"] + "purple", "teal", "lightgray"] light_colors = ["darkgray", "red", "green", "yellow", "blue", "fuchsia", "turquoise", "white"] @@ -139,7 +138,7 @@ class ColorizingStreamHandler(logging.StreamHandler): def __init__(self, exclude=None, *args, **kwargs): self.exclude = exclude - if is_python_version((2,6)): + if is_python_version((2, 6)): logging.StreamHandler.__init__(self, *args, **kwargs) else: super(ColorizingStreamHandler, self).__init__(*args, **kwargs) diff --git a/rq/worker.py b/rq/worker.py index 3ba4250..d350373 100644 --- a/rq/worker.py +++ b/rq/worker.py @@ -43,9 +43,9 @@ def iterable(x): def compact(l): return [x for x in l if x is not None] -_signames = dict((getattr(signal, signame), signame) \ - for signame in dir(signal) \ - if signame.startswith('SIG') and '_' not in signame) +_signames = dict((getattr(signal, signame), signame) + for signame in dir(signal) + if signame.startswith('SIG') and '_' not in signame) def signal_name(signum): @@ -68,8 +68,8 @@ class Worker(object): if connection is None: connection = get_current_connection() reported_working = connection.smembers(cls.redis_workers_keys) - workers = [cls.find_by_key(as_text(key), connection) for key in - reported_working] + workers = [cls.find_by_key(as_text(key), connection) + for key in reported_working] return compact(workers) @classmethod @@ -95,13 +95,12 @@ class Worker(object): worker._state = connection.hget(worker.key, 'state') or '?' if queues: worker.queues = [Queue(queue, connection=connection) - for queue in queues.split(',')] + for queue in queues.split(',')] return worker - def __init__(self, queues, name=None, - default_result_ttl=DEFAULT_RESULT_TTL, connection=None, - exc_handler=None, default_worker_ttl=DEFAULT_WORKER_TTL): # noqa + default_result_ttl=DEFAULT_RESULT_TTL, connection=None, + exc_handler=None, default_worker_ttl=DEFAULT_WORKER_TTL): # noqa if connection is None: connection = get_current_connection() self.connection = connection @@ -193,9 +192,8 @@ class Worker(object): self.log.debug('Registering birth of worker %s' % (self.name,)) if self.connection.exists(self.key) and \ not self.connection.hexists(self.key, 'death'): - raise ValueError( - 'There exists an active worker named \'%s\' ' - 'already.' % (self.name,)) + raise ValueError('There exists an active worker named \'%s\' ' + 'already.' % (self.name,)) key = self.key now = time.time() queues = ','.join(self.queue_names()) @@ -304,8 +302,8 @@ class Worker(object): qnames = self.queue_names() self.procline('Listening on %s' % ','.join(qnames)) self.log.info('') - self.log.info('*** Listening on %s...' % \ - green(', '.join(qnames))) + self.log.info('*** Listening on %s...' % + green(', '.join(qnames))) timeout = None if burst else max(1, self.default_worker_ttl - 60) try: result = self.dequeue_job_and_maintain_ttl(timeout) @@ -324,11 +322,13 @@ class Worker(object): # Use the public setter here, to immediately update Redis job.status = Status.STARTED self.log.info('%s: %s (%s)' % (green(queue.name), - blue(job.description), job.id)) + blue(job.description), job.id)) self.connection.expire(self.key, (job.timeout or 180) + 60) self.fork_and_perform_job(job) self.connection.expire(self.key, self.default_worker_ttl) + if job.status == 'finished': + queue.enqueue_waitlist(job) did_perform_work = True finally: @@ -336,19 +336,17 @@ class Worker(object): self.register_death() return did_perform_work - def dequeue_job_and_maintain_ttl(self, timeout): while True: try: return Queue.dequeue_any(self.queues, timeout, - connection=self.connection) + connection=self.connection) except DequeueTimeout: pass self.log.debug('Sending heartbeat to prevent worker timeout.') self.connection.expire(self.key, self.default_worker_ttl) - def fork_and_perform_job(self, job): """Spawns a work horse to perform the actual work and passes it a job. The worker will wait for the work horse and make sure it executes @@ -443,12 +441,10 @@ class Worker(object): return True - def handle_exception(self, job, *exc_info): """Walks the exception handler stack to delegate exception handling.""" - exc_string = ''.join( - traceback.format_exception_only(*exc_info[:2]) + - traceback.format_exception(*exc_info)) + exc_string = ''.join(traceback.format_exception_only(*exc_info[:2]) + + traceback.format_exception(*exc_info)) self.log.error(exc_string) for handler in reversed(self._exc_handlers): diff --git a/tests/test_job.py b/tests/test_job.py index d25f60f..bc70d5e 100644 --- a/tests/test_job.py +++ b/tests/test_job.py @@ -7,6 +7,7 @@ try: from cPickle import loads except ImportError: from pickle import loads +from rq.compat import as_text from rq.job import Job, get_current_job from rq.exceptions import NoSuchJobError, UnpickleError from rq.queue import Queue @@ -134,6 +135,22 @@ class TestJob(RQTestCase): sorted(self.testconn.hkeys(job.key)), [b'created_at', b'data', b'description']) + 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.save() + job = Job.create(func=some_calculation, dependency=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, dependency=parent_job.id) + job.save() + stored_job = Job.fetch(job.id) + self.assertEqual(stored_job._dependency_id, parent_job.id) + self.assertEqual(stored_job.dependency, parent_job) + def test_store_then_fetch(self): """Store, then fetch.""" job = Job.create(func=some_calculation, args=(3, 4), kwargs=dict(z=2)) @@ -265,3 +282,11 @@ class TestJob(RQTestCase): # Jobs with 0 TTL are immediately deleted job.cleanup(ttl=0) self.assertRaises(NoSuchJobError, Job.fetch, job.id, self.testconn) + + def test_register_dependency(self): + """Test that jobs updates the correct job waitlist""" + job = Job.create(func=say_hello) + job._dependency_id = 'id' + job.save() + job.register_dependency() + self.assertEqual(as_text(self.testconn.lpop('rq:job:id:waitlist')), job.id) diff --git a/tests/test_queue.py b/tests/test_queue.py index bd8975b..6ab14cc 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -277,6 +277,39 @@ class TestQueue(RQTestCase): self.testconn.srem(Queue.redis_queues_keys, s.key) self.assertEquals(len(Queue.all()), 2) + def test_enqueue_waitlist(self): + """Enqueueing a waitlist pushes all jobs in waitlist to queue""" + q = Queue() + parent_job = Job.create(func=say_hello) + parent_job.save() + job_1 = Job.create(func=say_hello, dependency=parent_job) + job_1.save() + job_1.register_dependency() + job_2 = Job.create(func=say_hello, dependency=parent_job) + job_2.save() + job_2.register_dependency() + + # After waitlist is enqueued, job_1 and job_2 should be in queue + self.assertEqual(q.job_ids, []) + q.enqueue_waitlist(parent_job) + self.assertEqual(q.job_ids, [job_1.id, job_2.id]) + self.assertFalse(self.testconn.exists(parent_job.waitlist_key)) + + 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) + q = Queue() + q.enqueue_call(say_hello, after=parent_job) + self.assertEqual(q.job_ids, []) + + # Jobs dependent on finished jobs are immediately enqueued + parent_job.status = 'finished' + parent_job.save() + job = q.enqueue_call(say_hello, after=parent_job) + self.assertEqual(q.job_ids, [job.id]) + + class TestFailedQueue(RQTestCase): def test_requeue_job(self): """Requeueing existing jobs.""" diff --git a/tests/test_worker.py b/tests/test_worker.py index d1b2632..db35975 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -234,3 +234,19 @@ class TestWorker(RQTestCase): self.assertEqual(job.is_queued, False) self.assertEqual(job.is_finished, False) self.assertEqual(job.is_failed, True) + + def test_job_dependency(self): + """Enqueue waitlisted jobs only if their parents don't fail""" + q = Queue() + w = Worker([q]) + parent_job = q.enqueue(say_hello) + job = q.enqueue_call(say_hello, after=parent_job) + w.work(burst=True) + job = Job.fetch(job.id) + self.assertEqual(job.status, 'finished') + + parent_job = q.enqueue(div_by_zero) + job = q.enqueue_call(say_hello, after=parent_job) + w.work(burst=True) + job = Job.fetch(job.id) + self.assertNotEqual(job.status, 'finished')