AirPlan-V2/test_p1_21_phase2.py

#!/usr/bin/env python3
"""3.2.17 中途变更处理（git驱动）Phase 2 功能测试
测试：ChangeClassifier + ImpactPropagator + 差异化失效 + BOUNDARY验证 + SAFE接口
"""

import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent / "lib"))


def test_change_classifier_implementation():
    """ChangeClassifier: 实现细节变更 → IMPLEMENTATION"""
    from air_runtime.change_classifier import ChangeClassifier, BlastRadius

    classifier = ChangeClassifier()
    old = "使用 ffmpeg 实现视频解码。\n\nimplementation detail: 可以用硬件加速 internally。"
    new = "使用 ffmpeg 实现视频解码。\n\n改用 gstreamer 硬件加速，optional 可用 private 实现。"

    result = classifier.classify("ADR-0005", old, new)
    assert result.blast_radius == BlastRadius.IMPLEMENTATION, \
        f"应为 IMPLEMENTATION，实际: {result.blast_radius}"
    assert result.adr_id == "ADR-0005"
    print("✓ IMPLEMENTATION 分类测试通过")
    return True


def test_change_classifier_interface():
    """ChangeClassifier: 接口变更 → INTERFACE"""
    from air_runtime.change_classifier import ChangeClassifier, BlastRadius

    classifier = ChangeClassifier()
    old = "使用 ffmpeg API 进行解码。"
    new = "使用 gstreamer API 进行解码。must implement 新的 contract 接口。caller 需要更新。"

    result = classifier.classify("ADR-0005", old, new)
    assert result.blast_radius == BlastRadius.INTERFACE, \
        f"应为 INTERFACE，实际: {result.blast_radius}"
    print("✓ INTERFACE 分类测试通过")
    return True


def test_change_classifier_global():
    """ChangeClassifier: 全局约束变更 → GLOBAL_CONSTRAINT"""
    from air_runtime.change_classifier import ChangeClassifier, BlastRadius

    classifier = ChangeClassifier()
    old = "使用 ffmpeg。"
    new = "coding standard: must 使用 RAII。all modules shall 遵循新规范。every module 需要更新。"

    result = classifier.classify("ADR-0005", old, new)
    assert result.blast_radius == BlastRadius.GLOBAL_CONSTRAINT, \
        f"应为 GLOBAL_CONSTRAINT，实际: {result.blast_radius}"
    print("✓ GLOBAL_CONSTRAINT 分类测试通过")
    return True


def test_change_classifier_new_adr():
    """ChangeClassifier: 新增 ADR → IMPLEMENTATION"""
    from air_runtime.change_classifier import ChangeClassifier, BlastRadius

    classifier = ChangeClassifier()
    result = classifier.classify("ADR-0020", None, "新的 ADR 内容。")
    assert result.blast_radius == BlastRadius.IMPLEMENTATION
    assert "new ADR" in result.reason
    print("✓ 新增 ADR 分类测试通过")
    return True


def test_change_classifier_deleted_adr():
    """ChangeClassifier: 删除 ADR → GLOBAL_CONSTRAINT"""
    from air_runtime.change_classifier import ChangeClassifier, BlastRadius

    classifier = ChangeClassifier()
    result = classifier.classify("ADR-0005", "旧的 ADR 内容。", None)
    assert result.blast_radius == BlastRadius.GLOBAL_CONSTRAINT
    assert "deleted" in result.reason.lower()
    print("✓ 删除 ADR 分类测试通过")
    return True


def test_change_classifier_conservative_fallback():
    """ChangeClassifier: 无法确定 → INTERFACE（保守）"""
    from air_runtime.change_classifier import ChangeClassifier, BlastRadius

    classifier = ChangeClassifier()
    result = classifier.classify("ADR-0005", "旧内容", "新内容")
    assert result.blast_radius == BlastRadius.INTERFACE
    assert "conservative" in result.reason.lower()
    print("✓ 保守回退分类测试通过")
    return True


