mrq/tests/test_scheduler.py at locked_jobs · 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
import time
import pytest
import datetime


# We want to test that launching the scheduler several times queues tasks
# only once.
PROCESS_CONFIGS = [
    ["--greenlets 1"],
    ["--greenlets 1 --processes 5"]
]


@pytest.mark.parametrize(["p_flags"], PROCESS_CONFIGS)
def test_scheduler_simple(worker, p_flags):

    worker.start(
        flags="--scheduler --config tests/fixtures/config-scheduler1.py %s" % p_flags)

    collection = worker.mongodb_jobs.tests_inserts
    scheduled_jobs = worker.mongodb_jobs.mrq_scheduled_jobs

    while not collection.count():
        time.sleep(1)

    # There are 4 test tasks with 5 second interval
    inserts = list(collection.find())
    assert len(inserts) == 4

    jobs = list(scheduled_jobs.find())
    assert len(jobs) == 4

    time.sleep(5)

    # They should have ran again.
    inserts = list(collection.find())
    assert len(inserts) == 8

    worker.stop(deps=False)

    collection.remove({})

    # Start with new config
    worker.start(
        deps=False, flags="--scheduler --config tests/fixtures/config-scheduler2.py %s" % p_flags)

    while not collection.count():
        time.sleep(1)

    jobs2 = list(scheduled_jobs.find())
    assert len(jobs2) == 4
    assert jobs != jobs2

    # Only 3 should have been replaced and ran immediately again because they
    # have different config.
    inserts = list(collection.find())
    print inserts
    assert len(inserts) == 3, inserts


@pytest.mark.parametrize(["p_flags"], PROCESS_CONFIGS)
def test_scheduler_dailytime(worker, p_flags):

    # Task is scheduled in 3 seconds
    worker.start(
        flags="--scheduler --config tests/fixtures/config-scheduler3.py %s" % p_flags,
        env={
            # We need to pass this in the environment so that each worker has the
            # exact same hash
            "MRQ_TEST_SCHEDULER_TIME": str(time.time() + 5)
        })

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

    # It should be done a first time immediately
    time.sleep(3)
    inserts = list(collection.find())
    assert len(inserts) == 2
    print inserts
    assert collection.find({"params.b": "test"}).count() == 1

    # Then a second time once the dailytime passes
    time.sleep(7)
    assert collection.find().count() == 4
    assert collection.find({"params.b": "test"}).count() == 2

    # Nothing more should happen today
    time.sleep(4)
    assert collection.find().count() == 4
    assert collection.find({"params.b": "test"}).count() == 2


@pytest.mark.skipif("True")  # patch disabling weekday alone
def test_scheduler_weekday(worker):
    # Task is scheduled in 3 seconds
    worker.start(
        flags="--scheduler --config tests/fixtures/config-scheduler4.py",
        env={
            # We need to pass this in the environment so that each worker has the
            # exact same hash
            "MRQ_TEST_SCHEDULER_TIME": str(time.time() + 5)
        })

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

    # It should be done a first time immediately
    time.sleep(3)
    inserts = list(collection.find())
    assert len(inserts) == 1
    print inserts
    assert collection.find({"params.weekday": datetime.datetime.utcnow().weekday()}).count() == 1


def test_scheduler_weekday_dailytime(worker):
    # Task is scheduled in 5 seconds
    worker.start(
        flags="--scheduler --config tests/fixtures/config-scheduler5.py",
        env={
            # We need to pass this in the environment so that each worker has the
            # exact same hash
            "MRQ_TEST_SCHEDULER_TIME": str(time.time() + 5)
        })

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

    # Should be launched a first time
    time.sleep(2)
    assert len(list(collection.find())) == 2

    # the dailytime passes
    time.sleep(7)
    inserts = list(collection.find())
    assert len(inserts) == 3
    print inserts
    assert collection.find({"params.weekday": datetime.datetime.utcnow().weekday(), "params.later": False}).count() == 2

    # more time passes and we do nothing
    time.sleep(7)
    inserts = list(collection.find())
    assert len(inserts) == 3
    print inserts
    assert collection.find({"params.weekday": datetime.datetime.utcnow().weekday(), "params.later": False}).count() == 2


def test_scheduler_monthday(worker):
    # Task is scheduled in 3 seconds
    worker.start(
        flags="--scheduler --config tests/fixtures/config-scheduler6.py",
        env={
            # We need to pass this in the environment so that each worker has the
            # exact same hash
            "MRQ_TEST_SCHEDULER_TIME": str(time.time() + 5)
        })

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

    # It should be done a first time immediately
    time.sleep(3)
    inserts = list(collection.find())
    assert len(inserts) == 1
    assert collection.find({"params.monthday": datetime.datetime.utcnow().day}).count() == 1