Проверьте, вызван ли Timer.cancel в модульном тесте - PullRequest
Новая
       15

Проверьте, вызван ли Timer.cancel в модульном тесте

3 голосов
/ 04 июня 2019

Я использую пакет 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 возвращает True во время операции run;

Вернуть, жив ли поток. Этот метод возвращает True непосредственно перед тем, как метод run () запускается до тех пор, пока метод run () не завершится. Функция модуля enumerate () возвращает список всех живых потоков.

Документация по методу cancel гласит следующее:

Остановите таймер и отмените выполнение действия таймера. Это будет работать, только если таймер все еще находится на стадии ожидания.

Означает ли это, что метод cancel не останавливает действие run? Или все еще находится в серой области после метода run и возвращает True по этой причине?

1 Ответ

1 голос
/ 05 июня 2019

С timer.is_alive() вы просто проверяете, жив ли сам поток таймера, поэтому, если вы хотите «проверить, был ли вызван timer.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()

Требуются дополнительные усилия для обеспечения уникального выполнения.Я написал возможное решение для этого в своем ответе здесь .

...