def test_impact_propagator_implementation():
    """ImpactPropagator: IMPLEMENTATION → 仅直接引用受影响"""
    from air_runtime.task_graph import TaskGraph, TaskNode, Edge
    from air_runtime.change_classifier import BlastRadius
    from air_runtime.impact_propagator import ImpactPropagator, ImpactLabel

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", task="ffmpeg 解码器", adr_refs=["ADR-0005"]))
    graph.add_node(TaskNode(id="T-002", task="集成测试"))
    graph.add_node(TaskNode(id="T-003", task="UI", adr_refs=["ADR-0008"]))
    graph.add_edge(Edge(source="T-001", target="T-002"))

    propagator = ImpactPropagator()
    result = propagator.propagate(graph, "ADR-0005", BlastRadius.IMPLEMENTATION)

    assert "T-001" in result.impacted
    assert "T-002" not in result.impacted  # 不传播到下游
    assert "T-003" in result.safe
    # 验证 meta 标记
    assert graph.nodes["T-001"].meta["impact_label"] == ImpactLabel.IMPACTED.value
    assert graph.nodes["T-002"].meta["impact_label"] == ImpactLabel.SAFE.value
    print("✓ IMPLEMENTATION 传播测试通过")
    return True


def test_impact_propagator_interface():
    """ImpactPropagator: INTERFACE → 直接引用 IMPACTED + 下游1跳 BOUNDARY"""
    from air_runtime.task_graph import TaskGraph, TaskNode, Edge
    from air_runtime.change_classifier import BlastRadius
    from air_runtime.impact_propagator import ImpactPropagator, ImpactLabel

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", task="ffmpeg 解码器", adr_refs=["ADR-0005"]))
    graph.add_node(TaskNode(id="T-002", task="集成测试"))  # 下游1跳
    graph.add_node(TaskNode(id="T-003", task="部署"))  # 下游2跳
    graph.add_node(TaskNode(id="T-004", task="UI", adr_refs=["ADR-0008"]))  # 无关
    graph.add_edge(Edge(source="T-001", target="T-002"))
    graph.add_edge(Edge(source="T-002", target="T-003"))

    propagator = ImpactPropagator()
    result = propagator.propagate(graph, "ADR-0005", BlastRadius.INTERFACE)

    assert "T-001" in result.impacted
    assert "T-002" in result.boundary, f"T-002 应为 BOUNDARY，实际 impacted={result.impacted}, boundary={result.boundary}"
    assert "T-003" in result.safe, f"T-003 应为 SAFE（下游2跳），实际: {result.safe}"
    assert "T-004" in result.safe
    assert graph.nodes["T-001"].meta["impact_label"] == ImpactLabel.IMPACTED.value
    assert graph.nodes["T-002"].meta["impact_label"] == ImpactLabel.BOUNDARY.value
    assert graph.nodes["T-003"].meta["impact_label"] == ImpactLabel.SAFE.value
    print("✓ INTERFACE 传播(含 BOUNDARY)测试通过")
    return True


def test_impact_propagator_global():
    """ImpactPropagator: GLOBAL_CONSTRAINT → 全图 IMPACTED"""
    from air_runtime.task_graph import TaskGraph, TaskNode
    from air_runtime.change_classifier import BlastRadius
    from air_runtime.impact_propagator import ImpactPropagator, ImpactLabel

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", task="任务1", adr_refs=["ADR-0005"]))
    graph.add_node(TaskNode(id="T-002", task="任务2"))
    graph.add_node(TaskNode(id="T-003", task="任务3", adr_refs=["ADR-0008"]))

    propagator = ImpactPropagator()
    result = propagator.propagate(graph, "ADR-0005", BlastRadius.GLOBAL_CONSTRAINT)

    assert len(result.impacted) == 3
    assert len(result.boundary) == 0
    assert len(result.safe) == 0
    for nid in ["T-001", "T-002", "T-003"]:
        assert graph.nodes[nid].meta["impact_label"] == ImpactLabel.IMPACTED.value
    print("✓ GLOBAL_CONSTRAINT 全图传播测试通过")
    return True


