检查单元测试中是否调用了Timer.Cancel
问题描述
我使用threading.Timer
包在x秒后执行一个方法。然而,在某些情况下,我希望更早地执行此方法并取消计时器(这样就不会调用两次)。如何对此进行单元测试?
我想知道计时器是否已停止,以便不再调用该方法。我现在使用以下代码,不幸的是is_alive still
返回True
from threading import Timer
Class X():
def __init__(self, timeout):
self.timer = Timer(timeout, self.some_method)
self.timer.start()
def some_method(self):
# Do something
def other_method(self):
self.timer.cancel()
self.some_method()
import unittest
Class TestX(unittest.TestCase):
def test_cancel_timer(self):
x = X(1000)
x.other_method()
self.assertFalse(x.timer.is_alive())
在run
操作期间,is_alive
方法返回True;
返回线程是否处于活动状态。 此方法在run()方法开始之前返回True,直到run()方法终止。模块函数的作用是:返回所有活动线程的列表。
有关cancel
方法的文档说明如下;
停止计时器,并取消执行计时器的操作。这仅在计时器仍处于等待阶段时才起作用。这是否意味着
cancel
方法不会停止run
操作?或者在Run方法之后仍处于灰色区域并因此返回True?
解决方案
使用timer.is_alive()
您只是检查计时器线程本身是否处于活动状态,因此,如果您要检查timer.cancel()
是否被调用,则您测试的是错误的内容。
它不会停止这是否意味着Cancel方法不会停止运行操作?
run()
-函数,对吧。timer.cancel()
只是在Event
对象中设置一个标志,由run
检查。您可以使用:测试是否设置了该标志
self.assertTrue(x.timer.finished.is_set())
遗憾的是,检查取消不足以防止重复执行,因为run
可能已经通过检查,如您在源代码中看到的:
# 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()
需要付出更多努力才能确保执行的唯一性。我已在我的答案here中列出了此问题的可能解决方案。
相关文章