mirror of
https://github.com/django-q2/django-q2.git
synced 2026-09-20 03:18:07 +08:00
cf37efb1eb
The test suite, especially for the brokers, was heavily dependant on the success of the tests preceeding it to pass. This commit should eliminate that dependency and allow tests to fail without affecting others. It also removes the need to cleanup globals manually after a test.
189 lines
6.5 KiB
Python
189 lines
6.5 KiB
Python
from multiprocessing import Event, Queue, Value
|
|
|
|
import pytest
|
|
|
|
from django_q.cluster import pusher, worker, monitor
|
|
from django_q.conf import Conf
|
|
from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group, delete_cached, \
|
|
async_iter, Chain, async_chain, Iter, Async
|
|
from django_q.brokers import get_broker
|
|
|
|
|
|
@pytest.fixture
|
|
def broker(monkeypatch):
|
|
monkeypatch.setattr(Conf, 'DJANGO_REDIS', 'default')
|
|
return get_broker()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_cached(broker):
|
|
broker.purge_queue()
|
|
broker.cache.clear()
|
|
group = 'cache_test'
|
|
# queue the tests
|
|
task_id = async('math.copysign', 1, -1, cached=True, broker=broker)
|
|
async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
|
|
async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
|
|
async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
|
|
async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
|
|
async('math.copysign', 1, -1, cached=True, broker=broker, group=group)
|
|
async('math.popysign', 1, -1, cached=True, broker=broker, group=group)
|
|
iter_id = async_iter('math.floor', [i for i in range(10)], cached=True)
|
|
# test wait on cache
|
|
# test wait timeout
|
|
assert result(task_id, wait=10, cached=True) is None
|
|
assert fetch(task_id, wait=10, cached=True) is None
|
|
assert result_group(group, wait=10, cached=True) is None
|
|
assert result_group(group, count=2, wait=10, cached=True) is None
|
|
assert fetch_group(group, wait=10, cached=True) is None
|
|
assert fetch_group(group, count=2, wait=10, cached=True) is None
|
|
# run a single inline cluster
|
|
task_count = 17
|
|
assert broker.queue_size() == task_count
|
|
task_queue = Queue()
|
|
stop_event = Event()
|
|
stop_event.set()
|
|
for i in range(task_count):
|
|
pusher(task_queue, stop_event, broker=broker)
|
|
assert broker.queue_size() == 0
|
|
assert task_queue.qsize() == task_count
|
|
task_queue.put('STOP')
|
|
result_queue = Queue()
|
|
worker(task_queue, result_queue, Value('f', -1))
|
|
assert result_queue.qsize() == task_count
|
|
result_queue.put('STOP')
|
|
monitor(result_queue)
|
|
assert result_queue.qsize() == 0
|
|
# assert results
|
|
assert result(task_id, wait=500, cached=True) == -1
|
|
assert fetch(task_id, wait=500, cached=True).result == -1
|
|
# make sure it's not in the db backend
|
|
assert fetch(task_id) is None
|
|
# assert group
|
|
assert count_group(group, cached=True) == 6
|
|
assert count_group(group, cached=True, failures=True) == 1
|
|
assert result_group(group, cached=True) == [-1, -1, -1, -1, -1]
|
|
assert len(result_group(group, cached=True, failures=True)) == 6
|
|
assert len(fetch_group(group, cached=True)) == 6
|
|
assert len(fetch_group(group, cached=True, failures=False)) == 5
|
|
delete_group(group, cached=True)
|
|
assert count_group(group, cached=True) is None
|
|
delete_cached(task_id)
|
|
assert result(task_id, cached=True) is None
|
|
assert fetch(task_id, cached=True) is None
|
|
# iter cached
|
|
assert result(iter_id) is None
|
|
assert result(iter_id, cached=True) is not None
|
|
broker.cache.clear()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_iter(broker):
|
|
broker.purge_queue()
|
|
broker.cache.clear()
|
|
it = [i for i in range(10)]
|
|
it2 = [(1, -1), (2, -1), (3, -4), (5, 6)]
|
|
it3 = (1, 2, 3, 4, 5)
|
|
t = async_iter('math.floor', it, sync=True)
|
|
t2 = async_iter('math.copysign', it2, sync=True)
|
|
t3 = async_iter('math.floor', it3, sync=True)
|
|
t4 = async_iter('math.floor', (1,), sync=True)
|
|
result_t = result(t)
|
|
assert result_t is not None
|
|
task_t = fetch(t)
|
|
assert task_t.result == result_t
|
|
assert result(t2) is not None
|
|
assert result(t3) is not None
|
|
assert result(t4)[0] == 1
|
|
# test iter class
|
|
i = Iter('math.copysign', sync=True, cached=True)
|
|
i.append(1, -1)
|
|
i.append(2, -1)
|
|
i.append(3, -4)
|
|
i.append(5, 6)
|
|
assert i.started is False
|
|
assert i.length() == 4
|
|
assert i.run() is not None
|
|
assert len(i.result()) == 4
|
|
assert len(i.fetch().result) == 4
|
|
i.append(1, -7)
|
|
assert i.result() is None
|
|
i.run()
|
|
assert len(i.result()) == 5
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_chain(broker):
|
|
broker.purge_queue()
|
|
broker.cache.clear()
|
|
task_chain = Chain(sync=True)
|
|
task_chain.append('math.floor', 1)
|
|
task_chain.append('math.copysign', 1, -1)
|
|
task_chain.append('math.floor', 2)
|
|
assert task_chain.length() == 3
|
|
assert task_chain.current() is None
|
|
task_chain.run()
|
|
r = task_chain.result(wait=1000)
|
|
assert task_chain.current() == task_chain.length()
|
|
assert len(r) == task_chain.length()
|
|
t = task_chain.fetch()
|
|
assert len(t) == task_chain.length()
|
|
task_chain.cached = True
|
|
task_chain.append('math.floor', 3)
|
|
assert task_chain.length() == 4
|
|
task_chain.run()
|
|
r = task_chain.result(wait=1000)
|
|
assert task_chain.current() == task_chain.length()
|
|
assert len(r) == task_chain.length()
|
|
t = task_chain.fetch()
|
|
assert len(t) == task_chain.length()
|
|
# test single
|
|
rid = async_chain(['django_q.tests.tasks.hello', 'django_q.tests.tasks.hello'], sync=True, cached=True)
|
|
assert result_group(rid, cached=True) == ['hello', 'hello']
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_async_class(broker, monkeypatch):
|
|
broker.purge_queue()
|
|
broker.cache.clear()
|
|
a = Async('math.copysign')
|
|
assert a.func == 'math.copysign'
|
|
a.args = (1, -1)
|
|
assert a.started is False
|
|
a.cached = True
|
|
assert a.cached is True
|
|
a.sync = True
|
|
assert a.sync is True
|
|
a.broker = broker
|
|
assert a.broker == broker
|
|
a.run()
|
|
assert a.result() == -1
|
|
assert a.fetch().result == -1
|
|
# again with kwargs
|
|
a = Async('math.copysign', 1, -1, cached=True, sync=True, broker=broker)
|
|
a.run()
|
|
assert a.result() == -1
|
|
# with q_options
|
|
a = Async('math.copysign', 1, -1, q_options={'cached': True, 'sync': False, 'broker': broker})
|
|
assert a.sync is False
|
|
a.sync = True
|
|
assert a.kwargs['q_options']['sync'] is True
|
|
a.run()
|
|
assert a.result() == -1
|
|
a.group = 'async_class_test'
|
|
assert a.group == 'async_class_test'
|
|
a.save = False
|
|
assert a.save is False
|
|
a.hook = 'djq.tests.tasks.hello'
|
|
assert a.hook == 'djq.tests.tasks.hello'
|
|
assert a.started is False
|
|
a.run()
|
|
assert a.result_group() == [-1]
|
|
assert a.fetch_group() == [a.fetch()]
|
|
# global overrides
|
|
monkeypatch.setattr(Conf, 'SYNC', True)
|
|
monkeypatch.setattr(Conf, 'CACHED', True)
|
|
a = Async('math.floor', 1.5)
|
|
a.run()
|
|
assert a.result() == 1
|