def test_impact_propagator_clear_labels():
    """ImpactPropagator: clear_labels 清除标记"""
    from air_runtime.task_graph import TaskGraph, TaskNode
    from air_runtime.change_classifier import BlastRadius
    from air_runtime.impact_propagator import ImpactPropagator

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", task="任务1", adr_refs=["ADR-0005"]))
    graph.add_node(TaskNode(id="T-002", task="任务2"))

    propagator = ImpactPropagator()
    propagator.propagate(graph, "ADR-0005", BlastRadius.GLOBAL_CONSTRAINT)
    assert "impact_label" in graph.nodes["T-001"].meta

    propagator.clear_labels(graph)
    assert "impact_label" not in graph.nodes["T-001"].meta
    assert "impact_label" not in graph.nodes["T-002"].meta
    print("✓ clear_labels 测试通过")
    return True


def test_differentiated_invalidation_impacted_done():
    """差异化失效: IMPACTED + DONE → INVALIDATED"""
    from air_runtime.task_graph import TaskGraph, TaskNode, PlanDelta
    from air_runtime.impact_propagator import ImpactLabel

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", status="DONE", task="ffmpeg 解码器", adr_refs=["ADR-0005"]))

    delta = PlanDelta()
    labels = {"T-001": ImpactLabel.IMPACTED.value}
    report = graph.invalidate_by_adr("ADR-0005", delta, impact_labels=labels)

    assert report.invalidated_completed == 1
    assert graph.nodes["T-001"].status == "INVALIDATED"
    assert "T-001" in delta.removed_tasks
    print("✓ IMPACTED+DONE→INVALIDATED 测试通过")
    return True


def test_differentiated_invalidation_boundary_done():
    """差异化失效: BOUNDARY + DONE → 保持 DONE，标记 verification_needed"""
    from air_runtime.task_graph import TaskGraph, TaskNode, PlanDelta
    from air_runtime.impact_propagator import ImpactLabel

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", status="DONE", task="集成测试", adr_refs=["ADR-0005"]))

    delta = PlanDelta()
    labels = {"T-001": ImpactLabel.BOUNDARY.value}
    report = graph.invalidate_by_adr("ADR-0005", delta, impact_labels=labels)

    assert graph.nodes["T-001"].status == "DONE"
    assert graph.nodes["T-001"].meta.get("verification_needed") == True
    assert "T-001" in report.boundary_task_ids
    assert "T-001" not in delta.removed_tasks
    print("✓ BOUNDARY+DONE→verification_needed 测试通过")
    return True


def test_differentiated_invalidation_safe_untouched():
    """差异化失效: SAFE + DONE → 完全不处理"""
    from air_runtime.task_graph import TaskGraph, TaskNode, PlanDelta
    from air_runtime.impact_propagator import ImpactLabel

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", status="DONE", task="UI 界面", adr_refs=["ADR-0008"]))

    delta = PlanDelta()
    labels = {"T-001": ImpactLabel.SAFE.value}
    report = graph.invalidate_by_adr("ADR-0005", delta, impact_labels=labels)

    assert graph.nodes["T-001"].status == "DONE"
    assert graph.nodes["T-001"].meta.get("verification_needed") != True
    assert "T-001" not in delta.removed_tasks
    assert report.invalidated_completed == 0
    print("✓ SAFE→不处理 测试通过")
    return True


def test_differentiated_invalidation_impacted_dispatched():
    """差异化失效: IMPACTED + DISPATCHED → INVALIDATED"""
    from air_runtime.task_graph import TaskGraph, TaskNode, PlanDelta
    from air_runtime.impact_propagator import ImpactLabel

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", status="DISPATCHED", task="ffmpeg 流媒体", adr_refs=["ADR-0005"]))

    delta = PlanDelta()
    labels = {"T-001": ImpactLabel.IMPACTED.value}
    report = graph.invalidate_by_adr("ADR-0005", delta, impact_labels=labels)

    assert report.terminated_in_progress == 1
    assert graph.nodes["T-001"].status == "INVALIDATED"
    print("✓ IMPACTED+DISPATCHED→INVALIDATED 测试通过")
    return True


