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
|
# -*- coding: utf-8 -*-
"""
Tests for discord.ext.tasks
"""
import asyncio
import datetime
import pytest
import sys
from discord import utils
from discord.ext import tasks
@pytest.mark.asyncio
async def test_explicit_initial_runs_tomorrow_single():
now = utils.utcnow()
if not ((0, 4) < (now.hour, now.minute) < (23, 59)):
await asyncio.sleep(5 * 60) # sleep for 5 minutes
now = utils.utcnow()
has_run = False
async def inner():
nonlocal has_run
has_run = True
time = utils.utcnow() - datetime.timedelta(minutes=1)
# a loop that should have an initial run tomorrow
loop = tasks.loop(time=datetime.time(hour=time.hour, minute=time.minute))(inner)
loop.start()
await asyncio.sleep(1)
try:
assert not has_run
finally:
loop.cancel()
@pytest.mark.asyncio
async def test_explicit_initial_runs_tomorrow_multi():
now = utils.utcnow()
if not ((0, 4) < (now.hour, now.minute) < (23, 59)):
await asyncio.sleep(5 * 60) # sleep for 5 minutes
now = utils.utcnow()
# multiple times that are in the past for today
times = []
for _ in range(3):
now -= datetime.timedelta(minutes=1)
times.append(datetime.time(hour=now.hour, minute=now.minute))
has_run = False
async def inner():
nonlocal has_run
has_run = True
# a loop that should have an initial run tomorrow
loop = tasks.loop(time=times)(inner)
loop.start()
await asyncio.sleep(1)
try:
assert not has_run
finally:
loop.cancel()
def test_task_regression_issue7659():
jst = datetime.timezone(datetime.timedelta(hours=9))
# 00:00, 03:00, 06:00, 09:00, 12:00, 15:00, 18:00, 21:00
times = [datetime.time(hour=h, tzinfo=jst) for h in range(0, 24, 3)]
@tasks.loop(time=times)
async def loop():
pass
before_midnight = datetime.datetime(2022, 3, 12, 23, 50, 59, tzinfo=jst)
after_midnight = before_midnight + datetime.timedelta(minutes=9, seconds=2)
expected_before_midnight = datetime.datetime(2022, 3, 13, 0, 0, 0, tzinfo=jst)
expected_after_midnight = datetime.datetime(2022, 3, 13, 3, 0, 0, tzinfo=jst)
assert loop._get_next_sleep_time(before_midnight) == expected_before_midnight
assert loop._get_next_sleep_time(after_midnight) == expected_after_midnight
today = datetime.date.today()
minute_before = [datetime.datetime.combine(today, time, tzinfo=jst) - datetime.timedelta(minutes=1) for time in times]
for before, expected_time in zip(minute_before, times):
expected = datetime.datetime.combine(today, expected_time, tzinfo=jst)
actual = loop._get_next_sleep_time(before)
assert actual == expected
def test_task_regression_issue7676():
jst = datetime.timezone(datetime.timedelta(hours=9))
# 00:00, 03:00, 06:00, 09:00, 12:00, 15:00, 18:00, 21:00
times = [datetime.time(hour=h, tzinfo=jst) for h in range(0, 24, 3)]
@tasks.loop(time=times)
async def loop():
pass
# Create pseudo UTC times
now = utils.utcnow()
today = now.date()
times_before_in_utc = [
datetime.datetime.combine(today, time, tzinfo=jst).astimezone(datetime.timezone.utc) - datetime.timedelta(minutes=1)
for time in times
]
for before, expected_time in zip(times_before_in_utc, times):
actual = loop._get_next_sleep_time(before)
actual_time = actual.timetz()
assert actual_time == expected_time
@pytest.mark.skipif(sys.version_info < (3, 9), reason="zoneinfo requires 3.9")
def test_task_is_imaginary():
import zoneinfo
tz = zoneinfo.ZoneInfo('America/New_York')
# 2:30 AM was skipped
dt = datetime.datetime(2022, 3, 13, 2, 30, tzinfo=tz)
assert tasks.is_imaginary(dt)
now = utils.utcnow()
# UTC time is never imaginary or ambiguous
assert not tasks.is_imaginary(now)
@pytest.mark.skipif(sys.version_info < (3, 9), reason="zoneinfo requires 3.9")
def test_task_is_ambiguous():
import zoneinfo
tz = zoneinfo.ZoneInfo('America/New_York')
# 1:30 AM happened twice
dt = datetime.datetime(2022, 11, 6, 1, 30, tzinfo=tz)
assert tasks.is_ambiguous(dt)
now = utils.utcnow()
# UTC time is never imaginary or ambiguous
assert not tasks.is_imaginary(now)
@pytest.mark.skipif(sys.version_info < (3, 9), reason="zoneinfo requires 3.9")
@pytest.mark.parametrize(
('dt', 'key', 'expected'),
[
(datetime.datetime(2022, 11, 6, 1, 30), 'America/New_York', datetime.datetime(2022, 11, 6, 1, 30, fold=1)),
(datetime.datetime(2022, 3, 13, 2, 30), 'America/New_York', datetime.datetime(2022, 3, 13, 3, 30)),
(datetime.datetime(2022, 4, 8, 2, 30), 'America/New_York', datetime.datetime(2022, 4, 8, 2, 30)),
(datetime.datetime(2023, 1, 7, 12, 30), 'UTC', datetime.datetime(2023, 1, 7, 12, 30)),
],
)
def test_task_date_resolve(dt, key, expected):
import zoneinfo
tz = zoneinfo.ZoneInfo(key)
actual = tasks.resolve_datetime(dt.replace(tzinfo=tz))
expected = expected.replace(tzinfo=tz)
assert actual == expected
|
