如何测试让函数按固定周期无限执行的Python装饰器？

背景 #

我开发了一个run_periodically装饰器，可让被装饰的函数每隔指定秒数无限执行。

装饰器示例代码 #

@run_periodically(cycle_time=1) # 单位：秒
def action(exec_time):
    print(f"exec_time: {exec_time}")

执行输出效果 #

调用action后会持续输出执行时间，格式如下：

exec_time: 2023-02-28 12:01:09.075368
exec_time: 2023-02-28 12:01:10.075368
exec_time: 2023-02-28 12:01:11.075368
exec_time: 2023-02-28 12:01:12.075368
...

当前测试用例 #

我编写了如下测试用例，虽然能运行，但担心受系统时钟波动、负载影响导致不稳定：

from datetime import datetime, timedelta
import pytest

def prepare_action(stop_time: datetime): 
    counter = 0

    @run_periodically(cycle_time=0.001)
    def action(exec_time: datetime):
        nonlocal counter
        counter += 1
        if exec_time > stop_time:
            assert counter == 3
            exit(42)

    return action

def test_run_periodically():
    stop_time = datetime.utcnow() + timedelta(milliseconds=3)
    action = prepare_action(stop_time=stop_time)
    with pytest.raises(SystemExit) as pytest_wrapped_e:
        action(exec_time=datetime.utcnow() + timedelta(milliseconds=1))
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 42

稳定测试这类无限工作流的方法 #

1. 依赖注入可控的时间源 #

直接依赖系统时钟是不稳定的核心原因，给装饰器添加时间提供者的注入点，测试时用模拟时间完全控制时间流逝：

改造装饰器

from datetime import datetime
import time

def run_periodically(cycle_time, time_provider=datetime.utcnow):
    def decorator(func):
        def wrapper(*args, **kwargs):
            while True:
                current_time = time_provider()
                func(current_time, *args, **kwargs)
                time.sleep(cycle_time)
        return wrapper
    return decorator

模拟时间的测试用例

class MockTimeProvider:
    def __init__(self, start_time):
        self.current_time = start_time
    
    def advance(self, delta):
        self.current_time += delta
    
    def get(self):
        return self.current_time

def test_run_periodically_with_mock_time():
    start_time = datetime.utcnow()
    mock_time = MockTimeProvider(start_time)
    stop_time = start_time + timedelta(milliseconds=3)
    counter = 0

    @run_periodically(cycle_time=0.001, time_provider=mock_time.get)
    def action(exec_time):
        nonlocal counter
        counter += 1
        if exec_time > stop_time:
            assert counter == 3
            exit(42)
    
    # 手动推进时间触发执行
    mock_time.advance(timedelta(milliseconds=1))
    action()
    mock_time.advance(timedelta(milliseconds=1))
    mock_time.advance(timedelta(milliseconds=1))
    
    with pytest.raises(SystemExit) as excinfo:
        mock_time.advance(timedelta(milliseconds=1))
    
    assert excinfo.value.code == 42
    assert counter == 3

2. 给测试添加超时兜底 #

即使保留真实时间逻辑，也要给测试设置超时，避免因系统负载过高导致测试无限挂起。用pytest的timeout标记即可：

@pytest.mark.timeout(5) # 超时时间设为5秒，覆盖预期执行时间
def test_run_periodically_with_timeout():
    stop_time = datetime.utcnow() + timedelta(milliseconds=3)
    action = prepare_action(stop_time=stop_time)
    with pytest.raises(SystemExit) as pytest_wrapped_e:
        action(exec_time=datetime.utcnow() + timedelta(milliseconds=1))
    assert pytest_wrapped_e.type == SystemExit
    assert pytest_wrapped_e.value.code == 42

3. 用线程隔离执行，主动终止工作流 #

把周期性执行逻辑放到后台线程，测试时主动发送终止信号，避免用exit()影响测试进程：

改造装饰器支持终止

import threading
from datetime import datetime

def run_periodically(cycle_time):
    def decorator(func):
        stop_event = threading.Event()
        
        def periodic_task():
            while not stop_event.is_set():
                current_time = datetime.utcnow()
                func(current_time)
                stop_event.wait(cycle_time)
        
        def wrapper():
            thread = threading.Thread(target=periodic_task, daemon=True)
            thread.start()
            return stop_event
        
        return wrapper
    return decorator

线程版测试用例

def test_run_periodically_with_thread():
    stop_after = timedelta(milliseconds=4)
    counter = 0

    @run_periodically(cycle_time=0.001)
    def action(exec_time):
        nonlocal counter
        counter += 1
    
    # 启动任务并获取终止信号
    stop_event = action()
    # 等待足够时间让任务执行
    threading.Event().wait(stop_after.total_seconds())
    # 主动终止任务
    stop_event.set()
    
    # 允许±1的误差，兼容系统调度延迟
    assert 2 <= counter <= 4

4. 解耦逻辑，拆分测试 #

把业务逻辑和周期调度逻辑分离，分别测试：

• 单独测试业务函数的正确性，验证其输入输出是否符合预期；
• 单独测试装饰器的调度逻辑，用模拟时间验证是否按指定间隔调用函数。

内容的提问来源于stack exchange，提问作者Nikodem Bienia