def test_backward_compat_uniform_invalidation():
    """向后兼容: impact_labels=None 时使用旧统一失效行为"""
    from air_runtime.task_graph import TaskGraph, TaskNode, Edge, PlanDelta

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", status="DONE", task="任务1", adr_refs=["ADR-0005"]))
    graph.add_node(TaskNode(id="T-002", status="DONE", task="任务2", adr_refs=["ADR-0005"]))
    graph.add_node(TaskNode(id="T-003", status="TODO", task="下游任务"))
    graph.add_edge(Edge(source="T-001", target="T-003"))

    delta = PlanDelta()
    report = graph.invalidate_by_adr("ADR-0005", delta)  # 不传 impact_labels

    assert report.invalidated_completed == 2
    assert report.cascaded_downstream == 1
    assert graph.nodes["T-001"].status == "INVALIDATED"
    assert graph.nodes["T-002"].status == "INVALIDATED"
    assert graph.nodes["T-003"].status == "INVALIDATED"
    print("✓ 向后兼容统一失效测试通过")
    return True


def test_partial_replanner_with_constraints():
    """局部重规划: replan_with_constraints 包含冻结接口"""
    from air_runtime.task_graph import TaskGraph, TaskNode
    from air_runtime.partial_replanner import PartialReplanner, Interface

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", status="INVALIDATED", task="失效任务", adr_refs=["ADR-0005"]))
    graph.add_node(TaskNode(id="T-002", status="DONE", task="安全任务", write_set=["src/safe.cpp"], adr_refs=["ADR-0008"]))

    replanner = PartialReplanner()
    frozen = [Interface(task_id="T-002", write_set=["src/safe.cpp"], adr_refs=["ADR-0008"])]
    delta = replanner.replan_with_constraints(graph, ["T-001"], frozen)

    request = delta.replan_request
    assert request["type"] == "partial-replan"
    # 稳定接口应包含 T-002
    stable_ids = {s["task_id"] for s in request["stableInterfaces"]}
    assert "T-002" in stable_ids
    print("✓ replan_with_constraints 测试通过")
    return True


def test_generate_verification_tasks():
    """局部重规划: generate_verification_tasks 为 BOUNDARY 生成验证任务"""
    from air_runtime.task_graph import TaskGraph, TaskNode
    from air_runtime.partial_replanner import PartialReplanner

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", status="DONE", task="集成测试",
                            files_dirs="tests/integration.cpp", write_set=["tests/integration.cpp"]))
    graph.add_node(TaskNode(id="T-002", status="TODO", task="未完成任务"))

    replanner = PartialReplanner()
    delta = replanner.generate_verification_tasks(graph, ["T-001", "T-002"])

    # T-001 DONE → 生成验证任务；T-002 TODO → 不生成
    assert len(delta.added_tasks) == 1
    vfy = delta.added_tasks[0]
    assert vfy.id == "T-001-VFY"
    assert vfy.status == "TODO"
    assert "边界验证" in vfy.task
    assert vfy.test_required == True

    # 验证边
    assert len(delta.edge_changes.added) == 1
    assert delta.edge_changes.added[0].source == "T-001"
    assert delta.edge_changes.added[0].target == "T-001-VFY"
    print("✓ generate_verification_tasks 测试通过")
    return True


def test_extract_safe_interfaces():
    """eng_mode: _extract_safe_interfaces 提取 SAFE+DONE 任务接口"""
    from air_runtime.task_graph import TaskGraph, TaskNode
    from air_runtime.impact_propagator import ImpactPropagator, ImpactLabel
    from air_runtime.change_classifier import BlastRadius
    from air_runtime.modes.eng_mode import _extract_safe_interfaces

    graph = TaskGraph()
    graph.add_node(TaskNode(id="T-001", status="DONE", task="安全任务",
                            write_set=["src/safe.cpp"], adr_refs=["ADR-0008"]))
    graph.add_node(TaskNode(id="T-002", status="DONE", task="受影响任务",
                            write_set=["src/broken.cpp"], adr_refs=["ADR-0005"]))
    graph.add_node(TaskNode(id="T-003", status="TODO", task="安全但未开始"))

    # 先传播标记
    propagator = ImpactPropagator()
    propagator.propagate(graph, "ADR-0005", BlastRadius.INTERFACE)

    interfaces = _extract_safe_interfaces(graph)
    safe_ids = {i.task_id for i in interfaces}
    assert "T-001" in safe_ids, f"T-001 是 SAFE+DONE 应有接口: {safe_ids}"
    assert "T-002" not in safe_ids, "T-002 是 IMPACTED 不应有接口"
    assert "T-003" not in safe_ids, "T-003 是 TODO 不应有接口"
    print("✓ _extract_safe_interfaces 测试通过")
    return True


