"""
def __init__(
- self, callback: Callable[[], None], callback_time: float, jitter: float = 0
+ self,
+ callback: Callable[[], None],
+ callback_time: Union[datetime.timedelta, float],
+ jitter: float = 0,
) -> None:
self.callback = callback
- if callback_time <= 0:
- raise ValueError("Periodic callback must have a positive callback_time")
- self.callback_time = callback_time
+ if isinstance(callback_time, datetime.timedelta):
+ self.callback_time = callback_time / datetime.timedelta(milliseconds=1)
+ else:
+ if callback_time <= 0:
+ raise ValueError("Periodic callback must have a positive callback_time")
+ self.callback_time = callback_time
self.jitter = jitter
self._running = False
self._timeout = None # type: object
with mock.patch("random.random", mock_random):
self.assertEqual(self.simulate_calls(pc, call_durations), expected)
+ def test_timedelta(self):
+ pc = PeriodicCallback(lambda: None, datetime.timedelta(minutes=1, seconds=23))
+ expected_callback_time = 83000
+ self.assertEqual(pc.callback_time, expected_callback_time)
+
class TestIOLoopConfiguration(unittest.TestCase):
def run_python(self, *statements):