Watchdog + Pytest + Allure ：文件变化自动触发测试及实时报告生成

import os
import time
import subprocess
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class TestAutoRunHandler(FileSystemEventHandler):
    def __init__(self, test_dir, report_dir="allure_reports", debounce_seconds=1):
        self.test_dir = os.path.abspath(test_dir)  # 监控的测试目录
        self.report_dir = report_dir  # allure报告输出目录
        self.debounce_seconds = debounce_seconds  # 防抖时间（避免短时间内多次触发）
        self.last_trigger_time = 0# 上次触发时间

    def _should_trigger(self):
        """防抖逻辑：两次触发间隔小于设定值则忽略"""
        current_time = time.time()
        if current_time - self.last_trigger_time < self.debounce_seconds:
            return False
        self.last_trigger_time = current_time
        return True

    def _run_tests(self):
        """执行pytest并生成allure报告"""
        print("\n" + "="*50)
        print(f"[{time.strftime('%H:%M:%S')}] 检测到文件变化，开始执行测试...")
        print("="*50)

        # 1. 清理旧报告（可选，根据需求决定是否保留历史）
        if os.path.exists(self.report_dir):
            subprocess.run(f"rm -rf {self.report_dir}", shell=True, check=True)

        # 2. 执行pytest，指定allure报告目录
        pytest_cmd = (
            f"pytest {self.test_dir} "
            f"--alluredir={self.report_dir} "
            "-v  # 显示详细测试日志"
        )
        try:
            # 运行pytest命令，输出实时打印到控制台
            subprocess.run(pytest_cmd, shell=True, check=True)
            print("\n测试执行完成，正在生成allure报告...")

            # 3. 自动打开allure报告（使用allure serve，会启动临时服务器）
            # 若希望生成静态文件，可替换为 `allure generate`
            allure_cmd = f"allure serve {self.report_dir} --port 8080"
            subprocess.Popen(allure_cmd, shell=True)  # 非阻塞运行，避免占用主线程
            print(f"报告已启动：http://localhost:8080（按 Ctrl+C 停止监控）")
        except subprocess.CalledProcessError as e:
            print(f"\n测试执行失败：{e}")

    def on_modified(self, event):
        """文件/目录被修改时触发"""
        # 过滤条件：仅监控测试目录下的.py文件（可根据需求调整）
        if (
            not event.is_directory 
            and event.src_path.startswith(self.test_dir) 
            and event.src_path.endswith(".py")
            andself._should_trigger()
        ):
            self._run_tests()

    def on_created(self, event):
        """新建文件/目录时触发（如新增测试用例）"""
        if (
            not event.is_directory 
            and event.src_path.startswith(self.test_dir) 
            and event.src_path.endswith(".py")
            andself._should_trigger()
        ):
            self._run_tests()

    def on_deleted(self, event):
        """删除文件/目录时触发（如删除测试用例）"""
        if (
            not event.is_directory 
            and event.src_path.startswith(self.test_dir) 
            and event.src_path.endswith(".py")
            andself._should_trigger()
        ):
            self._run_tests()

if __name__ == "__main__":
    # 配置：监控的测试目录（如./tests）、报告目录、防抖时间
    TEST_DIR = "./tests"# 替换为你的测试目录
    REPORT_DIR = "allure_reports"
    DEBOUNCE_SECONDS = 1# 1秒内多次修改只触发一次

    # 初始化处理器和观察者
    event_handler = TestAutoRunHandler(
        test_dir=TEST_DIR,
        report_dir=REPORT_DIR,
        debounce_seconds=DEBOUNCE_SECONDS
    )
    observer = Observer(daemon=True)
    # 递归监控测试目录（包括子目录）
    observer.schedule(event_handler, path=TEST_DIR, recursive=True)

    # 启动监控
    observer.start()
    print(f"开始监控目录：{TEST_DIR}")
    print(f"当 .py 文件发生修改/创建/删除时，将自动执行 pytest 并生成 allure 报告")
    print("按 Ctrl+C 停止监控...")

    # 保持主进程运行
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        print("\n监控已停止")
    observer.join()