fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-09-15 13:37:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add test for task finishes before timeout

Browse Source
This commit is contained in:
Janne Rönkkö
2019-01-27 15:28:28 +02:00
parent 888324cf48
commit d09e2658be
1 changed files with 22 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
django_q/tests/test_cluster.py
View File

@@ -264,6 +264,28 @@ def test_timeout(broker, cluster_config_timeout, async_task_kwargs):
broker.delete_queue()
@pytest.mark.django_db
@pytest.mark.parametrize('cluster_config_timeout, async_task_kwargs', (
(5, {}),
(10, {'timeout': 5}),
(1, {'timeout': 5}),
))
def test_timeout_task_finishes(broker, cluster_config_timeout, async_task_kwargs):
# set up the Sentinel
broker.list_key = 'timeout_test:q'
broker.purge_queue()
async_task('time.sleep', 3, broker=broker, **async_task_kwargs)
start_event = Event()
stop_event = Event()
# Set a timer to stop the Sentinel
threading.Timer(6, stop_event.set).start()
s = Sentinel(stop_event, start_event, broker=broker, timeout=cluster_config_timeout)
assert start_event.is_set()
assert s.status() == Conf.STOPPED
assert s.reincarnations == 0
broker.delete_queue()
@pytest.mark.django_db
def test_recycle(broker, monkeypatch):
# set up the Sentinel