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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
|
import threading
import time
from datetime import datetime, timezone
import anyio
import pytest
from anyio import fail_after
from apscheduler.enums import JobOutcome
from apscheduler.events import (
Event, JobAdded, ScheduleAdded, ScheduleRemoved, SchedulerStarted, SchedulerStopped, TaskAdded)
from apscheduler.exceptions import JobLookupError
from apscheduler.schedulers.async_ import AsyncScheduler
from apscheduler.schedulers.sync import Scheduler
from apscheduler.triggers.date import DateTrigger
pytestmark = pytest.mark.anyio
async def dummy_async_job(delay: float = 0, fail: bool = False) -> str:
await anyio.sleep(delay)
if fail:
raise RuntimeError('failing as requested')
else:
return 'returnvalue'
def dummy_sync_job(delay: float = 0, fail: bool = False) -> str:
time.sleep(delay)
if fail:
raise RuntimeError('failing as requested')
else:
return 'returnvalue'
class TestAsyncScheduler:
async def test_schedule_job(self) -> None:
def listener(received_event: Event) -> None:
received_events.append(received_event)
if len(received_events) == 5:
event.set()
received_events: list[Event] = []
event = anyio.Event()
scheduler = AsyncScheduler(start_worker=False)
scheduler.events.subscribe(listener)
trigger = DateTrigger(datetime.now(timezone.utc))
async with scheduler:
await scheduler.add_schedule(dummy_async_job, trigger, id='foo')
with fail_after(3):
await event.wait()
# The scheduler was first started
received_event = received_events.pop(0)
assert isinstance(received_event, SchedulerStarted)
# Then the task was added
received_event = received_events.pop(0)
assert isinstance(received_event, TaskAdded)
assert received_event.task_id == 'test_schedulers:dummy_async_job'
# Then a schedule was added
received_event = received_events.pop(0)
assert isinstance(received_event, ScheduleAdded)
assert received_event.schedule_id == 'foo'
# assert received_event.task_id == 'task_id'
# Then that schedule was processed and a job was added for it
received_event = received_events.pop(0)
assert isinstance(received_event, JobAdded)
assert received_event.schedule_id == 'foo'
assert received_event.task_id == 'test_schedulers:dummy_async_job'
# Then the schedule was removed since the trigger had been exhausted
received_event = received_events.pop(0)
assert isinstance(received_event, ScheduleRemoved)
assert received_event.schedule_id == 'foo'
# Finally, the scheduler was stopped
received_event = received_events.pop(0)
assert isinstance(received_event, SchedulerStopped)
# There should be no more events on the list
assert not received_events
async def test_get_job_result_success(self) -> None:
async with AsyncScheduler() as scheduler:
job_id = await scheduler.add_job(dummy_async_job, kwargs={'delay': 0.2})
result = await scheduler.get_job_result(job_id)
assert result.job_id == job_id
assert result.outcome is JobOutcome.success
assert result.return_value == 'returnvalue'
async def test_get_job_result_error(self) -> None:
async with AsyncScheduler() as scheduler:
job_id = await scheduler.add_job(dummy_async_job, kwargs={'delay': 0.2, 'fail': True})
result = await scheduler.get_job_result(job_id)
assert result.job_id == job_id
assert result.outcome is JobOutcome.error
assert isinstance(result.exception, RuntimeError)
assert str(result.exception) == 'failing as requested'
async def test_get_job_result_nowait_not_yet_ready(self) -> None:
async with AsyncScheduler() as scheduler:
job_id = await scheduler.add_job(dummy_async_job, kwargs={'delay': 0.2})
with pytest.raises(JobLookupError):
await scheduler.get_job_result(job_id, wait=False)
async def test_run_job_success(self) -> None:
async with AsyncScheduler() as scheduler:
return_value = await scheduler.run_job(dummy_async_job)
assert return_value == 'returnvalue'
async def test_run_job_failure(self) -> None:
async with AsyncScheduler() as scheduler:
with pytest.raises(RuntimeError, match='failing as requested'):
await scheduler.run_job(dummy_async_job, kwargs={'fail': True})
class TestSyncScheduler:
def test_schedule_job(self):
def listener(received_event: Event) -> None:
received_events.append(received_event)
if len(received_events) == 5:
event.set()
received_events: list[Event] = []
event = threading.Event()
scheduler = Scheduler(start_worker=False)
scheduler.events.subscribe(listener)
trigger = DateTrigger(datetime.now(timezone.utc))
with scheduler:
scheduler.add_schedule(dummy_sync_job, trigger, id='foo')
event.wait(3)
# The scheduler was first started
received_event = received_events.pop(0)
assert isinstance(received_event, SchedulerStarted)
# Then the task was added
received_event = received_events.pop(0)
assert isinstance(received_event, TaskAdded)
assert received_event.task_id == 'test_schedulers:dummy_sync_job'
# Then a schedule was added
received_event = received_events.pop(0)
assert isinstance(received_event, ScheduleAdded)
assert received_event.schedule_id == 'foo'
# Then that schedule was processed and a job was added for it
received_event = received_events.pop(0)
assert isinstance(received_event, JobAdded)
assert received_event.schedule_id == 'foo'
assert received_event.task_id == 'test_schedulers:dummy_sync_job'
# Then the schedule was removed since the trigger had been exhausted
received_event = received_events.pop(0)
assert isinstance(received_event, ScheduleRemoved)
assert received_event.schedule_id == 'foo'
# Finally, the scheduler was stopped
received_event = received_events.pop(0)
assert isinstance(received_event, SchedulerStopped)
# There should be no more events on the list
assert not received_events
def test_get_job_result(self) -> None:
with Scheduler() as scheduler:
job_id = scheduler.add_job(dummy_sync_job)
result = scheduler.get_job_result(job_id)
assert result.outcome is JobOutcome.success
assert result.return_value == 'returnvalue'
def test_get_job_result_error(self) -> None:
with Scheduler() as scheduler:
job_id = scheduler.add_job(dummy_sync_job, kwargs={'delay': 0.2, 'fail': True})
result = scheduler.get_job_result(job_id)
assert result.job_id == job_id
assert result.outcome is JobOutcome.error
assert isinstance(result.exception, RuntimeError)
assert str(result.exception) == 'failing as requested'
def test_get_job_result_nowait_not_yet_ready(self) -> None:
with Scheduler() as scheduler:
job_id = scheduler.add_job(dummy_sync_job, kwargs={'delay': 0.2})
with pytest.raises(JobLookupError):
scheduler.get_job_result(job_id, wait=False)
def test_run_job_success(self) -> None:
with Scheduler() as scheduler:
return_value = scheduler.run_job(dummy_sync_job)
assert return_value == 'returnvalue'
def test_run_job_failure(self) -> None:
with Scheduler() as scheduler:
with pytest.raises(RuntimeError, match='failing as requested'):
scheduler.run_job(dummy_sync_job, kwargs={'fail': True})
|