def test_adr_watcher_content_for_classification():
    """ADRWatcher: get_content_for_classification 返回旧/新内容"""
    from air_runtime.adr_watcher import ADRWatcher, ADRChange

    with tempfile.TemporaryDirectory() as tmpdir:
        adr_dir = Path(tmpdir)
        adr_file = adr_dir / "ADR-0005-ffmpeg.md"
        adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n使用 ffmpeg。")

        watcher = ADRWatcher(adr_dir)
        watcher.snapshot()

        # 修改
        adr_file.write_text("# ADR-0005\n\nStatus: superseded\n\n改用 gstreamer。")
        changes = watcher.detect_changes()
        assert len(changes) == 1

        old_content, new_content = watcher.get_content_for_classification(changes[0])
        assert old_content is not None
        assert "ffmpeg" in old_content
        assert new_content is not None
        assert "gstreamer" in new_content

        print("✓ get_content_for_classification 测试通过")
    return True


def test_full_differentiated_flow():
    """完整差异化流程：分类→传播→失效→验证→安全接口"""
    from air_runtime.adr_watcher import ADRWatcher
    from air_runtime.task_graph import TaskGraph, TaskNode, Edge, PlanDelta
    from air_runtime.change_classifier import ChangeClassifier
    from air_runtime.impact_propagator import ImpactPropagator, ImpactLabel
    from air_runtime.partial_replanner import PartialReplanner
    from air_runtime.modes.eng_mode import _extract_safe_interfaces

    with tempfile.TemporaryDirectory() as tmpdir:
        adr_dir = Path(tmpdir) / "adr"
        adr_dir.mkdir(parents=True)
        adr_file = adr_dir / "ADR-0005-ffmpeg.md"
        adr_file.write_text("# ADR-0005\n\nStatus: accepted\n\n使用 ffmpeg API 实现视频解码。")

        watcher = ADRWatcher(adr_dir)
        watcher.snapshot()

        # 修改 ADR: 接口变更
        adr_file.write_text(
            "# ADR-0005\n\nStatus: modified\n\n"
            "改用 gstreamer API。must implement 新接口。caller 需要更新签名。"
        )
        changes = watcher.detect_changes()
        change = changes[0]

        # Phase 1: 分类
        old_c, new_c = watcher.get_content_for_classification(change)
        classifier = ChangeClassifier()
        classification = classifier.classify(change.adr_id, old_c, new_c)
        assert classification.blast_radius.value == "interface"

        # Phase 1: 传播
        graph = TaskGraph()
        graph.add_node(TaskNode(id="T-001", status="DONE", task="ffmpeg 解码器",
                                write_set=["src/decoder.cpp"], adr_refs=["ADR-0005"]))
        graph.add_node(TaskNode(id="T-002", status="DONE", task="集成测试"))
        graph.add_node(TaskNode(id="T-003", status="DONE", task="UI 界面",
                                write_set=["src/ui.cpp"], adr_refs=["ADR-0008"]))
        graph.add_node(TaskNode(id="T-004", status="TODO", task="部署"))
        graph.add_edge(Edge(source="T-001", target="T-002"))
        graph.add_edge(Edge(source="T-002", target="T-004"))

        propagator = ImpactPropagator()
        prop_result = propagator.propagate(graph, "ADR-0005", classification.blast_radius)
        assert "T-001" in prop_result.impacted
        assert "T-002" in prop_result.boundary
        assert "T-003" in prop_result.safe
        assert "T-004" in prop_result.safe  # 下游2跳

        # Phase 1: 差异化失效
        impact_labels = {}
        for nid in prop_result.impacted:
            impact_labels[nid] = ImpactLabel.IMPACTED.value
        for nid in prop_result.boundary:
            impact_labels[nid] = ImpactLabel.BOUNDARY.value
        for nid in prop_result.safe:
            impact_labels[nid] = ImpactLabel.SAFE.value

        delta = PlanDelta()
        report = graph.invalidate_by_adr("ADR-0005", delta, impact_labels=impact_labels)

        # T-001 IMPACTED+DONE → INVALIDATED
        assert graph.nodes["T-001"].status == "INVALIDATED"
        # T-002 BOUNDARY+DONE → verification_needed
        assert graph.nodes["T-002"].status == "DONE"
        assert graph.nodes["T-002"].meta.get("verification_needed") == True
        # T-003 SAFE → untouched
        assert graph.nodes["T-003"].status == "DONE"
        # T-004 SAFE TODO → untouched
        assert graph.nodes["T-004"].status == "TODO"

        # Phase 2: 生成验证任务
        replanner = PartialReplanner()
        vfy_delta = replanner.generate_verification_tasks(graph, report.boundary_task_ids)
        assert len(vfy_delta.added_tasks) == 1
        assert vfy_delta.added_tasks[0].id == "T-002-VFY"

        # Phase 3: 提取 SAFE 接口
        safe_ifaces = _extract_safe_interfaces(graph)
        safe_ids = {i.task_id for i in safe_ifaces}
        assert "T-003" in safe_ids
        assert "T-001" not in safe_ids

        print("✓ 完整差异化流程测试通过")
    return True


