You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
180 lines
5.0 KiB
180 lines
5.0 KiB
# -*- 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
|
|
|