run_all_tests的副本.py

#!/usr/bin/env python3
"""
PHM 测试套件 — 评委一键运行
============================
运行方式:
    python3 tests/run_all_tests.py

或分别运行:
    python3 tests/test_verifier.py    # 校验器单元测试
    python3 tests/test_orchestrator.py # 多Agent集成测试
    python3 tests/test_demo.py         # Demo功能测试
"""

import sys
import os
import time
import json
import importlib.util

# 确保导入路径（兼容两种目录结构：平铺仓库 / src子目录仓库）
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, SCRIPT_DIR)  # 平铺结构：模块与脚本同目录

# src/子目录结构探测
_alt_root = os.path.dirname(SCRIPT_DIR)
for _sub in ('src', 'agents'):
    _p = os.path.join(_alt_root, _sub)
    if os.path.isdir(_p):
        sys.path.insert(0, _p)

# PROJECT_ROOT指向demo.py所在层级（平铺=脚本目录；嵌套=父目录）
_demo_candidates = [
    os.path.join(SCRIPT_DIR, 'demo.py'),
    os.path.join(SCRIPT_DIR, 'src', 'demo.py'),
    os.path.join(_alt_root, 'src', 'demo.py'),
]
DEMO_PATH = next((p for p in _demo_candidates if os.path.exists(p)), None)
PROJECT_ROOT = os.path.dirname(DEMO_PATH) if DEMO_PATH and 'src' in DEMO_PATH else SCRIPT_DIR


def test_verifier():
    """测试1: 运行时校验器"""
    print("\n" + "=" * 60)
    print("测试1: 运行时校验器 (RuntimeVerifier)")
    print("=" * 60)
    
    from runtime_verifier import PHMRuntimeVerifier
    verifier = PHMRuntimeVerifier()
    
    # 测试1a: 正确响应
    gt = {"d_eff": 12.6693, "eta": 0.6521, "phase": "CORRECTABLE"}
    response_ok = "d_eff = 12.6693, eta = 0.6521, phase = CORRECTABLE"
    result = verifier.verify(response_ok, gt)
    assert result["verified"] == True, f"正确响应应通过: {result}"
    print(f"  1a 正确响应 → ✅ 通过")
    
    # 测试1b: 编造数字
    response_bad = "d_eff = 99.5, eta = 0.95, phase = UNCORRECTABLE"
    result = verifier.verify(response_bad, gt)
    assert result["verified"] == False, f"编造数字应拦截: {result}"
    assert len(result["violations"]) >= 2, f"应有≥2个违规: {result}"
    print(f"  1b 编造数字 → ✅ 拦截 ({len(result['violations'])}个违规)")
    
    # 测试1c: 推导数字
    response_derived = "d_eff = 12.6693, 两倍d_eff = 25.3386"
    result = verifier.verify(response_derived, gt)
    assert result["verified"] == True, f"推导数字应通过: {result}"
    print(f"  1c 推导数字 → ✅ 通过")
    
    # 测试1d: 相态错误
    response_phase = "d_eff = 12.6693, eta = 0.6521, phase = UNCORRECTABLE"
    result = verifier.verify(response_phase, gt)
    assert result["verified"] == False, f"相态错误应拦截: {result}"
    print(f"  1d 相态错误 → ✅ 拦截")
    
    print(f"  结果: 4/4 通过")
    return True


