检查是否在单元测试中调用了 Timer.cancel答案

【问题标题】：Check if Timer.cancel is called in unit test检查是否在单元测试中调用了 Timer.cancel
【发布时间】：2019-06-04 09:58:34
【问题描述】：

我正在使用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())

形成文档，is_alive 方法在run 操作期间返回 True；

返回线程是否存活。此方法在 run() 方法开始之前直到 run() 方法终止之后才返回 True。模块函数 enumerate() 返回所有活动线程的列表。

cancel 方法的文档说明如下；

停止定时器，并取消定时器动作的执行。这仅在计时器仍处于等待阶段时才有效。

这是否意味着cancel 方法不会停止run 操作？还是在run方法之后仍然处于灰色区域并因此返回True？

【问题讨论】：

取消操作需要一些时间（例如，应该释放计时器分配的锁等等）。如果您在x.other_method() 之后和检查x.timer.is_alive() 之前添加一些暂停（例如time.sleep(.1)），则测试应该通过。
嗯，好吧，这似乎可行。但是当取消方法运行时，该方法永远不会被调用？例如。定时器在取消操作执行之后和取消操作完成之前正好用完。
一旦触发取消，该动作将不再执行；但是，请记住，计时器中两个操作之间的时间间隔不能保证与您在 Timer 构造函数中传递的timeout 完全相同，因此肯定有可能发生冲突。

标签： python python-multithreading python-unittest

【解决方案1】：

使用timer.is_alive()，您只是在检查计时器线程本身是否处于活动状态，因此如果您想“检查是否调用了timer.cancel()”，那么您正在测试错误的东西。

这是否意味着取消方法不会停止运行动作？

它不会停止run()-函数，对。 timer.cancel() 只是在Event-object 中设置一个标志，由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 中写了一个可能的解决方案。

【讨论】：