chendandan_homework/scripts/run_regression.py

from __future__ import annotations

"""TravelMind自动回归测试入口：依次运行全部本地单元测试，可选运行调用真实模型和外部API的集成测试。"""

import argparse
import subprocess
import sys
from dataclasses import dataclass


@dataclass(frozen=True)
class TestCommand:
    name: str
    args: list[str]


UNIT_TESTS = [
    TestCommand(
        name="所有本地单元测试",
        args=[
            sys.executable,
            "-m",
            "pytest",
            "tests",
            "-v",
        ],
    ),
]


INTEGRATION_TESTS = [
    TestCommand(
        name="MCP工具注册",
        args=[
            sys.executable,
            "-m",
            "scripts.test_mcp_registry",
        ],
    ),
    TestCommand(
        name="需求解析Graph",
        args=[
            sys.executable,
            "-m",
            "scripts.test_requirement_graph",
        ],
    ),
    TestCommand(
        name="资源查询Graph",
        args=[
            sys.executable,
            "-m",
            "scripts.test_resource_graph",
        ],
    ),
    TestCommand(
        name="地图研究Graph",
        args=[
            sys.executable,
            "-m",
            "scripts.test_research_graph",
        ],
    ),
    TestCommand(
        name="候选筛选Graph",
        args=[
            sys.executable,
            "-m",
            "scripts.test_selection_graph",
        ],
    ),
    TestCommand(
        name="路线评估Graph",
        args=[
            sys.executable,
            "-m",
            "scripts.test_route_graph",
        ],
    ),
    TestCommand(
        name="最终多角色Graph",
        args=[
            sys.executable,
            "-m",
            "scripts.test_final_graph",
        ],
    ),
]


def run_command(
    command: TestCommand,
) -> bool:
    """以子进程方式执行单个TestCommand，输出分隔线与结果。"""
    separator = "=" * 72

    print()
    print(separator)
    print(f"开始：{command.name}")
    print(separator)

    completed = subprocess.run(
        command.args,
        check=False,
    )

    if completed.returncode == 0:
        print(f"通过：{command.name}")
        return True

    print(
        f"失败：{command.name}，"
        f"returncode={completed.returncode}"
    )

    return False


def parse_args() -> argparse.Namespace:
    """解析命令行参数：--integration启用集成测试，--continue-on-error在单项失败后继续。"""
    parser = argparse.ArgumentParser(
        description=(
            "运行TravelMind自动回归测试。"
        )
    )

    parser.add_argument(
        "--integration",
        action="store_true",
        help=(
            "同时运行会调用真实模型、SerpApi"
            "和高德MCP的集成测试。"
        ),
    )

    parser.add_argument(
        "--continue-on-error",
        action="store_true",
        help="单项失败后继续执行后续测试。",
    )

    return parser.parse_args()


def main() -> None:
    """依次执行全部测试命令，汇总失败项并以非零退出码结束。"""
    args = parse_args()

    commands = list(UNIT_TESTS)

    if args.integration:
        commands.extend(INTEGRATION_TESTS)

    failed_commands: list[str] = []

    for command in commands:
        passed = run_command(command)

        if passed:
            continue

        failed_commands.append(command.name)

        if not args.continue_on_error:
            break

    print()
    print("=" * 72)

    if failed_commands:
        print("回归测试未通过：")

        for name in failed_commands:
            print(f"- {name}")

        raise SystemExit(1)

    print("全部回归测试通过。")


if __name__ == "__main__":
    main()