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

Reduce code duplication in timeout unit tests

Browse Source
This commit is contained in:
Janne Rönkkö
2019-01-26 20:44:31 +02:00
parent 7ffb3e4da2
commit 888324cf48
1 changed files with 7 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/tests/test_cluster.py
+7 -19
View File
@@ -244,32 +244,20 @@ def test_enqueue(broker, admin_user):
@pytest.mark.django_db
def test_timeout(broker):
@pytest.mark.parametrize('cluster_config_timeout, async_task_kwargs', (
(1, {}),
(10, {'timeout': 1}),
))
def test_timeout(broker, cluster_config_timeout, async_task_kwargs):
# set up the Sentinel
broker.list_key = 'timeout_test:q'
broker.purge_queue()
async_task('django_q.tests.tasks.count_forever', broker=broker)
async_task('django_q.tests.tasks.count_forever', broker=broker, **async_task_kwargs)
start_event = Event()
stop_event = Event()
# Set a timer to stop the Sentinel
threading.Timer(3, stop_event.set).start()
s = Sentinel(stop_event, start_event, broker=broker, timeout=1)
assert start_event.is_set()
assert s.status() == Conf.STOPPED
assert s.reincarnations == 1
broker.delete_queue()
@pytest.mark.django_db
def test_timeout_override(broker):
# set up the Sentinel
broker.list_key = 'timeout_override_test:q'
async_task('django_q.tests.tasks.count_forever', broker=broker, timeout=1)
start_event = Event()
stop_event = Event()
# Set a timer to stop the Sentinel
threading.Timer(3, stop_event.set).start()
s = Sentinel(stop_event, start_event, broker=broker, timeout=10)
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 == 1