Browse Source
* tests for testing regression of 19ad64a
* fix edge case for test running within 3 minutes of the start of the hour
pull/7584/head
committed by
GitHub
1 changed files with 77 additions and 0 deletions
@ -0,0 +1,77 @@ |
|||||
|
# -*- coding: utf-8 -*- |
||||
|
|
||||
|
""" |
||||
|
|
||||
|
Tests for discord.ext.tasks |
||||
|
|
||||
|
""" |
||||
|
|
||||
|
import asyncio |
||||
|
import datetime |
||||
|
|
||||
|
import pytest |
||||
|
|
||||
|
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() |
Loading…
Reference in new issue