Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Addlast keyword argument torun_repeating#1497

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to ourterms of service andprivacy statement. We’ll occasionally send you account related emails.

Already on GitHub?Sign in to your account

Closed
Show file tree
Hide file tree
Changes fromall commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletionsAUTHORS.rst
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -63,6 +63,7 @@ The following wonderful people contributed directly or indirectly to this projec
- `Oleg Sushchenko <https://github.com/feuillemorte>`_
- `Or Bin <https://github.com/OrBin>`_
- `overquota <https://github.com/overquota>`_
- `Paolo Lammens <https://github.com/plammens>`_
- `Patrick Hofmann <https://github.com/PH89>`_
- `Paul Larsen <https://github.com/PaulSonOfLars>`_
- `Pieter Schutz <https://github.com/eldinnie>`_
Expand Down
36 changes: 33 additions & 3 deletionstelegram/ext/jobqueue.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -89,7 +89,7 @@ def _put(self, job, time_spec=None, previous_t=None):
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``.
Defaults to now + ``job.interval``. # TODO: look into this
previous_t (optional):
Time at which the job last ran (``None`` if it hasn't run yet).

Expand DownExpand Up@@ -144,7 +144,7 @@ def run_once(self, callback, when, context=None, name=None):
self._put(job, time_spec=when)
return job

def run_repeating(self, callback, interval, first=None, context=None, name=None):
def run_repeating(self, callback, interval, first=None,last=None,context=None, name=None):
"""Creates a new ``Job`` that runs at specified intervals and adds it to the queue.

Args:
Expand All@@ -171,6 +171,11 @@ def run_repeating(self, callback, interval, first=None, context=None, name=None)
tomorrow.

Defaults to ``interval``
last (:obj:`int` | :obj:`float` | :obj:`datetime.timedelta` | \
:obj:`datetime.datetime` | :obj:`datetime.time`, optional):
Time after which the job should stop running.
This parameter will be interpreted depending on its type (same as for ``first``).
If ``None``, the job will run indefinitely.
context (:obj:`object`, optional): Additional data needed for the callback function.
Can be accessed through ``job.context`` in the callback. Defaults to ``None``.
name (:obj:`str`, optional): The name of the new job. Defaults to
Expand All@@ -189,6 +194,7 @@ def run_repeating(self, callback, interval, first=None, context=None, name=None)
job = Job(callback,
interval=interval,
repeat=True,
finish_time=last,
context=context,
name=name,
job_queue=self)
Expand DownExpand Up@@ -268,6 +274,9 @@ def tick(self):
self._set_next_peek(t)
break

delay_buffer = 0.01 # tolerance for last
if job.finish_time is not None and now > job.finish_time + delay_buffer:
job.schedule_removal() # job shouldn't run anymore
if job.removed:
self.logger.debug('Removing job %s', job.name)
continue
Expand All@@ -288,7 +297,7 @@ def tick(self):
if job.repeat and not job.removed:
self._put(job, previous_t=t)
else:
self.logger.debug('Dropping non-repeatingorremoved job %s', job.name)
self.logger.debug('Dropping non-repeating, removed,orfinished job %s', job.name)

def start(self):
"""Starts the job_queue thread."""
Expand DownExpand Up@@ -367,6 +376,10 @@ class Job(object):
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``.
finish_time (:obj:`int` | :obj:`float` | :obj:`datetime.timedelta` | \
:obj:`datetime.datetime` | :obj:`datetime.time`, optional):
Time after which the job shouldn't run anymore. Only valid if :attr:`repeat` is
``True``. Defaults to ``None``.
context (:obj:`object`, optional): Additional data needed for the callback function. Can be
accessed through ``job.context`` in the callback. Defaults to ``None``.
name (:obj:`str`, optional): The name of the new job. Defaults to ``callback.__name__``.
Expand All@@ -383,6 +396,7 @@ def __init__(self,
callback,
interval=None,
repeat=True,
finish_time=None,
context=None,
days=Days.EVERY_DAY,
name=None,
Expand All@@ -395,8 +409,10 @@ def __init__(self,

self._repeat = None
self._interval = None
self._finish_time = None
self.interval = interval
self.repeat = repeat
self.finish_time = finish_time

self._days = None
self.days = days
Expand DownExpand Up@@ -480,6 +496,20 @@ def repeat(self, repeat):
raise ValueError("'repeat' can not be set to 'True' when no 'interval' is set")
self._repeat = repeat

@property
def finish_time(self):
""":obj:`float`: Optional. Time at which the job should stop repeating."""
return self._finish_time

@finish_time.setter
def finish_time(self, time_spec):
if time_spec is not None:
if not self.repeat:
raise ValueError("'finish_time' cannot be set if job doesn't repeat")
self._finish_time = to_float_timestamp(time_spec)
else:
self._finish_time = None

@property
def days(self):
"""Tuple[:obj:`int`]: Optional. Defines on which days of the week the job should run."""
Expand Down
70 changes: 52 additions & 18 deletionstests/test_helpers.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015-2018
Expand All@@ -16,8 +17,8 @@
#
# 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 time

import pytest

Expand All@@ -29,6 +30,12 @@
from telegram.utils.helpers import _UtcOffsetTimezone, _datetime_to_float_timestamp


# TODO: move time test utils to tests.utils.time
# TODO: unify pytest.approx() calls when using now time

# constants
DAY_TOTAL_SECONDS = dtm.timedelta(days=1).total_seconds()

# 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()]
Expand All@@ -38,6 +45,18 @@
RELATIVE_TIME_SPECS = DELTA_TIME_SPECS + TIME_OF_DAY_TIME_SPECS
TIME_SPECS = ABSOLUTE_TIME_SPECS + RELATIVE_TIME_SPECS

# (naive UTC datetime, timestamp) input / expected pairs
NAIVE_DATETIME_FLOAT_TIMESTAMP_PAIRS = [
(dtm.datetime(2019, 11, 11, 0, 26, 16, 10**5), 1573431976.1),
(dtm.datetime(1970, 1, 1), 0.0),
(dtm.datetime(1970, 1, 1, microsecond=1), 1e-6)]


def microsecond_precision(x):
"""Utility to make equality assertions up to microsecond precision
(when floating point arithmetic means we don't have any guarantee beyond that)"""
return pytest.approx(x, abs=1e-6)


class TestHelpers(object):
def test_escape_markdown(self):
Expand All@@ -46,20 +65,24 @@ def test_escape_markdown(self):

assert expected_str == helpers.escape_markdown(test_str)

def test_to_float_timestamp_absolute_naive(self):
@pytest.mark.parametrize(['datetime', 'expected_float_timestamp'],
NAIVE_DATETIME_FLOAT_TIMESTAMP_PAIRS, ids=str)
def test_to_float_timestamp_absolute_naive(self, datetime, expected_float_timestamp):
"""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
assert helpers.to_float_timestamp(datetime) == expected_float_timestamp

def test_to_float_timestamp_absolute_aware(self, timezone):
@pytest.mark.parametrize('datetime_float_timestamp_pair',
NAIVE_DATETIME_FLOAT_TIMESTAMP_PAIRS, ids=str)
def test_to_float_timestamp_absolute_aware(self, datetime_float_timestamp_pair, 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())
datetime, float_timestamp = datetime_float_timestamp_pair
aware_datetime = datetime.replace(tzinfo=timezone)
expected = float_timestamp - timezone.utcoffset(None).total_seconds()
assert helpers.to_float_timestamp(aware_datetime) == microsecond_precision(expected)

def test_to_float_timestamp_absolute_no_reference(self):
"""A reference timestamp is only relevant for relative time specifications"""
Expand All@@ -71,17 +94,28 @@ 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
assert (helpers.to_float_timestamp(time_spec, reference_t)
== microsecond_precision(reference_t + delta))

def test_to_float_timestamp_time_of_day(self):
@pytest.mark.parametrize('delta', [dtm.timedelta(hours=1),
dtm.timedelta(microseconds=1)],
ids=lambda d: 'delta={}'.format(d))
def test_to_float_timestamp_time_of_day(self, delta):
"""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)
hour = 12
ref_datetime = dtm.datetime(1970, 1, 1, hour=hour)
ref_t = _datetime_to_float_timestamp(ref_datetime)
# make sure that ref_datetime ± delta falls within the same day
delta = min(delta, dtm.timedelta(hours=24 - hour), dtm.timedelta(hours=hour))
delta_seconds = delta.total_seconds()

# test for a time of day that is still to come, and one in the past (same day)
time_future, time_past = (ref_datetime + delta).time(), (ref_datetime - delta).time()
# pytest.approx(..., abs=1e-6): check up to microseconds (allow floating point arithmetic)
assert (helpers.to_float_timestamp(time_future, ref_t)
== microsecond_precision(ref_t + delta_seconds))
assert (helpers.to_float_timestamp(time_past, ref_t)
== microsecond_precision(ref_t - delta_seconds + DAY_TOTAL_SECONDS))

def test_to_float_timestamp_time_of_day_timezone(self, timezone):
"""Conversion from timezone-aware time-of-day specification to timestamp"""
Expand All@@ -95,7 +129,7 @@ def test_to_float_timestamp_time_of_day_timezone(self, timezone):
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.approx(ref_t + (-utc_offset.total_seconds() %DAY_TOTAL_SECONDS)))

@pytest.mark.parametrize('time_spec', RELATIVE_TIME_SPECS, ids=str)
def test_to_float_timestamp_default_reference(self, time_spec):
Expand Down
7 changes: 7 additions & 0 deletionstests/test_jobqueue.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -126,6 +126,13 @@ def test_run_repeating_first_timezone(self, job_queue, timezone):
sleep(0.001)
assert self.result == 1

def test_run_repeating_last(self, job_queue):
job_queue.run_repeating(self.job_run_once, 0.05, last=0.1)
sleep(0.12)
assert self.result == 2
sleep(0.1)
assert self.result == 2

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)
Expand Down
[8]ページ先頭
©2009-2025 Movatter.jp