OpenSource/python-telegram-bot
OpenSource/python-telegram-bot
1
0
Fork 0
mirror of https://github.com/python-telegram-bot/python-telegram-bot.git synced 2025-01-04 02:01:48 +01:00
Code Issues Projects Releases Packages Wiki Activity

Review

Browse source
This commit is contained in:
Hinrich Mahler 2024-12-31 10:14:24 +01:00
parent 1f8124f3cb
commit 9cabf0e03b
4 changed files with 71 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
telegram/_utils/datetime.py
View file

@ -126,6 +126,12 @@ def to_float_timestamp(
return reference_timestamp + time_object
if tzinfo is None:
# We do this here rather than in the signature to ensure that we can make calls like
# to_float_timestamp(
# time, tzinfo=bot.defaults.tzinfo if bot.defaults else None
# )
# This ensures clean separation of concerns, i.e. the default timezone should not be
# the responsibility of the caller
tzinfo = UTC
if isinstance(time_object, dtm.time):
@ -137,6 +143,8 @@ def to_float_timestamp(
aware_datetime = dtm.datetime.combine(reference_date, time_object)
if aware_datetime.tzinfo is None:
# datetime.combine uses the tzinfo of `time_object`, which might be None
# so we still need to localize
aware_datetime = localize(aware_datetime, tzinfo)
# if the time of day has passed today, use tomorrow

2
telegram/ext/_jobqueue.py
View file

@ -197,6 +197,8 @@ class JobQueue(Generic[CCT]):
dtm.datetime.now(tz=time.tzinfo or self.scheduler.timezone).date(), time
)
if date_time.tzinfo is None:
# dtm.combine uses the tzinfo of `time`, which might be None, so we still have
# to localize it
date_time = localize(date_time, self.scheduler.timezone)
if shift_day and date_time <= dtm.datetime.now(UTC):
date_time += dtm.timedelta(days=1)

82
tests/auxil/bot_method_checks.py
View file

@ -41,15 +41,11 @@ from telegram import (
Sticker,
TelegramObject,
)
from telegram._utils.datetime import to_timestamp
from telegram._utils.defaultvalue import DEFAULT_NONE, DefaultValue
from telegram.constants import InputMediaType
from telegram.ext import Defaults, ExtBot
from telegram.request import RequestData
from tests.auxil.envvars import TEST_WITH_OPT_DEPS
if TEST_WITH_OPT_DEPS:
pass
FORWARD_REF_PATTERN = re.compile(r"ForwardRef\('(?P<class_name>\w+)'\)")
""" A pattern to find a class name in a ForwardRef typing annotation.
@ -396,6 +392,15 @@ def make_assertion_for_link_preview_options(
)
_EUROPE_BERLIN_TS = to_timestamp(
dtm.datetime(2000, 1, 1, 0, tzinfo=zoneinfo.ZoneInfo("Europe/Berlin"))
)
_UTC_TS = to_timestamp(dtm.datetime(2000, 1, 1, 0), tzinfo=zoneinfo.ZoneInfo("UTC"))
_AMERICA_NEW_YORK_TS = to_timestamp(
dtm.datetime(2000, 1, 1, 0, tzinfo=zoneinfo.ZoneInfo("America/New_York"))
)
async def make_assertion(
url,
request_data: RequestData,
@ -535,14 +540,11 @@ async def make_assertion(
for key in date_keys:
date_param = data.pop(key)
if date_param:
if manual_value_expected and date_param != 946681200:
if manual_value_expected and date_param != _EUROPE_BERLIN_TS:
pytest.fail(f"Non-naive `{key}` should have been interpreted as Europe/Berlin.")
if (
not any((manually_passed_value, expected_defaults_value))
and date_param != 946684800
):
if not any((manually_passed_value, expected_defaults_value)) and date_param != _UTC_TS:
pytest.fail(f"Naive `{key}` should have been interpreted as UTC")
if default_value_expected and date_param != 946702800:
if default_value_expected and date_param != _AMERICA_NEW_YORK_TS:
pytest.fail(f"Naive `{key}` should have been interpreted as America/New_York")
if method_name in ["get_file", "get_small_file", "get_big_file"]:
@ -641,35 +643,35 @@ async def check_defaults_handling(
request.post = assertion_callback
assert await method(**kwargs) in expected_return_values
# # 2: test that we get the manually passed non-None value
# kwargs = build_kwargs(
# shortcut_signature, kwargs_need_default, manually_passed_value="non-None-value"
# )
# assertion_callback = functools.partial(
# make_assertion,
# manually_passed_value="non-None-value",
# kwargs_need_default=kwargs_need_default,
# method_name=method.__name__,
# return_value=return_value,
# expected_defaults_value=expected_defaults_value,
# )
# request.post = assertion_callback
# assert await method(**kwargs) in expected_return_values
#
# # 3: test that we get the manually passed None value
# kwargs = build_kwargs(
# shortcut_signature, kwargs_need_default, manually_passed_value=None
# )
# assertion_callback = functools.partial(
# make_assertion,
# manually_passed_value=None,
# kwargs_need_default=kwargs_need_default,
# method_name=method.__name__,
# return_value=return_value,
# expected_defaults_value=expected_defaults_value,
# )
# request.post = assertion_callback
# assert await method(**kwargs) in expected_return_values
# 2: test that we get the manually passed non-None value
kwargs = build_kwargs(
shortcut_signature, kwargs_need_default, manually_passed_value="non-None-value"
)
assertion_callback = functools.partial(
make_assertion,
manually_passed_value="non-None-value",
kwargs_need_default=kwargs_need_default,
method_name=method.__name__,
return_value=return_value,
expected_defaults_value=expected_defaults_value,
)
request.post = assertion_callback
assert await method(**kwargs) in expected_return_values
# 3: test that we get the manually passed None value
kwargs = build_kwargs(
shortcut_signature, kwargs_need_default, manually_passed_value=None
)
assertion_callback = functools.partial(
make_assertion,
manually_passed_value=None,
kwargs_need_default=kwargs_need_default,
method_name=method.__name__,
return_value=return_value,
expected_defaults_value=expected_defaults_value,
)
request.post = assertion_callback
assert await method(**kwargs) in expected_return_values
except Exception as exc:
raise exc
finally:

28
tests/ext/test_jobqueue.py
View file

@ -21,13 +21,12 @@ import calendar
import contextlib
import datetime as dtm
import logging
import platform
import time
import pytest
from telegram.ext import ApplicationBuilder, CallbackContext, ContextTypes, Defaults, Job, JobQueue
from tests.auxil.envvars import GITHUB_ACTION, TEST_WITH_OPT_DEPS
from tests.auxil.envvars import TEST_WITH_OPT_DEPS
from tests.auxil.pytest_classes import make_bot
from tests.auxil.slots import mro_slots
@ -65,13 +64,13 @@ class TestNoJobQueue:
Job(None)
@pytest.mark.skipif(
not TEST_WITH_OPT_DEPS, reason="Only relevant if the optional dependency is installed"
)
@pytest.mark.skipif(
bool(GITHUB_ACTION and platform.system() in ["Windows", "Darwin"]),
reason="On Windows & MacOS precise timings are not accurate.",
)
# @pytest.mark.skipif(
# not TEST_WITH_OPT_DEPS, reason="Only relevant if the optional dependency is installed"
# )
# @pytest.mark.skipif(
# bool(GITHUB_ACTION and platform.system() in ["Windows", "Darwin"]),
# reason="On Windows & MacOS precise timings are not accurate.",
# )
@pytest.mark.flaky(10, 1) # Timings aren't quite perfect
class TestJobQueue:
result = 0
@ -364,6 +363,17 @@ class TestJobQueue:
scheduled_time = job_queue.jobs()[0].next_t.timestamp()
assert scheduled_time == pytest.approx(expected_time)
async def test_time_unit_dt_aware_time(self, job_queue, timezone):
# Testing running at a specific tz-aware time today
delta, now = 0.5, dtm.datetime.now(timezone)
expected_time = now + dtm.timedelta(seconds=delta)
when = expected_time.timetz()
expected_time = expected_time.timestamp()
job_queue.run_once(self.job_datetime_tests, when)
await asyncio.sleep(0.6)
assert self.job_time == pytest.approx(expected_time)
async def test_run_daily(self, job_queue):
delta, now = 1, dtm.datetime.now(UTC)
time_of_day = (now + dtm.timedelta(seconds=delta)).time()