Function test_cluster

django_q/tests/test_cluster.py:105–133

(broker)

@pytest.mark.django_db
def test_cluster(broker):
    broker.list_key = "cluster_test:q"
    broker.delete_queue()
    task = async_task(
        "django_q.tests.tasks.count_letters", DEFAULT_WORDLIST, broker=broker
    )
    assert broker.queue_size() == 1
    task_queue = Queue()
    assert task_queue.qsize() == 0
    result_queue = Queue()
    assert result_queue.qsize() == 0
    event = Event()
    event.set()
    # Test push
    pusher(task_queue, event, broker=broker)
    assert task_queue.qsize() == 1
    assert queue_size(broker=broker) == 0
    # Test work
    task_queue.put("STOP")
    worker(task_queue, result_queue, Value("f", -1))
    assert task_queue.qsize() == 0
    assert result_queue.qsize() == 1
    # Test monitor
    result_queue.put("STOP")
    monitor(result_queue)
    assert result_queue.qsize() == 0
    # check result
    assert result(task) == 1506
    broker.delete_queue()


@pytest.mark.django_db