test_tutorial_06.py

"""教程 06 验收脚本：完全离线，不需要 API Key 或监控后端。"""

from __future__ import annotations

import importlib.util
import io
import sys
import tempfile
from pathlib import Path


if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")


ROOT = Path(__file__).parent
EXAMPLES = ROOT / "examples" / "06-evaluation"
sys.path.insert(0, str(EXAMPLES))

import evaluation_core as core
import evaluation_metrics as metrics
import observability


def load_example(filename: str):
    path = EXAMPLES / filename
    module_name = f"tutorial06_{path.stem}"
    spec = importlib.util.spec_from_file_location(module_name, path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"无法加载示例：{path}")
    module = importlib.util.module_from_spec(spec)
    sys.modules[module_name] = module
    spec.loader.exec_module(module)
    return module


def run_test(name: str, test_func) -> bool:
    print(f"\n【{name}】")
    print("-" * 60)
    try:
        test_func()
        print(f"✓ {name}通过")
        return True
    except Exception as exc:
        print(f"✗ {name}失败：{exc}")
        return False


dataset = load_example("01_dataset_and_slices.py")
judge = load_example("03_llm_as_judge.py")
monitoring = load_example("04_tracing_and_monitoring.py")
gate = load_example("05_regression_gate.py")


def test_dataset_schema_and_roundtrip() -> None:
    cases = dataset.validate_dataset(core.SUPPORT_CASES)
    assert len(cases) == 5
    assert len(dataset.slice_by_tag(cases, "security")) == 1
    assert dataset.tag_distribution(cases)["happy_path"] == 3

    with tempfile.TemporaryDirectory() as directory:
        path = Path(directory) / "eval.jsonl"
        dataset.write_jsonl(path, cases)
        loaded = dataset.read_jsonl(path)
    assert loaded == cases
    print("✓ Schema、标签切片和 UTF-8 JSONL 往返正确")


def test_dataset_rejects_duplicates() -> None:
    duplicate = (core.SUPPORT_CASES[0], core.SUPPORT_CASES[0])
    try:
        dataset.validate_dataset(duplicate)
    except ValueError as exc:
        assert "重复" in str(exc)
    else:
        raise AssertionError("重复 case_id 未被拒绝")
    print("✓ 重复 case_id 会在评估前失败")


def test_deterministic_metrics() -> None:
    report = metrics.evaluate_dataset(core.SUPPORT_CASES, core.RuleBasedSupportAgent())
    aggregate = report.aggregate()
    assert aggregate["task_success"] == 1.0
    assert aggregate["citation_recall"] == 1.0
    assert not report.failures()
    print("✓ 健康版本的任务、答案和引用指标全部通过")


def test_metrics_locate_regressions() -> None:
    agent = core.RuleBasedSupportAgent(regressions={"refund_policy", "vpn_citation"})
    report = metrics.evaluate_dataset(core.SUPPORT_CASES, agent)
    aggregate = report.aggregate()
    failed_ids = {item.case_id for item in report.failures()}
    assert aggregate["task_success"] < 1.0
    assert aggregate["citation_recall"] < 1.0
    assert {"refund-policy-001", "vpn-guide-001"} <= failed_ids
    print("✓ 聚合指标下降可下钻到退款和 VPN 样例")


def test_judge_protocol() -> None:
    parsed = judge.parse_judge_result('{"score": 4, "reason": "正确但不够完整"}')
    assert parsed.score == 4
    assert parsed.normalized_score == 0.75

    invalid_outputs = (
        "not-json",
        '{"score": 6, "reason": "越界"}',
        '{"score": 4, "reason": "ok", "extra": true}',
    )
    for raw in invalid_outputs:
        try:
            judge.parse_judge_result(raw)
        except ValueError:
            pass
        else:
            raise AssertionError(f"非法裁判输出未被拒绝：{raw}")
    print("✓ 裁判结果严格校验 JSON、字段和分数范围")


