[tasks] Handle imaginary or ambiguous times due to DST transitions

discord/ext/tasks/__init__.py

@@ -64,6 +64,48 @@ FT = TypeVar('FT', bound=_func)
 ET = TypeVar('ET', bound=Callable[[Any, BaseException], Awaitable[Any]])
 
 
+def is_ambiguous(dt: datetime.datetime) -> bool:
+    if dt.tzinfo is None or isinstance(dt.tzinfo, datetime.timezone):
+        # Naive or fixed timezones are never ambiguous
+        return False
+
+    before = dt.replace(fold=0)
+    after = dt.replace(fold=1)
+
+    same_offset = before.utcoffset() == after.utcoffset()
+    same_dst = before.dst() == after.dst()
+    return not (same_offset and same_dst)
+
+
+def is_imaginary(dt: datetime.datetime) -> bool:
+    if dt.tzinfo is None or isinstance(dt.tzinfo, datetime.timezone):
+        # Naive or fixed timezones are never imaginary
+        return False
+
+    tz = dt.tzinfo
+    dt = dt.replace(tzinfo=None)
+    roundtrip = dt.replace(tzinfo=tz).astimezone(datetime.timezone.utc).astimezone(tz).replace(tzinfo=None)
+    return dt != roundtrip
+
+
+def resolve_datetime(dt: datetime.datetime) -> datetime.datetime:
+    if dt.tzinfo is None or isinstance(dt.tzinfo, datetime.timezone):
+        # Naive or fixed requires no resolution
+        return dt
+
+    if is_imaginary(dt):
+        # Largest gap is probably 24 hours
+        tomorrow = dt + datetime.timedelta(days=1)
+        yesterday = dt - datetime.timedelta(days=1)
+        # utcoffset shouldn't return None since these are aware instances
+        # If it returns None then the timezone implementation was broken from the get go
+        return dt + (tomorrow.utcoffset() - yesterday.utcoffset())  # type: ignore
+    elif is_ambiguous(dt):
+        return dt.replace(fold=1)
+    else:
+        return dt
+
+
 class SleepHandle:
     __slots__ = ('future', 'loop', 'handle')
 
@@ -596,7 +638,7 @@ class Loop(Generic[LF]):
             date = now.date()
             time = self._time[index]
 
-        return datetime.datetime.combine(date, time, tzinfo=time.tzinfo or datetime.timezone.utc)
+        return resolve_datetime(datetime.datetime.combine(date, time, tzinfo=time.tzinfo or datetime.timezone.utc))
 
     def _start_time_relative_to(self, now: datetime.datetime) -> Optional[int]:
         # now kwarg should be a datetime.datetime representing the time "now"

tests/test_ext_tasks.py

@@ -10,6 +10,7 @@ import asyncio
 import datetime
 
 import pytest
+import sys
 
 from discord import utils
 from discord.ext import tasks
@@ -95,3 +96,53 @@ def test_task_regression_issue7659():
 
     assert loop._get_next_sleep_time(before_midnight) == expected_before_midnight
     assert loop._get_next_sleep_time(after_midnight) == expected_after_midnight
+
+
+@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