def test_orchestrator():
    """测试2: 多Agent编排"""
    print("\n" + "=" * 60)
    print("测试2: 多Agent编排 (Orchestrator)")
    print("=" * 60)
    
    import networkx as nx
    from orchestrator import PHMOrchestrator, build_demo_network
    
    # 测试2a: 正常运行
    G = build_demo_network()
    orch = PHMOrchestrator(mode='simulated')
    result = orch.run(G, query="测试查询")
    
    assert result["final_status"] == "verified", f"应返回verified: {result['final_status']}"
    assert len(result["agents"]) == 3, f"应有3个Agent: {len(result['agents'])}"
    assert result["verification"]["verified"] == True
    print(f"  2a 正常运行 → ✅ verified (3 agents, {result['total_elapsed_ms']:.1f}ms)")
    
    # 测试2b: 确定性（重复运行结果一致）
    result2 = orch.run(G, query="测试查询")
    d1 = result["agents"][0]["output"]["d_eff"]
    d2 = result2["agents"][0]["output"]["d_eff"]
    assert d1 == d2, f"确定性失败: {d1} != {d2}"
    print(f"  2b 确定性 → ✅ d_eff={d1} (两次一致)")
    
    # 测试2c: 执行日志
    assert len(orch.execution_log) >= 2, f"应有≥2条日志: {len(orch.execution_log)}"
    for log in orch.execution_log:
        assert "run_id" in log
        assert "timestamp" in log
        assert "agents" in log
        assert "final_status" in log
    print(f"  2c 执行日志 → ✅ {len(orch.execution_log)}条记录")
    
    # 测试2d: 上下文链
    for log in orch.execution_log:
        for link in log["context_chain"]:
            assert "step" in link
            assert "from" in link
            assert "to" in link
            assert "format" in link
    print(f"  2d 上下文链 → ✅ JSON传递完整")
    
    print(f"  结果: 4/4 通过")
    return True


def test_demo():
    """测试3: Demo功能"""
    print("\n" + "=" * 60)
    print("测试3: Demo功能 (HSECMiniEngine)")
    print("=" * 60)
    
    import networkx as nx
    import numpy as np
    
    # 导入demo中的引擎（自动探测路径）
    if DEMO_PATH is None:
        raise FileNotFoundError("demo.py not found (searched: script dir, src/, parent/src/)")
    spec = importlib.util.spec_from_file_location("demo", DEMO_PATH)
    demo = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(demo)
    
    # 测试3a: 引擎分析
    G = demo.build_test_network()
    engine = demo.HSECMiniEngine(d_0=12.0)
    result = engine.analyze(G)
    
    assert "phase_transition" in result
    assert "hidden_risks" in result
    assert "network_stats" in result
    
    pt = result["phase_transition"]
    assert "d_eff_current" in pt
    assert "eta" in pt
    assert "phase" in pt
    print(f"  3a 引擎分析 → ✅ d_eff={pt['d_eff_current']}, phase={pt['phase']}")
    
    # 测试3b: 网络构建
    assert G.number_of_nodes() > 20, f"网络应>20节点: {G.number_of_nodes()}"
    assert G.number_of_edges() > 30, f"网络应>30边: {G.number_of_edges()}"
    print(f"  3b 网络构建 → ✅ {G.number_of_nodes()}节点/{G.number_of_edges()}边")
    
    # 测试3c: 隐蔽风险检测
    assert len(result["hidden_risks"]) > 0, "应有隐蔽风险节点"
    print(f"  3c 隐蔽风险 → ✅ {len(result['hidden_risks'])}个节点")
    
    # 测试3d: 校验器
    verifier = demo.MiniVerifier(pt)
    bare = verifier.verify("d_eff = 99.5, eta = 0.95")
    assert len(bare["ungrounded"]) >= 1, f"应有ungrounded: {bare}"
    print(f"  3d 校验器 → ✅ 检出{len(bare['ungrounded'])}个编造数字")
    
    print(f"  结果: 4/4 通过")
    return True


def main():
    print("╔" + "═" * 58 + "╗")
    print("║  PHM 测试套件 — 评委一键运行" + " " * 27 + "║")
    print("╚" + "═" * 58 + "╝")
    
    start = time.time()
    results = []
    
    tests = [
        ("校验器", test_verifier),
        ("多Agent编排", test_orchestrator),
        ("Demo功能", test_demo),
    ]
    
    for name, test_fn in tests:
        try:
            test_fn()
            results.append((name, "PASS"))
        except Exception as e:
            print(f"  ❌ 失败: {e}")
            results.append((name, f"FAIL: {e}"))
    
    elapsed = time.time() - start
    
    print("\n" + "=" * 60)
    print("测试结果汇总")
    print("=" * 60)
    all_pass = True
    for name, status in results:
        icon = "✅" if status == "PASS" else "❌"
        print(f"  {icon} {name}: {status}")
        if status != "PASS":
            all_pass = False
    
    print(f"\n  总耗时: {elapsed:.2f}s")
    print(f"  总结果: {'全部通过 ✅' if all_pass else '存在失败 ❌'}")
    print("=" * 60)
    
    return 0 if all_pass else 1


if __name__ == "__main__":
    sys.exit(main())