Fix UTC/local inconsistencies for naive datetimes (#1506)

This commit is contained in:
Paolo Lammens 2019-11-15 20:51:22 +00:00 committed by Noam Meltzer
parent 10c9ec2313
commit 4e717a172b
9 changed files with 357 additions and 98 deletions

View file

@ -29,6 +29,7 @@ from threading import Thread, Lock, Event
from telegram.ext.callbackcontext import CallbackContext
from telegram.utils.deprecate import TelegramDeprecationWarning
from telegram.utils.helpers import to_float_timestamp, _UTC
class Days(object):
@ -78,30 +79,29 @@ class JobQueue(object):
"""
self._dispatcher = dispatcher
def _put(self, job, next_t=None, last_t=None):
if next_t is None:
next_t = job.interval
if next_t is None:
raise ValueError('next_t is None')
def _put(self, job, time_spec=None, previous_t=None):
"""
Enqueues the job, scheduling its next run at the correct time.
if isinstance(next_t, datetime.datetime):
next_t = (next_t - datetime.datetime.now()).total_seconds()
Args:
job (telegram.ext.Job): job to enqueue
time_spec (optional):
Specification of the time for which the job should be scheduled. The precise
semantics of this parameter depend on its type (see
:func:`telegram.ext.JobQueue.run_repeating` for details).
Defaults to now + ``job.interval``.
previous_t (optional):
Time at which the job last ran (``None`` if it hasn't run yet).
elif isinstance(next_t, datetime.time):
next_datetime = datetime.datetime.combine(datetime.date.today(), next_t)
if datetime.datetime.now().time() > next_t:
next_datetime += datetime.timedelta(days=1)
next_t = (next_datetime - datetime.datetime.now()).total_seconds()
elif isinstance(next_t, datetime.timedelta):
next_t = next_t.total_seconds()
next_t += last_t or time.time()
self.logger.debug('Putting job %s with t=%f', job.name, next_t)
"""
# get time at which to run:
time_spec = time_spec or job.interval
if time_spec is None:
raise ValueError("no time specification given for scheduling non-repeating job")
next_t = to_float_timestamp(time_spec, reference_timestamp=previous_t)
# enqueue:
self.logger.debug('Putting job %s with t=%f', job.name, time_spec)
self._queue.put((next_t, job))
# Wake up the loop if this job should be executed next
@ -141,7 +141,7 @@ class JobQueue(object):
"""
job = Job(callback, repeat=False, context=context, name=name, job_queue=self)
self._put(job, next_t=when)
self._put(job, time_spec=when)
return job
def run_repeating(self, callback, interval, first=None, context=None, name=None):
@ -192,7 +192,7 @@ class JobQueue(object):
context=context,
name=name,
job_queue=self)
self._put(job, next_t=first)
self._put(job, time_spec=first)
return job
def run_daily(self, callback, time, days=Days.EVERY_DAY, context=None, name=None):
@ -203,7 +203,8 @@ class JobQueue(object):
job. It should take ``bot, job`` as parameters, where ``job`` is the
:class:`telegram.ext.Job` instance. It can be used to access its ``Job.context``
or change it to a repeating job.
time (:obj:`datetime.time`): Time of day at which the job should run.
time (:obj:`datetime.time`): Time of day at which the job should run. If the timezone
(``time.tzinfo``) is ``None``, UTC will be assumed.
days (Tuple[:obj:`int`], optional): Defines on which days of the week the job should
run. Defaults to ``EVERY_DAY``
context (:obj:`object`, optional): Additional data needed for the callback function.
@ -225,10 +226,11 @@ class JobQueue(object):
interval=datetime.timedelta(days=1),
repeat=True,
days=days,
tzinfo=time.tzinfo,
context=context,
name=name,
job_queue=self)
self._put(job, next_t=time)
self._put(job, time_spec=time)
return job
def _set_next_peek(self, t):
@ -272,7 +274,7 @@ class JobQueue(object):
if job.enabled:
try:
current_week_day = datetime.datetime.now().weekday()
current_week_day = datetime.datetime.now(job.tzinfo).date().weekday()
if any(day == current_week_day for day in job.days):
self.logger.debug('Running job %s', job.name)
job.run(self._dispatcher)
@ -284,7 +286,7 @@ class JobQueue(object):
self.logger.debug('Skipping disabled job %s', job.name)
if job.repeat and not job.removed:
self._put(job, last_t=t)
self._put(job, previous_t=t)
else:
self.logger.debug('Dropping non-repeating or removed job %s', job.name)
@ -358,10 +360,11 @@ class Job(object):
It should take ``bot, job`` as parameters, where ``job`` is the
:class:`telegram.ext.Job` instance. It can be used to access it's :attr:`context`
or change it to a repeating job.
interval (:obj:`int` | :obj:`float` | :obj:`datetime.timedelta`, optional): The interval in
which the job will run. If it is an :obj:`int` or a :obj:`float`, it will be
interpreted as seconds. If you don't set this value, you must set :attr:`repeat` to
``False`` and specify :attr:`next_t` when you put the job into the job queue.
interval (:obj:`int` | :obj:`float` | :obj:`datetime.timedelta`, optional): The time
interval between executions of the job. If it is an :obj:`int` or a :obj:`float`,
it will be interpreted as seconds. If you don't set this value, you must set
:attr:`repeat` to ``False`` and specify :attr:`time_spec` when you put the job into
the job queue.
repeat (:obj:`bool`, optional): If this job should be periodically execute its callback
function (``True``) or only once (``False``). Defaults to ``True``.
context (:obj:`object`, optional): Additional data needed for the callback function. Can be
@ -371,7 +374,9 @@ class Job(object):
Defaults to ``Days.EVERY_DAY``
job_queue (:class:`telegram.ext.JobQueue`, optional): The ``JobQueue`` this job belongs to.
Only optional for backward compatibility with ``JobQueue.put()``.
tzinfo (:obj:`datetime.tzinfo`, optional): timezone associated to this job. Used when
checking the day of the week to determine whether a job should run (only relevant when
``days is not Days.EVERY_DAY``). Defaults to UTC.
"""
def __init__(self,
@ -381,19 +386,21 @@ class Job(object):
context=None,
days=Days.EVERY_DAY,
name=None,
job_queue=None):
job_queue=None,
tzinfo=_UTC):
self.callback = callback
self.context = context
self.name = name or callback.__name__
self._repeat = repeat
self._repeat = None
self._interval = None
self.interval = interval
self.repeat = repeat
self._days = None
self.days = days
self.tzinfo = tzinfo
self._job_queue = weakref.proxy(job_queue) if job_queue is not None else None

View file

@ -17,7 +17,12 @@
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""This module contains helper functions."""
import datetime as dtm # dtm = "DateTime Module"
import time
from collections import defaultdict
from numbers import Number
try:
import ujson as json
@ -27,7 +32,6 @@ from html import escape
import re
import signal
from datetime import datetime
# From https://stackoverflow.com/questions/2549939/get-signal-names-from-numbers-in-python
_signames = {v: k
@ -40,54 +44,154 @@ def get_signal_name(signum):
return _signames[signum]
# Not using future.backports.datetime here as datetime value might be an input from the user,
# making every isinstace() call more delicate. So we just use our own compat layer.
if hasattr(datetime, 'timestamp'):
# Python 3.3+
def _timestamp(dt_obj):
return dt_obj.timestamp()
else:
# Python < 3.3 (incl 2.7)
from time import mktime
def _timestamp(dt_obj):
return mktime(dt_obj.timetuple())
def escape_markdown(text):
"""Helper function to escape telegram markup symbols."""
escape_chars = '\*_`\['
return re.sub(r'([%s])' % escape_chars, r'\\\1', text)
def to_timestamp(dt_obj):
# -------- date/time related helpers --------
# TODO: add generic specification of UTC for naive datetimes to docs
if hasattr(dtm, 'timezone'):
# Python 3.3+
def _datetime_to_float_timestamp(dt_obj):
if dt_obj.tzinfo is None:
dt_obj = dt_obj.replace(tzinfo=_UTC)
return dt_obj.timestamp()
_UtcOffsetTimezone = dtm.timezone
_UTC = dtm.timezone.utc
else:
# Python < 3.3 (incl 2.7)
# hardcoded timezone class (`datetime.timezone` isn't available in py2)
class _UtcOffsetTimezone(dtm.tzinfo):
def __init__(self, offset):
self.offset = offset
def tzname(self, dt):
return 'UTC +{}'.format(self.offset)
def utcoffset(self, dt):
return self.offset
def dst(self, dt):
return dtm.timedelta(0)
_UTC = _UtcOffsetTimezone(dtm.timedelta(0))
__EPOCH_DT = dtm.datetime.fromtimestamp(0, tz=_UTC)
__NAIVE_EPOCH_DT = __EPOCH_DT.replace(tzinfo=None)
# _datetime_to_float_timestamp
# Not using future.backports.datetime here as datetime value might be an input from the user,
# making every isinstace() call more delicate. So we just use our own compat layer.
def _datetime_to_float_timestamp(dt_obj):
epoch_dt = __EPOCH_DT if dt_obj.tzinfo is not None else __NAIVE_EPOCH_DT
return (dt_obj - epoch_dt).total_seconds()
_datetime_to_float_timestamp.__doc__ = \
"""Converts a datetime object to a float timestamp (with sub-second precision).
If the datetime object is timezone-naive, it is assumed to be in UTC."""
def to_float_timestamp(t, reference_timestamp=None):
"""
Converts a given time object to a float POSIX timestamp.
Used to convert different time specifications to a common format. The time object
can be relative (i.e. indicate a time increment, or a time of day) or absolute.
Any objects from the :module:`datetime` module that are timezone-naive will be assumed
to be in UTC.
``None`` s are left alone (i.e. ``to_float_timestamp(None)`` is ``None``).
Args:
dt_obj (:class:`datetime.datetime`):
t (int | float | datetime.timedelta | datetime.datetime | datetime.time):
Time value to convert. The semantics of this parameter will depend on its type:
* :obj:`int` or :obj:`float` will be interpreted as "seconds from ``reference_t``"
* :obj:`datetime.timedelta` will be interpreted as
"time increment from ``reference_t``"
* :obj:`datetime.datetime` will be interpreted as an absolute date/time value
* :obj:`datetime.time` will be interpreted as a specific time of day
reference_timestamp (float, optional): POSIX timestamp that indicates the absolute time
from which relative calculations are to be performed (e.g. when ``t`` is given as an
:obj:`int`, indicating "seconds from ``reference_t``"). Defaults to now (the time at
which this function is called).
If ``t`` is given as an absolute representation of date & time (i.e. a
``datetime.datetime`` object), ``reference_timestamp`` is not relevant and so its
value should be ``None``. If this is not the case, a ``ValueError`` will be raised.
Returns:
int:
(float | None) The return value depends on the type of argument ``t``. If ``t`` is
given as a time increment (i.e. as a obj:`int`, :obj:`float` or
:obj:`datetime.timedelta`), then the return value will be ``reference_t`` + ``t``.
Else if it is given as an absolute date/time value (i.e. a :obj:`datetime.datetime`
object), the equivalent value as a POSIX timestamp will be returned.
Finally, if it is a time of the day without date (i.e. a :obj:`datetime.time`
object), the return value is the nearest future occurrence of that time of day.
Raises:
TypeError: if `t`'s type is not one of those described above
"""
if not dt_obj:
return None
return int(_timestamp(dt_obj))
if reference_timestamp is None:
reference_timestamp = time.time()
elif isinstance(t, dtm.datetime):
raise ValueError('t is an (absolute) datetime while reference_timestamp is not None')
if isinstance(t, dtm.timedelta):
return reference_timestamp + t.total_seconds()
elif isinstance(t, Number):
return reference_timestamp + t
elif isinstance(t, dtm.time):
if t.tzinfo is not None:
reference_dt = dtm.datetime.fromtimestamp(reference_timestamp, tz=t.tzinfo)
else:
reference_dt = dtm.datetime.utcfromtimestamp(reference_timestamp) # assume UTC
reference_date = reference_dt.date()
reference_time = reference_dt.timetz()
if reference_time > t: # if the time of day has passed today, use tomorrow
reference_date += dtm.timedelta(days=1)
return _datetime_to_float_timestamp(dtm.datetime.combine(reference_date, t))
elif isinstance(t, dtm.datetime):
return _datetime_to_float_timestamp(t)
raise TypeError('Unable to convert {} object to timestamp'.format(type(t).__name__))
def to_timestamp(dt_obj, reference_timestamp=None):
"""
Wrapper over :func:`to_float_timestamp` which returns an integer (the float value truncated
down to the nearest integer).
See the documentation for :func:`to_float_timestamp` for more details.
"""
return int(to_float_timestamp(dt_obj, reference_timestamp)) if dt_obj is not None else None
def from_timestamp(unixtime):
"""
Converts an (integer) unix timestamp to a naive datetime object in UTC.
``None`` s are left alone (i.e. ``from_timestamp(None)`` is ``None``).
Args:
unixtime (int):
unixtime (int): integer POSIX timestamp
Returns:
datetime.datetime:
equivalent :obj:`datetime.datetime` value in naive UTC if ``timestamp`` is not
``None``; else ``None``
"""
if not unixtime:
if unixtime is None:
return None
return datetime.utcfromtimestamp(unixtime)
return dtm.datetime.utcfromtimestamp(unixtime)
# -------- end --------
def mention_html(user_id, name):

View file

@ -27,9 +27,11 @@ from time import sleep
import pytest
from telegram import Bot, Message, User, Chat, MessageEntity, Update, \
InlineQuery, CallbackQuery, ShippingQuery, PreCheckoutQuery, ChosenInlineResult
from telegram import (Bot, Message, User, Chat, MessageEntity, Update,
InlineQuery, CallbackQuery, ShippingQuery, PreCheckoutQuery,
ChosenInlineResult)
from telegram.ext import Dispatcher, JobQueue, Updater, BaseFilter
from telegram.utils.helpers import _UtcOffsetTimezone
from tests.bots import get_bot
TRAVIS = os.getenv('TRAVIS', False)
@ -258,3 +260,13 @@ def get_false_update_fixture_decorator_params():
@pytest.fixture(scope='function', **get_false_update_fixture_decorator_params())
def false_update(request):
return Update(update_id=1, **request.param)
@pytest.fixture(params=[1, 2], ids=lambda h: 'UTC +{hour:0>2}:00'.format(hour=h))
def utc_offset(request):
return datetime.timedelta(hours=request.param)
@pytest.fixture()
def timezone(utc_offset):
return _UtcOffsetTimezone(utc_offset)

View file

@ -19,7 +19,7 @@
import os
import sys
import time
from datetime import datetime
import datetime as dtm
from platform import python_implementation
import pytest
@ -108,7 +108,7 @@ class TestBot(object):
assert message.text == message.text
assert message.forward_from.username == message.from_user.username
assert isinstance(message.forward_date, datetime)
assert isinstance(message.forward_date, dtm.datetime)
@flaky(3, 1)
@pytest.mark.timeout(10)
@ -615,7 +615,7 @@ class TestBot(object):
assert bot.restrict_chat_member(channel_id,
95205500,
chat_permissions,
until_date=datetime.now())
until_date=dtm.datetime.utcnow())
@flaky(3, 1)
@pytest.mark.timeout(10)

