script-build-and-framework-tmp/visualization/backend/tests/test_host_client.py

import asyncio
from pathlib import Path

import httpx

from app.host_client import HostClient


def test_backend_has_no_local_runtime_file_dependency() -> None:
    source = "\n".join(
        path.read_text(encoding="utf-8")
        for path in (Path(__file__).parents[1] / "app").glob("*.py")
    )

    assert "SCRIPT_BUILD_AGENT_DATA_ROOT" not in source
    assert ".local" not in source


def test_host_client_reads_authoritative_api_and_forwards_only_auth_headers() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/api/pattern/script_builds"
        assert request.headers["authorization"] == "Bearer test"
        assert "x-secret-debug" not in request.headers
        return httpx.Response(200, json={"items": []})

    client = HostClient(
        "http://host.test",
        transport=httpx.MockTransport(handler),
    )

    async def run() -> dict:
        try:
            return await client.list_runs(
                {"authorization": "Bearer test", "x-secret-debug": "do-not-forward"}
            )
        finally:
            await client.aclose()

    payload = asyncio.run(run())

    assert payload == {"items": []}


def test_journey_reads_incrementally_and_caches_immutable_inputs() -> None:
    calls: list[str] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        path = request.url.path
        calls.append(str(request.url))
        if path.endswith("/mission"):
            return httpx.Response(
                200,
                json={
                    "tasks": [
                        {
                            "task_id": "task-1",
                            "current_spec": {
                                "version": 1,
                                "context_refs": ["script-build://task-contracts/sha256/contract-1"],
                            },
                        }
                    ]
                },
            )
        if path.endswith("/input-summary"):
            return httpx.Response(200, json={"input_snapshot_id": "snapshot-1"})
        if path.endswith("/mission/events"):
            if request.url.params.get("after"):
                return httpx.Response(200, json={"events": [], "next_cursor": "cursor-1"})
            return httpx.Response(
                200,
                json={
                    "events": [{"sequence": 1}],
                    "next_cursor": "cursor-1",
                },
            )
        if path.endswith("/traces"):
            return httpx.Response(
                200,
                json={"traces": [{"trace_id": "root-1", "status": "running"}]},
            )
        if path.endswith("/contract"):
            return httpx.Response(200, json={"task_id": "task-1", "task_kind": "structure"})
        if path.endswith("/messages"):
            after = int(request.url.params.get("after_sequence", "0"))
            messages = [{"sequence": 1, "role": "assistant"}] if after == 0 else []
            return httpx.Response(200, json={"messages": messages})
        raise AssertionError(path)

    async def run() -> tuple[dict, dict]:
        client = HostClient("http://host.test", transport=httpx.MockTransport(handler))
        try:
            first = await client.journey_source(7, {})
            second = await client.journey_source(7, {})
            return first, second
        finally:
            await client.aclose()

    first, second = asyncio.run(run())

    assert first["events"] == second["events"] == [{"sequence": 1}]
    assert first["messages"] == second["messages"]
    assert sum("input-summary" in value for value in calls) == 1
    assert sum(value.endswith("/contract") for value in calls) == 1
    assert any("after_sequence=1" in value for value in calls)