From f08773f37d8f205f20755bbcac5d20edca8794ef Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Fri, 26 Jun 2015 17:20:09 +0200 Subject: [PATCH] Improved cluster test --- django_q/__init__.py | 3 +- django_q/core.py | 1 - django_q/tests/tasks.py | 3 ++ django_q/tests/test_cluster.py | 67 ++++++++++++++++++++++------------ 4 files changed, 49 insertions(+), 25 deletions(-) diff --git a/django_q/__init__.py b/django_q/__init__.py index 712f024..3604677 100644 --- a/django_q/__init__.py +++ b/django_q/__init__.py @@ -23,4 +23,5 @@ def get_task(name): :return: the full task object :rtype: Task """ - return Task.objects.get(name=name) + if Task.objects.filter(name=name).exists(): + return Task.objects.get(name=name) diff --git a/django_q/core.py b/django_q/core.py index aeabc31..b77121c 100644 --- a/django_q/core.py +++ b/django_q/core.py @@ -20,7 +20,6 @@ from multiprocessing import Queue, Event, Process, current_process import socket import sys from time import sleep -import gc try: import cPickle as pickle diff --git a/django_q/tests/tasks.py b/django_q/tests/tasks.py index 7e328a6..256bff4 100644 --- a/django_q/tests/tasks.py +++ b/django_q/tests/tasks.py @@ -15,6 +15,9 @@ def count_letters(tup): def count_letters2(obj): return count_letters(obj.get_words()) +def get_task_name(task): + return task.name + def result(obj): print('RESULT HOOK {} : {}'.format(obj.name, obj.result)) diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 2c6871b..5e893f5 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -1,6 +1,5 @@ import sys import os -from time import sleep from multiprocessing import Queue, Event import pytest @@ -10,7 +9,7 @@ sys.path.insert(0, myPath + '/../') from django_q.core import Cluster, r, async, pusher, worker, monitor, Sentinel from django_q.humanhash import DEFAULT_WORDLIST -from django_q import result, get_task +from django_q import result, get_task, Task from django_q.tests.tasks import multiply @@ -30,10 +29,10 @@ def test_cluster_initial(): c = Cluster() assert c.sentinel is None assert c.is_idle - c.start() + assert c.start() > 0 assert c.sentinel.is_alive() is True assert c.is_running - c.stop() + assert c.stop() is True assert c.sentinel.is_alive() is False assert c.has_stopped @@ -77,30 +76,25 @@ def test_cluster(): @pytest.mark.django_db -def run_cluster(): - list_key = 'run_test:q' +def test_async(): + list_key = 'cluster_test:q' r.delete(list_key) - c = Cluster(list_key=list_key) - assert c.start() > 0 - while c.stat.task_q_size > 0 and c.stat.done_q_size > 0: - sleep(0.5) - assert c.stop() is True - r.delete(list_key) - - -@pytest.mark.django_db -def blah_async(): - a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_q.assert_result') - b = async('django_q.tests.tasks.count_letters2', WordClass(), hook='django_q.tests.test_q.assert_result') + a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_cluster.assert_result', + list_key=list_key) + b = async('django_q.tests.tasks.count_letters2', WordClass(), hook='django_q.tests.test_cluster.assert_result', + list_key=list_key) # unknown argument c = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, 'oneargumentoomany', - hook='django_q.tests.test_q.assert_bad_result') + hook='django_q.tests.test_cluster.assert_bad_result', list_key=list_key) # unknown function - d = async('django_q.tests.tasks.does_not_exist', WordClass(), hook='django_q.tests.test_q.assert_bad_result') + d = async('django_q.tests.tasks.does_not_exist', WordClass(), hook='django_q.tests.test_cluster.assert_bad_result', + list_key=list_key) # function without result - e = async('django_q.tests.tasks.countdown', 100000) + e = async('django_q.tests.tasks.countdown', 100000, list_key=list_key) # function as instance - f = async(multiply, 753, 2, hook=assert_result) + f = async(multiply, 753, 2, hook=assert_result, list_key=list_key) + # model as argument + g = async('django_q.tests.tasks.get_task_name', Task(name='John'), list_key=list_key) # check if everything has a task name assert isinstance(a, str) assert isinstance(b, str) @@ -108,8 +102,28 @@ def blah_async(): assert isinstance(d, str) assert isinstance(e, str) assert isinstance(f, str) + assert isinstance(g, str) # run the cluster to execute the tasks - run_cluster() + task_count = 7 + assert r.llen(list_key) == task_count + task_queue = Queue() + stop_event = Event() + stop_event.set() + # push the tasks + for i in range(task_count): + pusher(task_queue, stop_event, list_key=list_key) + assert r.llen(list_key) == 0 + assert task_queue.qsize() == task_count + task_queue.put('STOP') + # let a worker handle them + result_queue = Queue() + worker(task_queue, result_queue) + assert result_queue.qsize() == task_count + result_queue.put('STOP') + # store the results + monitor(result_queue) + assert result_queue.qsize() == 0 + # Check the results # task a result_a = get_task(a) assert result_a is not None @@ -138,6 +152,13 @@ def blah_async(): assert result_f is not None assert result_f.success is True assert result(f) == 1506 + # task g + result_g = get_task(g) + assert result_g is not None + assert result_g.success is True + assert result(g) == 'John' + r.delete(list_key) + # not sure if this actually asserts, but it is called @pytest.mark.django_db