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
|
from __future__ import annotations
from datetime import datetime, timedelta
import pytest
from apscheduler import MaxIterationsReached
from apscheduler.triggers.combining import AndTrigger, OrTrigger
from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.interval import IntervalTrigger
class TestAndTrigger:
@pytest.mark.parametrize("threshold", [1, 0])
def test_two_datetriggers(self, timezone, serializer, threshold):
date1 = datetime(2020, 5, 16, 14, 17, 30, 254212, tzinfo=timezone)
date2 = datetime(2020, 5, 16, 14, 17, 31, 254212, tzinfo=timezone)
trigger = AndTrigger(
[DateTrigger(date1), DateTrigger(date2)], threshold=threshold
)
if serializer:
trigger = serializer.deserialize(serializer.serialize(trigger))
if threshold:
# date2 was within the threshold so it will not be used
assert trigger.next() == date1
assert trigger.next() is None
def test_max_iterations(self, timezone, serializer):
start_time = datetime(2020, 5, 16, 14, 17, 30, 254212, tzinfo=timezone)
trigger = AndTrigger(
[
IntervalTrigger(seconds=4, start_time=start_time),
IntervalTrigger(
seconds=4, start_time=start_time + timedelta(seconds=2)
),
]
)
if serializer:
trigger = serializer.deserialize(serializer.serialize(trigger))
pytest.raises(MaxIterationsReached, trigger.next)
def test_repr(self, timezone, serializer):
start_time = datetime(2020, 5, 16, 14, 17, 30, 254212, tzinfo=timezone)
trigger = AndTrigger(
[
IntervalTrigger(seconds=4, start_time=start_time),
IntervalTrigger(
seconds=4, start_time=start_time + timedelta(seconds=2)
),
]
)
if serializer:
trigger = serializer.deserialize(serializer.serialize(trigger))
assert repr(trigger) == (
"AndTrigger([IntervalTrigger(seconds=4, "
"start_time='2020-05-16 14:17:30.254212+02:00'), IntervalTrigger(seconds=4, "
"start_time='2020-05-16 14:17:32.254212+02:00')], threshold=1.0, max_iterations=10000)"
)
class TestOrTrigger:
def test_two_datetriggers(self, timezone, serializer):
date1 = datetime(2020, 5, 16, 14, 17, 30, 254212, tzinfo=timezone)
date2 = datetime(2020, 5, 18, 15, 1, 53, 940564, tzinfo=timezone)
trigger = OrTrigger([DateTrigger(date1), DateTrigger(date2)])
if serializer:
trigger = serializer.deserialize(serializer.serialize(trigger))
assert trigger.next() == date1
assert trigger.next() == date2
assert trigger.next() is None
def test_two_interval_triggers(self, timezone, serializer):
start_time = datetime(2020, 5, 16, 14, 17, 30, 254212, tzinfo=timezone)
end_time1 = start_time + timedelta(seconds=16)
end_time2 = start_time + timedelta(seconds=18)
trigger = OrTrigger(
[
IntervalTrigger(seconds=4, start_time=start_time, end_time=end_time1),
IntervalTrigger(seconds=6, start_time=start_time, end_time=end_time2),
]
)
if serializer:
trigger = serializer.deserialize(serializer.serialize(trigger))
assert trigger.next() == start_time
assert trigger.next() == start_time + timedelta(seconds=4)
assert trigger.next() == start_time + timedelta(seconds=6)
assert trigger.next() == start_time + timedelta(seconds=8)
assert trigger.next() == start_time + timedelta(seconds=12)
assert trigger.next() == start_time + timedelta(seconds=16)
# The end time of the 4 second interval has been reached
assert trigger.next() == start_time + timedelta(seconds=18)
# The end time of the 6 second interval has been reached
assert trigger.next() is None
def test_repr(self, timezone):
date1 = datetime(2020, 5, 16, 14, 17, 30, 254212, tzinfo=timezone)
date2 = datetime(2020, 5, 18, 15, 1, 53, 940564, tzinfo=timezone)
trigger = OrTrigger([DateTrigger(date1), DateTrigger(date2)])
assert repr(trigger) == (
"OrTrigger([DateTrigger('2020-05-16 14:17:30.254212+02:00'), "
"DateTrigger('2020-05-18 15:01:53.940564+02:00')])"
)
|
