zhifu-platform/tests/test_evaluation_validation.py

import hashlib
import json
from datetime import UTC, datetime
from pathlib import Path
from uuid import uuid4

import pytest
from pydantic import ValidationError

from zhifu.evaluation import Dataset, Judgment, RetrievalManifest, Review, digest, write_report


def test_agent_dataset_keeps_200_unique_cases_and_separate_groups():
    dataset = Dataset.model_validate_json(Path("evaluations/agent-scenarios.json").read_bytes())
    assert len(dataset.cases()) == 200
    assert sum(case["split"] == "holdout" for case in dataset.cases()) == 100
    invalid = dataset.model_dump()
    invalid["groups"][1]["prompts"][0] = invalid["groups"][0]["prompts"][0] + " "
    with pytest.raises(ValidationError, match="唯一"):
        Dataset.model_validate(invalid)
    invalid = dataset.model_dump()
    invalid["groups"][0]["split"] = "holdout"
    with pytest.raises(ValidationError, match="各需"):
        Dataset.model_validate(invalid)


def test_review_requires_existing_unchanged_evidence():
    path = Path(".work") / f"review-validation-{uuid4().hex}.json"
    path.parent.mkdir(exist_ok=True)
    write_report(path, {"purpose": "审核字段校验测试资料"})
    try:
        review = Review(reviewer="测试审核字段", reviewed_at=datetime.now(UTC),
                        evidence_path=str(path),
                        evidence_sha256=hashlib.sha256(path.read_bytes()).hexdigest())
        review.verify()
        write_report(path, {"purpose": "内容变化校验"})
        with pytest.raises(ValueError, match="摘要"):
            review.verify()
        review.reviewed_at = datetime.now()
        with pytest.raises(ValueError, match="时区"):
            review.verify()
    finally:
        path.unlink()


def test_judgment_rejects_passing_permission_or_duplicate_failures():
    body = {"reviewer": "测试审核字段", "reviewed_at": datetime.now(UTC),
            "evidence_path": "unused", "evidence_sha256": "a" * 64, "case_id": "case",
            "observation_sha256": "b" * 64, "verdict": "PASSED",
            "permission_violations": 1, "duplicate_external_operations": 0}
    with pytest.raises(ValidationError, match="不能登记为通过"):
        Judgment.model_validate(body)
    body.update(permission_violations=0, duplicate_external_operations=1)
    with pytest.raises(ValidationError, match="不能登记为通过"):
        Judgment.model_validate(body)


def test_retrieval_requires_known_documents_and_production_review():
    identifier = str(uuid4())
    body = {"dataset_version": "schema-validation", "origin": "local_regression",
            "store_id": str(uuid4()), "documents": {identifier: 1},
            "queries": [{"query": "换货资料", "expected": [identifier], "category": "规则"}],
            "embedding_model": "BAAI/bge-m3", "embedding_revision": "pinned",
            "total_timeout_seconds": 10}
    assert RetrievalManifest.model_validate(body).origin == "local_regression"
    body["origin"] = "production_reviewed"
    with pytest.raises(ValidationError, match="人工相关性"):
        RetrievalManifest.model_validate(body)
    evidence = {"review_kind": "agent", "reviewer": "Agent审核", "reviewed_at": datetime.now(UTC),
                "evidence_path": "unused", "evidence_sha256": "a" * 64}
    body.update(review=evidence, deployment_review=evidence)
    with pytest.raises(ValidationError, match="人工相关性"):
        RetrievalManifest.model_validate(body)
    body["origin"] = "local_regression"
    body["queries"][0]["expected"] = [str(uuid4())]
    with pytest.raises(ValidationError, match="未登记资料"):
        RetrievalManifest.model_validate(body)


def test_report_replacement_preserves_json_and_canonical_digest():
    path = Path(".work") / f"report-validation-{uuid4().hex}.json"
    try:
        write_report(path, {"cases": {"a": "NOT_RUN"}})
        write_report(path, {"cases": {"a": "REVIEW_REQUIRED"}})
        assert json.loads(path.read_text(encoding="utf-8"))["cases"]["a"] == "REVIEW_REQUIRED"
        assert digest({"a": 1, "b": 2}) == digest({"b": 2, "a": 1})
        assert not path.with_suffix(".json.pending").exists()
    finally:
        path.unlink(missing_ok=True)