def test_judge_calibration() -> None:
    case = core.SUPPORT_CASES[0]
    fake_judge = judge.FakeJudge()
    healthy = core.RuleBasedSupportAgent().invoke(case.question)
    broken = core.RuleBasedSupportAgent(regressions={"refund_policy"}).invoke(case.question)
    predictions = [
        judge.judge_case(case, healthy, fake_judge).score,
        judge.judge_case(case, broken, fake_judge).score,
    ]
    assert predictions == [5, 1]
    assert judge.mean_absolute_error(predictions, [5, 1]) == 0.0
    print("✓ Fake 裁判可重复，并能与人工标签计算偏差")


def test_trace_hierarchy_and_redaction() -> None:
    collector = observability.TraceCollector()
    result = monitoring.run_observed(
        core.RuleBasedSupportAgent(), core.SUPPORT_CASES[0].question, collector
    )
    assert result.intent == "refund"
    root, route, render = collector.spans
    assert root.status == route.status == render.status == "ok"
    assert route.trace_id == root.trace_id == render.trace_id
    assert route.parent_span_id == root.span_id
    assert root.attributes["prompt"] == "[REDACTED]"
    assert root.attributes["user_email"] == "[REDACTED_EMAIL]"
    print("✓ 父子 Span 共享 trace_id，Prompt 和邮箱均已脱敏")


def test_trace_error_path() -> None:
    collector = observability.TraceCollector()
    try:
        with collector.span("failing_step", {"token": "secret"}):
            raise RuntimeError("模拟失败")
    except RuntimeError:
        pass
    else:
        raise AssertionError("异常未向调用方传播")
    span = collector.spans[0]
    assert span.status == "error"
    assert "RuntimeError" in (span.error or "")
    assert span.attributes["token"] == "[REDACTED]"
    print("✓ 异常向外传播，同时记录错误状态和脱敏属性")


def test_metrics_window_and_alerts() -> None:
    window = observability.MetricsWindow()
    agent = core.RuleBasedSupportAgent()
    for case in core.SUPPORT_CASES:
        window.observe(agent.invoke(case.question), quality_score=1.0)
    summary = window.summary()
    assert summary["request_count"] == 5.0
    assert summary["success_rate"] == 1.0
    assert summary["p95_latency_ms"] >= summary["p50_latency_ms"]
    assert summary["total_tokens"] > 0
    assert summary["estimated_cost_usd"] > 0

    quality_rule = observability.AlertRule("quality_score", "lt", 1.01, "warning")
    assert quality_rule.evaluate(summary)
    print("✓ 请求量、成功率、P50/P95、Token、成本与质量告警正确")


def test_regression_gate() -> None:
    baseline = core.RuleBasedSupportAgent(version="baseline")
    healthy = core.RuleBasedSupportAgent(version="healthy")
    broken = core.RuleBasedSupportAgent(
        version="broken", regressions={"refund_policy", "vpn_citation"}
    )
    healthy_gate = gate.run_gate(core.SUPPORT_CASES, baseline, healthy)
    broken_gate = gate.run_gate(core.SUPPORT_CASES, baseline, broken)
    assert healthy_gate.passed
    assert not broken_gate.passed
    assert any("citation_recall" in reason for reason in broken_gate.reasons)
    assert any("policy" in reason for reason in broken_gate.reasons)
    print("✓ 健康候选放行，质量下降和关键切片失败会阻断回归版本")


if __name__ == "__main__":
    print("=" * 80)
    print("教程 06 代码验证")
    print("=" * 80)

    tests = [
        ("评估数据集", test_dataset_schema_and_roundtrip),
        ("重复样例校验", test_dataset_rejects_duplicates),
        ("确定性指标", test_deterministic_metrics),
        ("回归定位", test_metrics_locate_regressions),
        ("裁判协议", test_judge_protocol),
        ("裁判校准", test_judge_calibration),
        ("Trace 层级与脱敏", test_trace_hierarchy_and_redaction),
        ("Trace 异常路径", test_trace_error_path),
        ("指标窗口与告警", test_metrics_window_and_alerts),
        ("发布回归门禁", test_regression_gate),
    ]
    results = [(name, run_test(name, func)) for name, func in tests]

    print("\n" + "=" * 80)
    print("测试总结")
    print("=" * 80)
    for name, passed in results:
        print(f"{'✅ PASS' if passed else '❌ FAIL'} - {name}")
    if not all(passed for _, passed in results):
        sys.exit(1)
    print("\n🎉 所有测试通过！教程 06 代码验证完成。")