video-image-production-build/tests/support/validator_fixtures.py

from __future__ import annotations

from pathlib import Path

from langchain_core.messages import AIMessage

from production_build_agents.contracts.models import (
    ArtifactRejection,
    GlobalDataStageValidatorCandidate,
    StageRequirementResult,
    ValidationCriterionResult,
    ValidatorCandidate,
)
from tests.support.fake_models import ToolAwareFakeChatModel
from production_build_agents.agents.validator.stage_context import required_stage_audit_paths


_BRIEF_PATH = (
    Path(__file__).parents[1] / "fixtures" / "minimal_brief.json"
)
STAGE_AUDIT_PATHS = required_stage_audit_paths(str(_BRIEF_PATH))


def build_stage_read_call(call_id: str) -> AIMessage:
    return AIMessage(
        content="",
        tool_calls=[
            {
                "name": "read_production_brief",
                "args": {"source_paths": STAGE_AUDIT_PATHS},
                "id": call_id,
                "type": "tool_call",
            }
        ],
    )


def build_validator_candidate(
    *,
    run_id: str = "test-run",
    plan_id: str = "GlobalDataPlan",
    task_id: str = "Task1",
    plan_version: int = 1,
    verdict: str = "PASS",
    expectation_id: str = "Requirement1-Expectation1",
    verification_capability: str = "document_content",
) -> ValidatorCandidate:
    return ValidatorCandidate(
        run_id=run_id,
        plan_id=plan_id,
        plan_version=plan_version,
        task_id=task_id,
        executor_run_id=(
            f"{run_id}-executor-{task_id}-v{plan_version}"
        ),
        criterion_results=[
            ValidationCriterionResult(
                expectation_id=expectation_id,
                verification_capability=verification_capability,
                verdict=verdict,
                evidence=[f"{task_id} 的验收证据"],
                reason=f"{task_id} 的 Expectation 验收结论",
            )
        ],
        summary=f"{task_id} 验收为 {verdict}",
    )


def build_validator_model(
    *candidates: ValidatorCandidate,
    stage_candidate: GlobalDataStageValidatorCandidate | None = None,
    include_stage: bool = True,
) -> ToolAwareFakeChatModel:
    responses = [
        AIMessage(content=candidate.model_dump_json())
        for candidate in candidates
    ]
    if include_stage and (candidates or stage_candidate):
        selected = stage_candidate or build_stage_validator_candidate(
            run_id=candidates[-1].run_id,
            plan_id=candidates[-1].plan_id,
            plan_version=candidates[-1].plan_version,
        )
        responses.extend(
            [
                build_stage_read_call("stage-read-production-brief"),
                AIMessage(content=selected.model_dump_json()),
            ]
        )
    return ToolAwareFakeChatModel(
        responses=responses
    )


def build_stage_validator_candidate(
    *,
    run_id: str = "test-run",
    plan_id: str = "GlobalDataPlan",
    plan_version: int = 1,
    requirement_ids: tuple[str, ...] = ("Requirement1",),
    verdict: str = "PASS",
    missing_critical: bool = False,
    missing_minor: bool = False,
    artifact_rejections: list[ArtifactRejection] | None = None,
) -> GlobalDataStageValidatorCandidate:
    return GlobalDataStageValidatorCandidate(
        run_id=run_id,
        plan_id=plan_id,
        plan_version=plan_version,
        inspected_source_paths=STAGE_AUDIT_PATHS,
        requirement_results=[
            StageRequirementResult(
                requirement_id=requirement_id,
                verdict=verdict,
                evidence_artifact_ids=[],
                evidence=[f"{requirement_id} 的阶段证据"],
                reason=f"{requirement_id} 阶段验收为 {verdict}",
            )
            for requirement_id in requirement_ids
        ],
        missing_requirements=[
            *(
                [
                    {
                        "description": "缺少关键共享素材",
                        "importance": "critical",
                        "source_paths": ["$.核心制作点[0]"],
                        "expected_artifact_types": ["image"],
                    }
                ]
                if missing_critical
                else []
            ),
            *(
                [
                    {
                        "description": "缺少非关键参考说明",
                        "importance": "minor",
                        "source_paths": ["$.核心制作点[0]"],
                        "expected_artifact_types": ["document"],
                    }
                ]
                if missing_minor
                else []
            ),
        ],
        artifact_rejections=artifact_rejections or [],
        summary="Global Data 阶段验收完成",
    )