问题描述

我使用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方法的文档说明如下；

停止计时器，并取消执行计时器的操作。这仅在计时器仍处于等待阶段时才起作用。

推荐答案

使用timer.is_alive()您只是检查计时器线程本身是否处于活动状态，因此，如果您要检查timer.cancel()是否被调用，则您测试的是错误的内容。

这是否意味着Cancel方法不会停止运行操作？

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中列出了此问题的可能解决方案。

这篇关于检查单元测试中是否调用了Timer.Cancel的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持编程学习网！