Check If Timer.cancel Is Called In Unit Test
I'm using the threading.Timer package to execute a method after x seconds. However, in some cases I want to execute this method earlier and cancel the timer (so it isn't called twi
Solution 1:
With timer.is_alive()
you are just checking if the timer-thread itself is alive, so if you want to "check if timer.cancel()
was called", you're testing for the wrong thing.
Does this mean that the cancel method does not stop the run action?
It does not stop the run()
-function, right. timer.cancel()
just sets a flag in an Event
-object which gets checked by run
. You can test if the flag is set with:
self.assertTrue(x.timer.finished.is_set())
Unfortunately, checking for cancellation is not enough to prevent repeated execution, since run
can have already crossed the check like you can see in the source code:
# threading.py (Python 3.7.1):
class Timer(Thread):
"""Call a function after a specified number of seconds:
t = Timer(30.0, f, args=None, kwargs=None)
t.start()
t.cancel() # stop the timer's action if it's still waiting
"""
def __init__(self, interval, function, args=None, kwargs=None):
Thread.__init__(self)
self.interval = interval
self.function = function
self.args = args if args is not None else []
self.kwargs = kwargs if kwargs is not None else {}
self.finished = Event()
def cancel(self):
"""Stop the timer if it hasn't finished yet."""
self.finished.set()
def run(self):
self.finished.wait(self.interval)
if not self.finished.is_set():
self.function(*self.args, **self.kwargs)
self.finished.set()
Some more effort is needed to ensure unique execution. I've written up a possible solution to this in my answer here.
Post a Comment for "Check If Timer.cancel Is Called In Unit Test"