Improved cluster test

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)

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

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))
 

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():
+def test_async():
-    list_key = 'run_test:q'
+    list_key = 'cluster_test:q'
     r.delete(list_key)
-    c = Cluster(list_key=list_key)
+    a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_cluster.assert_result',
-    assert c.start() > 0
+              list_key=list_key)
-    while c.stat.task_q_size > 0 and c.stat.done_q_size > 0:
+    b = async('django_q.tests.tasks.count_letters2', WordClass(), hook='django_q.tests.test_cluster.assert_result',
-        sleep(0.5)
+              list_key=list_key)
-    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')
     # 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