video-image-production-build/tests/observability/test_full_capture.py

from __future__ import annotations

import json
from pathlib import Path

from production_build_agents.observability.full_capture import (
    expand_global_data_result,
    expand_global_data_state,
    expand_node_update,
    load_observation_record,
)


def test_full_capture_reads_json_and_markdown_records(
    tmp_path: Path,
) -> None:
    input_path = tmp_path / "production_final.json"
    report_path = tmp_path / "PreprocessReport.v1.md"
    input_path.write_text(
        json.dumps(
            {
                "title": "完整标题",
                "image": "https://internal.example/image.png",
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    report_path.write_text("# 完整预处理报告", encoding="utf-8")

    state = {
        "run_id": "full-run",
        "input_path": str(input_path),
        "preprocess_report_path": str(report_path),
        "plan_history": {
            "1": {
                "json_uri": str(input_path),
                "remote_uri": "https://internal.example/reference.json",
            }
        },
        "event_log": ["完整事件"],
    }
    observed = expand_global_data_state(state)

    assert observed["state"] == state
    assert observed["records"]["input_path"]["content"]["title"] == (
        "完整标题"
    )
    assert observed["records"]["preprocess_report_path"]["content"] == (
        "# 完整预处理报告"
    )
    assert (
        observed["records"]["plan_history.1.json_uri"]["content"]["title"]
        == "完整标题"
    )
    assert "plan_history.1.remote_uri" not in observed["records"]


def test_full_capture_preserves_node_state_error_and_media_url(
    tmp_path: Path,
) -> None:
    delivery_path = tmp_path / "GlobalDataStageDelivery.v1.json"
    delivery_path.write_text(
        json.dumps(
            {
                "body": "真实正文",
                "video_url": "https://internal.example/final.mp4",
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    state = {"run_id": "full-run", "status": "RUNNING"}
    error = RuntimeError("真实内部错误内容")
    observed = expand_node_update(
        "finalize_global_data",
        state,
        {
            "status": "FAILED",
            "error": "模型返回失败",
            "global_data_delivery_path": str(delivery_path),
        },
        exception=error,
    )

    assert observed["state_after"]["error"] == "模型返回失败"
    assert observed["exception"]["message"] == "真实内部错误内容"
    assert (
        observed["records"]["global_data_delivery_path"]["content"][
            "video_url"
        ]
        == "https://internal.example/final.mp4"
    )


def test_full_capture_result_includes_full_final_state(
    tmp_path: Path,
) -> None:
    result = {
        "run_id": "full-run",
        "status": "COMPLETED",
        "event_log": ["计划完成", "阶段验收通过"],
        "task_records": {
            "Task1": {
                "executor_summary": "完整执行摘要",
                "validator_summary": "完整校验摘要",
            }
        },
    }

    observed = expand_global_data_result(
        result,
        execution_mode="fresh",
        graph_invoked=True,
    )

    assert observed["result"] == result
    assert observed["execution_mode"] == "fresh"
    assert observed["graph_invoked"] is True


def test_binary_record_is_not_embedded(tmp_path: Path) -> None:
    media_path = tmp_path / "video.mp4"
    media_path.write_bytes(b"\x00\x01\x02")

    observed = load_observation_record(media_path)

    assert observed is not None
    assert observed["path"] == str(media_path)
    assert observed["content_omitted"] == "non_text_record"
    assert "content" not in observed