def test_cascade_report_extended_fields():
    """CascadeReport: 3.2.17 扩展字段 boundary_task_ids, blast_radius"""
    from air_runtime.task_graph import CascadeReport

    report = CascadeReport(
        invalidated_completed=1,
        terminated_in_progress=0,
        cascaded_downstream=0,
        rollback_ref="ref-001",
        invalidated_task_ids=["T-001"],
        boundary_task_ids=["T-002"],
        blast_radius="interface",
    )
    assert report.boundary_task_ids == ["T-002"]
    assert report.blast_radius == "interface"

    # 默认值
    report2 = CascadeReport()
    assert report2.boundary_task_ids == []
    assert report2.blast_radius == ""
    print("✓ CascadeReport 扩展字段测试通过")
    return True


def main():
    print("=" * 50)
    print("3.2.17 中途变更处理 Phase 2 功能测试")
    print("=" * 50)

    tests = [
        ("IMPLEMENTATION 分类", test_change_classifier_implementation),
        ("INTERFACE 分类", test_change_classifier_interface),
        ("GLOBAL_CONSTRAINT 分类", test_change_classifier_global),
        ("新增 ADR 分类", test_change_classifier_new_adr),
        ("删除 ADR 分类", test_change_classifier_deleted_adr),
        ("保守回退分类", test_change_classifier_conservative_fallback),
        ("IMPLEMENTATION 传播", test_impact_propagator_implementation),
        ("INTERFACE 传播+BOUNDARY", test_impact_propagator_interface),
        ("GLOBAL 全图传播", test_impact_propagator_global),
        ("clear_labels", test_impact_propagator_clear_labels),
        ("IMPACTED+DONE→INVALIDATED", test_differentiated_invalidation_impacted_done),
        ("BOUNDARY+DONE→verification", test_differentiated_invalidation_boundary_done),
        ("SAFE→不处理", test_differentiated_invalidation_safe_untouched),
        ("IMPACTED+DISPATCHED→INVALIDATED", test_differentiated_invalidation_impacted_dispatched),
        ("向后兼容统一失效", test_backward_compat_uniform_invalidation),
        ("replan_with_constraints", test_partial_replanner_with_constraints),
        ("generate_verification_tasks", test_generate_verification_tasks),
        ("_extract_safe_interfaces", test_extract_safe_interfaces),
        ("get_content_for_classification", test_adr_watcher_content_for_classification),
        ("完整差异化流程", test_full_differentiated_flow),
        ("CascadeReport 扩展字段", test_cascade_report_extended_fields),
    ]

    passed = 0
    failed = 0

    for name, test_fn in tests:
        try:
            test_fn()
            passed += 1
        except Exception as e:
            print(f"✗ {name} 失败: {e}")
            import traceback
            traceback.print_exc()
            failed += 1

    print("=" * 50)
    print(f"测试结果: {passed} 通过, {failed} 失败")
    print("=" * 50)

    return failed == 0


if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)