mrq/tests/test_retry.py at 826ccbd04c167d7cef9ec32d5dca72747d803900 · 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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
from mrq.job import Job
import datetime
from mrq.queue import Queue
import time


def test_retry(worker):

    job_id = worker.send_task(
        "tests.tasks.general.Retry", {"queue": "noexec", "delay": 60}, block=False)

    job_data = Job(job_id).wait(poll_interval=0.01, full_data=True)

    assert job_data["queue"] == "noexec"
    assert job_data["status"] == "retry"
    assert job_data["dateretry"] > datetime.datetime.utcnow()
    assert job_data.get("result") is None


def test_retry_otherqueue_delay_zero(worker):

    # delay = 0 should requeue right away.
    job_id = worker.send_task(
        "tests.tasks.general.Retry", {"queue": "noexec", "delay": 0}, block=False)

    time.sleep(1)

    assert worker.mongodb_logs.tests_inserts.find().count() == 1

    assert Queue("default").size() == 0
    assert Queue("noexec").size() == 1
    assert Queue("noexec").list_job_ids() == [str(job_id)]


def test_retry_otherqueue_delay_nonzero(worker):

    # delay = 0 should requeue right away.
    worker.send_task("tests.tasks.general.Retry", {
        "queue": "noexec",
        "delay": 2
    }, block=True, accept_statuses=["retry"])

    assert Queue("default").size() == 0
    assert Queue("noexec").size() == 0

    job_id = worker.mongodb_jobs.mrq_jobs.find()[0]["_id"]

    job = Job(job_id).fetch()
    assert job.data["status"] == "retry"

    # Should do nothing yet
    worker.send_task("mrq.basetasks.cleaning.RequeueRetryJobs", {}, block=True)

    assert Queue("default").size() == 0
    assert Queue("noexec").size() == 0

    time.sleep(2)

    # Should requeue
    worker.send_task("mrq.basetasks.cleaning.RequeueRetryJobs", {}, block=True)

    assert Queue("default").size() == 0
    assert Queue("noexec").size() == 1

    job = Job(job_id).fetch()
    assert job.data["status"] == "queued"


def test_retry_max_retries(worker):

    # Task has maxretries=1
    worker.start(flags="--config tests/fixtures/config-retry1.py")

    worker.send_task("tests.tasks.general.Retry", {

    }, block=True, accept_statuses=["retry"])

    assert Queue("default").size() == 0

    job_id = worker.mongodb_jobs.mrq_jobs.find()[0]["_id"]

    job = Job(job_id).fetch()
    assert job.data["status"] == "retry"
    assert job.data["retry_count"] == 1

    time.sleep(2)

    # Should requeue
    worker.send_task("mrq.basetasks.cleaning.RequeueRetryJobs", {}, block=True)

    time.sleep(2)

    assert Queue("default").size() == 0

    job = Job(job_id).fetch()
    assert job.data["status"] == "maxretries"
    assert job.data["retry_count"] == 1

    # Then, manual requeue from the dashboard should reset the retry_count field.
    params = {
        "action": "requeue",
        "status": "maxretries",
        "destination_queue": "noexec"
    }

    worker.send_task("mrq.basetasks.utils.JobAction", params, block=True)

    job = Job(job_id).fetch()
    assert job.data["status"] == "queued"
    assert job.data["queue"] == "noexec"
    assert job.data["retry_count"] == 0


def test_retry_max_retries_zero(worker):

    # Task has maxretries=1
    worker.start(flags="--config tests/fixtures/config-retry1.py")

    worker.send_task("tests.tasks.general.Retry", {
        "max_retries": 0
    }, block=True, accept_statuses=["maxretries"])

    assert Queue("default").size() == 0

    job_id = worker.mongodb_jobs.mrq_jobs.find()[0]["_id"]

    job = Job(job_id).fetch()
    assert job.data["status"] == "maxretries"