1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37
|
from rq import get_current_job
from rq.job import Job
from rq.worker import Worker
from django_rq.queues import DjangoRQ
class DummyJob(Job):
pass
class DummyQueue(DjangoRQ):
"""Just Fake class for the following test"""
class DummyWorker(Worker):
pass
try:
from rq_scheduler import Scheduler
class DummyScheduler(Scheduler):
pass
except ImportError:
pass
def access_self():
job = get_current_job()
assert job
return job.id
def failing_job():
raise ValueError
|