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

from __future__ import annotations

import json
from pathlib import Path
from types import SimpleNamespace

from obagent_sdk import observe
from obagent_sdk.integrations import langgraph

from production_build_agents.observability.global_data import (
    GlobalDataObservation,
)


class _RunHandle:
    def __init__(self) -> None:
        self.finished: list[dict] = []

    def finish(self, **payload) -> None:
        self.finished.append(payload)


class _RunManager:
    def __init__(self, handle: _RunHandle) -> None:
        self.handle = handle
        self.exits: list[tuple] = []

    def __enter__(self) -> _RunHandle:
        return self.handle

    def __exit__(self, *args):
        self.exits.append(args)
        if args[1] is not None:
            raise args[1]
        return False


def test_global_data_context_forces_sanitized_manual_collection(
    monkeypatch,
    tmp_path: Path,
) -> None:
    captured: dict = {}
    handle = _RunHandle()
    manager = _RunManager(handle)

    def fake_run(**kwargs):
        captured.update(kwargs)
        return manager

    monkeypatch.setattr(observe, "run", fake_run)
    monkeypatch.setattr(
        langgraph,
        "graph_spec",
        lambda _graph: {"nodes": [{"key": "preprocess"}], "edges": []},
    )
    settings = SimpleNamespace(
        configure_sdk=lambda: True,
        project="safe-project",
        project_name="安全项目",
    )
    result = {
        "run_id": "safe-run",
        "protocol_version": "0.3",
        "status": "COMPLETED",
        "phase": "FINALIZE",
        "input_path": "/private/客户标题.json",
        "output_dir": str(tmp_path),
        "event_log": ["客户正文"],
    }

    with GlobalDataObservation(
        project_root=tmp_path,
        thread_id="safe-run",
        input_sha256="a" * 64,
        protocol_version="0.3",
        execution_mode="fresh",
        graph=object(),
        settings=settings,
    ) as observation:
        observation.finish(result, graph_invoked=True)

    encoded = json.dumps(
        {"run": captured, "finish": handle.finished},
        ensure_ascii=False,
    )
    assert captured["auto_collect"] is False
    assert captured["payload"]["input_sha256"] == "a" * 64
    assert captured["meta"]["redaction_mode"] == "allowlist_v1"
    assert handle.finished[0]["ok"] is True
    assert "/private/客户标题.json" not in encoded
    assert "客户正文" not in encoded