video-image-production-build/tests/visualization/test_store.py

from pathlib import Path

from visualization.web.store import FileRunStore


PROJECT_ROOT = Path(__file__).resolve().parents[2]


def test_discovers_current_global_and_production_runs() -> None:
    store = FileRunStore(PROJECT_ROOT / "demo_output")

    refs = store.discover()
    kinds = {ref.kind for ref in refs}

    assert "global_data" in kinds
    assert "segment" in kinds


def test_global_run_projects_upstream_compatible_views() -> None:
    store = FileRunStore(PROJECT_ROOT / "demo_output")
    ref = next(
        item for item in store.discover()
        if item.key == "global-data-stage-live-v22"
    )

    brief = store.brief(ref)
    detail = store.detail(ref)
    snapshots = store.snapshots(ref)
    flat = store.flat(ref, 0)

    assert brief["status"] == "success"
    assert brief["raw_status"] == "COMPLETED"
    assert brief["business_label"] == "Global Data · 4 个资料任务"
    assert brief["business_progress"] == "4 已通过"
    assert brief["cost_status"] in {"reported", "unavailable"}
    assert detail["steps"]
    assert detail["overview"] == {
        "plan_rounds": 1,
        "execute_rounds": 4,
        "task_count": 4,
        "passed": 4,
        "failed": 0,
        "retries": 0,
    }
    assert snapshots["snapshots"]
    assert flat["root"]["module_key"] == "production.global_data.workflow"
    assert flat["round_count"] >= 1
    assert flat["rounds"][0]["instances"]
    executor = next(
        item
        for item in flat["rounds"][0]["instances"]
        if item["module_key"] == "production.global_data.execute"
        and item["task_id"] == "Task1"
    )
    assert executor["output"]["images"]


def test_production_run_reads_latest_versioned_plan_without_terminal_report() -> None:
    store = FileRunStore(PROJECT_ROOT / "demo_output")
    ref = next(
        item for item in store.discover()
        if item.key == "v22-production-v06-live-v3"
    )

    detail = store.detail(ref)
    flat = store.flat(ref, 0)

    assert detail["agent_name"] == "segment_production"
    assert detail["input_payload"]["schema_version"] == "0.6"
    assert detail["overview"]["plan_rounds"] == 1
    assert detail["overview"]["task_count"] == 4
    assert detail["overview"]["execute_rounds"] > 0
    assert flat["round_count"] == 1
    assert flat["rounds"][0]["instances"][0]["module_key"] == (
        "production.segment.execute"
    )
    assert (
        flat["rounds"][0]["instances"][0]["input"]["blocks"][0]["source"]
        == "segment_packages/Segment1.v1.json"
    )
    assert flat["root"]["module"]["fingerprint"] == "0.6"


def test_failed_production_run_uses_formal_summary_and_explains_failure() -> None:
    store = FileRunStore(PROJECT_ROOT / "demo_output")
    ref = next(
        item for item in store.discover()
        if item.key == "v23-context-production-from-v22-01"
    )

    brief = store.brief(ref)
    detail = store.detail(ref)
    story = detail["business_story"]

    assert brief["raw_status"] == "FAILED"
    assert brief["status"] == "failed"
    assert brief["business_label"] == "Production · 4 个 Segment"
    assert brief["business_progress"] == "停在 Segment1 · EXECUTE_SEGMENT"
    assert detail["error_message"].startswith("Segment Executor 连续 3 次")
    assert story["protocol_version"] == "0.7"
    assert story["status"] == "failed"
    assert story["failure"]["code"] == "SEGMENT_EXECUTION_FAILED"
    segment = story["rounds"][0]["steps"][0]
    assert segment["id"] == "Segment1"
    assert segment["status"] == "failed"
    assert segment["judgment"]["verdict"] == "FAIL"
    assert "正式音色参考 Artifact" in segment["judgment"]["summary"]
    assert len(segment["tools"]) == 12


def test_global_business_story_exposes_reason_basis_judgment_and_output() -> None:
    store = FileRunStore(PROJECT_ROOT / "demo_output")
    ref = next(
        item for item in store.discover()
        if item.key == "v23-context-global-data-01"
    )

    story = store.detail(ref)["business_story"]
    task = next(
        item for item in story["rounds"][0]["steps"]
        if item["id"] == "Task1"
    )

    assert story["headline"].startswith("把 Production 需要的")
    assert task["reason"] == "确认核心人物素材的身份一致性与技术完整性"
    assert any(item["label"] == "业务来源" for item in task["basis"])
    assert task["judgment"]["verdict"] == "PASS"
    assert task["judgment"]["criteria"]
    assert task["outputs"][0]["media_url"].startswith("/api/runs/")
    assert task["tools"][0]["name"] == "probe_media"


def test_production_tool_operations_are_not_duplicated_across_segments() -> None:
    store = FileRunStore(PROJECT_ROOT / "demo_output")
    ref = next(
        item for item in store.discover()
        if item.key == "v23-context-production-from-v22-01"
    )

    flat = store.flat(ref, 0)
    executors = [
        item for item in flat["rounds"][0]["instances"]
        if item["module_key"] == "production.segment.execute"
    ]

    assert len(executors[0]["flow"]) == 12
    assert all(not item["flow"] for item in executors[1:])


def test_completed_global_run_without_summary_uses_stage_delivery_status() -> None:
    store = FileRunStore(PROJECT_ROOT / "demo_output")
    ref = next(
        item for item in store.discover()
        if item.key == "global-data-stage-live-v19"
    )

    brief = store.brief(ref)

    assert brief["raw_status"] == "GLOBAL_DATA_COMPLETED"
    assert brief["status"] == "success"


def test_media_path_cannot_escape_run_directory() -> None:
    store = FileRunStore(PROJECT_ROOT / "demo_output")
    ref = next(
        item for item in store.discover()
        if item.key == "global-data-stage-live-v22"
    )

    try:
        store.media_path(ref, "../v22-production-plan.v1.json")
    except FileNotFoundError:
        pass
    else:
        raise AssertionError("media path traversal must be rejected")