View file

@ -46,7 +46,7 @@ class TestChatMember(object):
assert chat_member.status == self.status
def test_de_json_all_args(self, bot, user):
time = datetime.datetime.now()
time = datetime.datetime.utcnow()
json_dict = {'user': user.to_dict(),
'status': self.status,
'until_date': to_timestamp(time),

View file

@ -27,7 +27,7 @@ import re
@pytest.fixture(scope='function')
def update():
return Update(0, Message(0, User(0, 'Testuser', False), datetime.datetime.now(),
return Update(0, Message(0, User(0, 'Testuser', False), datetime.datetime.utcnow(),
Chat(0, 'private')))
@ -138,7 +138,7 @@ class TestFilters(object):
assert isinstance(matches, list)
assert len(matches) == 2
assert all([type(res) == SRE_TYPE for res in matches])
update.message.forward_date = datetime.datetime.now()
update.message.forward_date = datetime.datetime.utcnow()
result = filter(update)
assert result
assert isinstance(result, dict)
@ -248,7 +248,7 @@ class TestFilters(object):
assert result
def test_filters_reply(self, update):
another_message = Message(1, User(1, 'TestOther', False), datetime.datetime.now(),
another_message = Message(1, User(1, 'TestOther', False), datetime.datetime.utcnow(),
Chat(0, 'private'))
update.message.text = 'test'
assert not Filters.reply(update)
@ -475,7 +475,7 @@ class TestFilters(object):
def test_filters_forwarded(self, update):
assert not Filters.forwarded(update)
update.message.forward_date = datetime.datetime.now()
update.message.forward_date = datetime.datetime.utcnow()
assert Filters.forwarded(update)
def test_filters_game(self, update):
@ -616,7 +616,7 @@ class TestFilters(object):
def test_and_filters(self, update):
update.message.text = 'test'
update.message.forward_date = datetime.datetime.now()
update.message.forward_date = datetime.datetime.utcnow()
assert (Filters.text & Filters.forwarded)(update)
update.message.text = '/test'
assert not (Filters.text & Filters.forwarded)(update)
@ -625,7 +625,7 @@ class TestFilters(object):
assert not (Filters.text & Filters.forwarded)(update)
update.message.text = 'test'
update.message.forward_date = datetime.datetime.now()
update.message.forward_date = datetime.datetime.utcnow()
assert (Filters.text & Filters.forwarded & Filters.private)(update)
def test_or_filters(self, update):
@ -640,7 +640,7 @@ class TestFilters(object):
def test_and_or_filters(self, update):
update.message.text = 'test'
update.message.forward_date = datetime.datetime.now()
update.message.forward_date = datetime.datetime.utcnow()
assert (Filters.text & (Filters.status_update | Filters.forwarded))(update)
update.message.forward_date = False
assert not (Filters.text & (Filters.forwarded | Filters.status_update))(update)

View file

@ -16,6 +16,9 @@
#
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
import time
import datetime as dtm
import pytest
from telegram import Sticker
@ -23,6 +26,17 @@ from telegram import Update
from telegram import User
from telegram.message import Message
from telegram.utils import helpers
from telegram.utils.helpers import _UtcOffsetTimezone, _datetime_to_float_timestamp
# sample time specification values categorised into absolute / delta / time-of-day
ABSOLUTE_TIME_SPECS = [dtm.datetime.now(tz=_UtcOffsetTimezone(dtm.timedelta(hours=-7))),
dtm.datetime.utcnow()]
DELTA_TIME_SPECS = [dtm.timedelta(hours=3, seconds=42, milliseconds=2), 30, 7.5]
TIME_OF_DAY_TIME_SPECS = [dtm.time(12, 42, tzinfo=_UtcOffsetTimezone(dtm.timedelta(hours=-7))),
dtm.time(12, 42)]
RELATIVE_TIME_SPECS = DELTA_TIME_SPECS + TIME_OF_DAY_TIME_SPECS
TIME_SPECS = ABSOLUTE_TIME_SPECS + RELATIVE_TIME_SPECS
class TestHelpers(object):
@ -32,6 +46,76 @@ class TestHelpers(object):
assert expected_str == helpers.escape_markdown(test_str)
def test_to_float_timestamp_absolute_naive(self):
"""Conversion from timezone-naive datetime to timestamp.
Naive datetimes should be assumed to be in UTC.
"""
datetime = dtm.datetime(2019, 11, 11, 0, 26, 16, 10**5)
assert helpers.to_float_timestamp(datetime) == 1573431976.1
def test_to_float_timestamp_absolute_aware(self, timezone):
"""Conversion from timezone-aware datetime to timestamp"""
# we're parametrizing this with two different UTC offsets to exclude the possibility
# of an xpass when the test is run in a timezone with the same UTC offset
datetime = dtm.datetime(2019, 11, 11, 0, 26, 16, 10**5, tzinfo=timezone)
assert (helpers.to_float_timestamp(datetime) ==
1573431976.1 - timezone.utcoffset(None).total_seconds())
def test_to_float_timestamp_absolute_no_reference(self):
"""A reference timestamp is only relevant for relative time specifications"""
with pytest.raises(ValueError):
helpers.to_float_timestamp(dtm.datetime(2019, 11, 11), reference_timestamp=123)
@pytest.mark.parametrize('time_spec', DELTA_TIME_SPECS, ids=str)
def test_to_float_timestamp_delta(self, time_spec):
"""Conversion from a 'delta' time specification to timestamp"""
reference_t = 0
delta = time_spec.total_seconds() if hasattr(time_spec, 'total_seconds') else time_spec
assert helpers.to_float_timestamp(time_spec, reference_t) == reference_t + delta
def test_to_float_timestamp_time_of_day(self):
"""Conversion from time-of-day specification to timestamp"""
hour, hour_delta = 12, 1
ref_t = _datetime_to_float_timestamp(dtm.datetime(1970, 1, 1, hour=hour))
# test for a time of day that is still to come, and one in the past
time_future, time_past = dtm.time(hour + hour_delta), dtm.time(hour - hour_delta)
assert helpers.to_float_timestamp(time_future, ref_t) == ref_t + 60 * 60 * hour_delta
assert helpers.to_float_timestamp(time_past, ref_t) == ref_t + 60 * 60 * (24 - hour_delta)
def test_to_float_timestamp_time_of_day_timezone(self, timezone):
"""Conversion from timezone-aware time-of-day specification to timestamp"""
# we're parametrizing this with two different UTC offsets to exclude the possibility
# of an xpass when the test is run in a timezone with the same UTC offset
utc_offset = timezone.utcoffset(None)
ref_datetime = dtm.datetime(1970, 1, 1, 12)
ref_t, time_of_day = _datetime_to_float_timestamp(ref_datetime), ref_datetime.time()
# first test that naive time is assumed to be utc:
assert helpers.to_float_timestamp(time_of_day, ref_t) == pytest.approx(ref_t)
# test that by setting the timezone the timestamp changes accordingly:
assert (helpers.to_float_timestamp(time_of_day.replace(tzinfo=timezone), ref_t) ==
pytest.approx(ref_t + (-utc_offset.total_seconds() % (24 * 60 * 60))))
@pytest.mark.parametrize('time_spec', RELATIVE_TIME_SPECS, ids=str)
def test_to_float_timestamp_default_reference(self, time_spec):
"""The reference timestamp for relative time specifications should default to now"""
now = time.time()
assert (helpers.to_float_timestamp(time_spec)
== pytest.approx(helpers.to_float_timestamp(time_spec, reference_timestamp=now)))
@pytest.mark.parametrize('time_spec', TIME_SPECS, ids=str)
def test_to_timestamp(self, time_spec):
# delegate tests to `to_float_timestamp`
assert helpers.to_timestamp(time_spec) == int(helpers.to_float_timestamp(time_spec))
def test_to_timestamp_none(self):
# this 'convenience' behaviour has been left left for backwards compatibility
assert helpers.to_timestamp(None) is None
def test_from_timestamp(self):
assert helpers.from_timestamp(1573431976) == dtm.datetime(2019, 11, 11, 0, 26, 16)
def test_create_deep_linked_url(self):
username = 'JamesTheMock'
@ -63,7 +147,6 @@ class TestHelpers(object):
helpers.create_deep_linked_url("abc", None)
def test_effective_message_type(self):
def build_test_message(**kwargs):
config = dict(
message_id=1,

View file

@ -16,7 +16,7 @@
#
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
import datetime
import datetime as dtm
import os
import sys
import time
@ -28,6 +28,7 @@ from flaky import flaky
from telegram.ext import JobQueue, Updater, Job, CallbackContext
from telegram.utils.deprecate import TelegramDeprecationWarning
from telegram.utils.helpers import _UtcOffsetTimezone
@pytest.fixture(scope='function')
@ -83,6 +84,24 @@ class TestJobQueue(object):
sleep(0.02)
assert self.result == 1
def test_run_once_timezone(self, job_queue, timezone):
"""Test the correct handling of aware datetimes.
Set the target datetime to utcnow + x hours (naive) with the timezone set to utc + x hours,
which is equivalent to now.
"""
# we're parametrizing this with two different UTC offsets to exclude the possibility
# of an xpass when the test is run in a timezone with the same UTC offset
when = (dtm.datetime.utcnow() + timezone.utcoffset(None)).replace(tzinfo=timezone)
job_queue.run_once(self.job_run_once, when)
sleep(0.001)
assert self.result == 1
def test_run_once_no_time_spec(self, job_queue):
# test that an appropiate exception is raised if a job is attempted to be scheduled
# without specifying a time
with pytest.raises(ValueError):
job_queue.run_once(self.job_run_once, when=None)
def test_job_with_context(self, job_queue):
job_queue.run_once(self.job_run_once_with_context, 0.01, context=5)
sleep(0.02)
@ -100,6 +119,13 @@ class TestJobQueue(object):
sleep(0.07)
assert self.result == 1
def test_run_repeating_first_timezone(self, job_queue, timezone):
"""Test correct scheduling of job when passing a timezone-aware datetime as ``first``"""
first = (dtm.datetime.utcnow() + timezone.utcoffset(None)).replace(tzinfo=timezone)
job_queue.run_repeating(self.job_run_once, 0.05, first=first)
sleep(0.001)
assert self.result == 1
def test_multiple(self, job_queue):
job_queue.run_once(self.job_run_once, 0.01)
job_queue.run_once(self.job_run_once, 0.02)
@ -183,7 +209,7 @@ class TestJobQueue(object):
def test_time_unit_dt_timedelta(self, job_queue):
# Testing seconds, minutes and hours as datetime.timedelta object
# This is sufficient to test that it actually works.
interval = datetime.timedelta(seconds=0.05)
interval = dtm.timedelta(seconds=0.05)
expected_time = time.time() + interval.total_seconds()
job_queue.run_once(self.job_datetime_tests, interval)
@ -192,43 +218,70 @@ class TestJobQueue(object):
def test_time_unit_dt_datetime(self, job_queue):
# Testing running at a specific datetime
delta = datetime.timedelta(seconds=0.05)
when = datetime.datetime.now() + delta
expected_time = time.time() + delta.total_seconds()
delta, now = dtm.timedelta(seconds=0.05), time.time()
when = dtm.datetime.utcfromtimestamp(now) + delta
expected_time = now + delta.total_seconds()
job_queue.run_once(self.job_datetime_tests, when)
sleep(0.06)
assert pytest.approx(self.job_time) == expected_time
assert self.job_time == pytest.approx(expected_time)
def test_time_unit_dt_time_today(self, job_queue):
# Testing running at a specific time today
delta = 0.05
when = (datetime.datetime.now() + datetime.timedelta(seconds=delta)).time()
expected_time = time.time() + delta
delta, now = 0.05, time.time()
when = (dtm.datetime.utcfromtimestamp(now) + dtm.timedelta(seconds=delta)).time()
expected_time = now + delta
job_queue.run_once(self.job_datetime_tests, when)
sleep(0.06)
assert pytest.approx(self.job_time) == expected_time
assert self.job_time == pytest.approx(expected_time)
def test_time_unit_dt_time_tomorrow(self, job_queue):
# Testing running at a specific time that has passed today. Since we can't wait a day, we
# test if the jobs next_t has been calculated correctly
delta = -2
when = (datetime.datetime.now() + datetime.timedelta(seconds=delta)).time()
expected_time = time.time() + delta + 60 * 60 * 24
# test if the job's next scheduled execution time has been calculated correctly
delta, now = -2, time.time()
when = (dtm.datetime.utcfromtimestamp(now) + dtm.timedelta(seconds=delta)).time()
expected_time = now + delta + 60 * 60 * 24
job_queue.run_once(self.job_datetime_tests, when)
assert pytest.approx(job_queue._queue.get(False)[0]) == expected_time
assert job_queue._queue.get(False)[0] == pytest.approx(expected_time)
def test_run_daily(self, job_queue):
delta = 0.5
time_of_day = (datetime.datetime.now() + datetime.timedelta(seconds=delta)).time()
expected_time = time.time() + 60 * 60 * 24 + delta
delta, now = 0.1, time.time()
time_of_day = (dtm.datetime.utcfromtimestamp(now) + dtm.timedelta(seconds=delta)).time()
expected_reschedule_time = now + delta + 24 * 60 * 60
job_queue.run_daily(self.job_run_once, time_of_day)
sleep(0.6)
sleep(0.2)
assert self.result == 1
assert pytest.approx(job_queue._queue.get(False)[0]) == expected_time
assert job_queue._queue.get(False)[0] == pytest.approx(expected_reschedule_time)
def test_run_daily_with_timezone(self, job_queue):
"""test that the weekday is retrieved based on the job's timezone
We set a job to run at the current UTC time of day (plus a small delay buffer) with a
timezone that is---approximately (see below)---UTC +24, and set it to run on the weekday
after the current UTC weekday. The job should therefore be executed now (because in UTC+24,
the time of day is the same as the current weekday is the one after the current UTC
weekday).
"""
now = time.time()
utcnow = dtm.datetime.utcfromtimestamp(now)
delta = 0.1
# must subtract one minute because the UTC offset has to be strictly less than 24h
# thus this test will xpass if run in the interval [00:00, 00:01) UTC time
# (because target time will be 23:59 UTC, so local and target weekday will be the same)
target_tzinfo = _UtcOffsetTimezone(dtm.timedelta(days=1, minutes=-1))
target_datetime = (utcnow + dtm.timedelta(days=1, minutes=-1, seconds=delta)).replace(
tzinfo=target_tzinfo)
target_time = target_datetime.timetz()
target_weekday = target_datetime.date().weekday()
expected_reschedule_time = now + delta + 24 * 60 * 60
job_queue.run_daily(self.job_run_once, time=target_time, days=(target_weekday,))
sleep(delta + 0.1)
assert self.result == 1
assert job_queue._queue.get(False)[0] == pytest.approx(expected_reschedule_time)
def test_warnings(self, job_queue):
j = Job(self.job_run_once, repeat=False)

View file

@ -35,12 +35,12 @@ def message(bot):
@pytest.fixture(scope='function',
params=[
{'forward_from': User(99, 'forward_user', False),
'forward_date': datetime.now()},
'forward_date': datetime.utcnow()},
{'forward_from_chat': Chat(-23, 'channel'),
'forward_from_message_id': 101,
'forward_date': datetime.now()},
'forward_date': datetime.utcnow()},
{'reply_to_message': Message(50, None, None, None)},
{'edit_date': datetime.now()},
{'edit_date': datetime.utcnow()},
{'text': 'a text message',
'enitites': [MessageEntity('bold', 10, 4),
MessageEntity('italic', 16, 7)]},
@ -114,7 +114,7 @@ def message_params(bot, request):
class TestMessage(object):
id = 1
from_user = User(2, 'testuser', False)
date = datetime.now()
date = datetime.utcnow()
chat = Chat(3, 'private')
test_entities = [{'length': 4, 'offset': 10, 'type': 'bold'},
{'length': 7, 'offset': 16, 'type': 'italic'},