mirror of https://github.com/peter4431/rq.git
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
69 lines
1.4 KiB
Python
69 lines
1.4 KiB
Python
"""
|
|
This file contains all jobs that are used in tests. Each of these test
|
|
fixtures has a slighty different characteristics.
|
|
"""
|
|
import time
|
|
from rq import Connection
|
|
from rq.decorators import job
|
|
from rq import get_current_job
|
|
|
|
|
|
def say_hello(name=None):
|
|
"""A job with a single argument and a return value."""
|
|
if name is None:
|
|
name = 'Stranger'
|
|
return 'Hi there, %s!' % (name,)
|
|
|
|
|
|
def do_nothing():
|
|
"""The best job in the world."""
|
|
pass
|
|
|
|
|
|
def div_by_zero(x):
|
|
"""Prepare for a division-by-zero exception."""
|
|
return x / 0
|
|
|
|
|
|
def some_calculation(x, y, z=1):
|
|
"""Some arbitrary calculation with three numbers. Choose z smartly if you
|
|
want a division by zero exception.
|
|
"""
|
|
return x * y / z
|
|
|
|
|
|
def create_file(path):
|
|
"""Creates a file at the given path. Actually, leaves evidence that the
|
|
job ran."""
|
|
with open(path, 'w') as f:
|
|
f.write('Just a sentinel.')
|
|
|
|
|
|
def create_file_after_timeout(path, timeout):
|
|
time.sleep(timeout)
|
|
create_file(path)
|
|
|
|
|
|
def access_self():
|
|
job = get_current_job()
|
|
return job.id
|
|
|
|
|
|
class Calculator(object):
|
|
"""Test instance methods."""
|
|
def __init__(self, denominator):
|
|
self.denominator = denominator
|
|
|
|
def calculate(self, x, y):
|
|
return x * y / self.denominator
|
|
|
|
|
|
with Connection():
|
|
@job(queue='default')
|
|
def decorated_job(x, y):
|
|
return x + y
|
|
|
|
|
|
def long_running_job():
|
|
time.sleep(10)
|