mirror of
https://github.com/django-q2/django-q2.git
synced 2026-09-26 03:58:11 +08:00
Ilan Steemers
4e38925cea
Workers now share a ctype with the sentinel to indicate if they are currently executing a task. This timer is incremented each guard loop until the worker resets it after finishing a job or until it reaches the TIMEOUT value and the worker is terminated by the sentinel.
61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
from multiprocessing import Queue, Event, Value
|
|
|
|
import pytest
|
|
|
|
from django_q.core import scheduler, pusher, worker, monitor, redis_client, schedule as create_schedule
|
|
from django_q import Schedule, get_task
|
|
|
|
|
|
@pytest.fixture
|
|
def r():
|
|
return redis_client
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_scheduler(r):
|
|
list_key = 'scheduler_test:q'
|
|
r.delete(list_key)
|
|
schedule = create_schedule('math.copysign',
|
|
1, -1,
|
|
hook='django_q.tests.tasks.result',
|
|
schedule_type=Schedule.HOURLY,
|
|
repeats=1)
|
|
assert schedule.last_run() is None
|
|
# run scheduler
|
|
scheduler(list_key=list_key)
|
|
# set up the workflow
|
|
task_queue = Queue()
|
|
stop_event = Event()
|
|
stop_event.set()
|
|
# push it
|
|
pusher(task_queue, stop_event, list_key=list_key, r=r)
|
|
assert task_queue.qsize() == 1
|
|
assert r.llen(list_key) == 0
|
|
task_queue.put('STOP')
|
|
# let a worker handle them
|
|
result_queue = Queue()
|
|
worker(task_queue, result_queue, Value('b', -1))
|
|
assert result_queue.qsize() == 1
|
|
result_queue.put('STOP')
|
|
# store the results
|
|
monitor(result_queue)
|
|
assert result_queue.qsize() == 0
|
|
schedule.refresh_from_db()
|
|
assert schedule.repeats == 0
|
|
assert schedule.last_run() is not None
|
|
assert schedule.success() is True
|
|
task = get_task(schedule.task)
|
|
assert task is not None
|
|
assert task.success is True
|
|
assert task.result < 0
|
|
for t in Schedule.TYPE:
|
|
schedule = create_schedule('django_q.tests.tasks.word_multiply',
|
|
2,
|
|
word='django',
|
|
schedule_type=t[0],
|
|
repeats=1,
|
|
hook='django_q.tests.tasks.result'
|
|
)
|
|
assert schedule is not None
|
|
assert schedule.last_run() is None
|