<filename>tests/test_ext_tasks.py
# -*- 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

<filename>tests/test_ext_tasks.py
# -*- 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

# -*- coding: utf-8 -*- Tests for discord.ext.tasks # sleep for 5 minutes # a loop that should have an initial run tomorrow # sleep for 5 minutes # multiple times that are in the past for today # a loop that should have an initial run tomorrow # 00:00, 03:00, 06:00, 09:00, 12:00, 15:00, 18:00, 21:00 # 00:00, 03:00, 06:00, 09:00, 12:00, 15:00, 18:00, 21:00 # Create pseudo UTC times # 2:30 AM was skipped # UTC time is never imaginary or ambiguous # 1:30 AM happened twice # UTC time is never imaginary or ambiguous