mrq/tests/test_timeout.py at master · pricingassistant/mrq · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
import time


def test_timeout_normal(worker):

    worker.start(flags="--config tests/fixtures/config1.py")

    r = worker.send_task(
        "tests.tasks.general.TimeoutFromConfig", {"a": 1, "b": 2}, block=True)
    assert r == 3

    r = worker.send_task("tests.tasks.general.TimeoutFromConfig", {
                         "a": 1, "b": 2, "sleep": 1000}, block=True, accept_statuses=["timeout"])
    assert r != 3

    r = worker.send_task("tests.tasks.general.TimeoutFromConfig", {
                         "a": 1, "b": 2, "sleep": 1000, "broadexcept": True}, block=True, accept_statuses=["timeout"])
    assert r != 3

def test_timeout_global_config(worker):

    worker.start(env={"MRQ_DEFAULT_JOB_TIMEOUT": "1"})

    assert worker.send_task("tests.tasks.general.Add", {"a": 41, "b": 1, "sleep": 0}) == 42
    assert worker.send_task("tests.tasks.general.Add", {"a": 41, "b": 1, "sleep": 2}, block=True, accept_statuses=["timeout"]) != 42


def test_timeout_subpool(worker):

    worker.start(env={"MRQ_DEFAULT_JOB_TIMEOUT": "1"})

    jobs = worker.send_task("tests.tasks.general.ListJobsByGreenlets", {}, block=True)

    other_jobs = [j for j in jobs["job_ids"] if j != jobs["current_job_id"]]
    assert len(other_jobs) == 0
    # Only one greenlet in current job
    assert jobs["job_ids"] == [jobs["current_job_id"]]

    r = worker.send_task("tests.tasks.general.SubPool", {
        "pool_size": 3,
        "inner_params": [1000, 1001, 1002, 1003, 1004]
    }, block=True, accept_statuses=["timeout"])
    assert r is None

    time.sleep(1)

    jobs = worker.send_task("tests.tasks.general.ListJobsByGreenlets", {}, block=True)

    other_jobs = [j for j in jobs["job_ids"] if j != jobs["current_job_id"]]
    print("Leftover greenlets: %s" % other_jobs)
    assert len(other_jobs) == 0
    # Only one greenlet in current job
    assert jobs["job_ids"] == [jobs["current_job_id"]]