如何在pytest测试运行时后台持续收集系统及应用指标？

1. 线程+Fixture：轻量的单测试级收集 #

用线程替代多进程，配合pytest的function级fixture，实现每个测试独立收集指标，逻辑简单且开销低。

实现代码 #

首先安装psutil（用于系统/进程指标收集）：

pip install psutil

在测试文件或conftest.py中定义fixture：

import pytest
import threading
import time
import psutil

def _collect_metrics(log_file, stop_signal):
    """后台指标收集函数"""
    with open(log_file, "a", encoding="utf-8") as f:
        while not stop_signal.is_set():
            # 收集系统级指标
            cpu_usage = psutil.cpu_percent(interval=1)
            mem_usage = psutil.virtual_memory().percent
            # 若需收集应用进程指标，可传入应用PID获取
            # app_proc = psutil.Process(app_pid)
            # app_cpu = app_proc.cpu_percent(interval=1)
            
            log_content = f"{time.strftime('%Y-%m-%d %H:%M:%S')} | CPU: {cpu_usage}% | 内存: {mem_usage}%\n"
            f.write(log_content)
            time.sleep(2)  # 调整收集间隔

@pytest.fixture(scope="function")
def metrics_collector(request):
    # 为每个测试生成独立日志文件
    log_path = f"test_metrics_{request.node.name}.log"
    stop_event = threading.Event()
    # 启动后台收集线程（设为守护线程避免阻塞测试退出）
    collector = threading.Thread(target=_collect_metrics, args=(log_path, stop_event), daemon=True)
    collector.start()
    
    yield  # 执行测试逻辑
    
    # 测试结束后停止收集
    stop_event.set()
    collector.join()

使用方式 #

在需要收集指标的测试函数中引入fixture即可：

def test_destructive_db_operation(metrics_collector):
    # 测试操作：如删除数据库数据、调用写接口
    # ...
    # 断言逻辑
    assert True

2. 钩子函数：全局测试会话级收集 #

如果需要为所有测试统一收集指标，无需逐个测试加fixture，可利用pytest的会话钩子实现全局收集。

实现代码（写入conftest.py） #

import threading
import time
import psutil
import pytest

_stop_event = None
_collector_thread = None

def _global_collector():
    """全局指标收集函数"""
    with open("global_test_metrics.log", "a", encoding="utf-8") as f:
        while not _stop_event.is_set():
            cpu = psutil.cpu_percent(interval=1)
            mem = psutil.virtual_memory().percent
            log_line = f"{time.strftime('%Y-%m-%d %H:%M:%S')} | CPU: {cpu}% | 内存: {mem}%\n"
            f.write(log_line)
            time.sleep(2)

@pytest.hookimpl(tryfirst=True)
def pytest_sessionstart(session):
    """测试会话启动时启动收集线程"""
    global _stop_event, _collector_thread
    _stop_event = threading.Event()
    _collector_thread = threading.Thread(target=_global_collector, daemon=True)
    _collector_thread.start()

@pytest.hookimpl(trylast=True)
def pytest_sessionfinish(session, exitstatus):
    """测试会话结束时停止收集"""
    global _stop_event, _collector_thread
    _stop_event.set()
    _collector_thread.join()

3. 为何不选subprocess或multiprocessing？ #

• subprocess：需要单独维护外部脚本，还要处理进程启动/停止、日志同步，复杂度远高于线程方案。
• multiprocessing：进程间数据共享、通信成本高，对于这种简单的定期收集场景，线程的轻量性更合适，无需额外开销。

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