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["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"
    )


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")