mrq/tests/test_pause.py at performance · 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
from mrq.job import Job
import pytest
from mrq.queue import Queue, send_task
import time
from mrq.context import set_current_config, get_config


def test_pause_resume(worker):

    worker.start(flags="--paused_queues_refresh_interval=0.1")

    Queue("high").pause()

    assert Queue("high").is_paused()

    # wait for the paused_queues list to be refreshed
    time.sleep(2)

    job_id1 = send_task(
        "tests.tasks.general.MongoInsert", {"a": 41},
        queue="high")

    job_id2 = send_task(
        "tests.tasks.general.MongoInsert", {"a": 43},
        queue="low")

    time.sleep(5)

    job1 = Job(job_id1).fetch().data
    job2 = Job(job_id2).fetch().data

    assert job1["status"] == "queued"

    assert job2["status"] == "success"
    assert job2["result"] == {"a": 43}

    assert worker.mongodb_jobs.tests_inserts.count() == 1

    Queue("high").resume()

    Job(job_id1).wait(poll_interval=0.01)

    job1 = Job(job_id1).fetch().data

    assert job1["status"] == "success"
    assert job1["result"] == {"a": 41}

    assert worker.mongodb_jobs.tests_inserts.count() == 2


def test_pause_refresh_interval(worker):

    """ Tests that a refresh interval of 0 disables the pause functionnality """

    worker.start(flags="--paused_queues_refresh_interval=0")

    Queue("high").pause()

    assert Queue("high").is_paused()

    # wait for the paused_queues list to be refreshed
    time.sleep(2)

    job_id1 = send_task(
        "tests.tasks.general.MongoInsert", {"a": 41},
        queue="high")

    time.sleep(5)

    job1 = Job(job_id1).fetch().data

    assert job1["status"] == "success"
    assert job1["result"] == {"a": 41}


def test_pause_subqueue(worker):

    # set config in current context in order to have a subqueue delimiter
    set_current_config(get_config(config_type="worker"))

    worker.start(queues="high high/", flags="--subqueues_refresh_interval=1 --paused_queues_refresh_interval=1")

    Queue("high").pause()

    assert Queue("high/").is_paused()

    # wait for the paused_queues list to be refreshed
    time.sleep(2)

    job_id1 = send_task(
        "tests.tasks.general.MongoInsert", {"a": 41},
        queue="high")

    job_id2 = send_task(
        "tests.tasks.general.MongoInsert", {"a": 43},
        queue="high/subqueue")

    # wait a bit to make sure the jobs status will still be queued
    time.sleep(5)

    job1 = Job(job_id1).fetch().data
    job2 = Job(job_id2).fetch().data

    assert job1["status"] == "queued"
    assert job2["status"] == "queued"

    assert worker.mongodb_jobs.tests_inserts.count() == 0

    Queue("high/").resume()

    Job(job_id1).wait(poll_interval=0.01)

    Job(job_id2).wait(poll_interval=0.01)

    job1 = Job(job_id1).fetch().data
    job2 = Job(job_id2).fetch().data

    assert job1["status"] == "success"
    assert job1["result"] == {"a": 41}

    assert job2["status"] == "success"
    assert job2["result"] == {"a": 43}

    assert worker.mongodb_jobs.tests_inserts.count() == 2