script-build-and-framework-tmp/script_build_host/tests/test_production_composition.py

from __future__ import annotations

import json
from pathlib import Path

import pytest
from agent import AgentRunner, FileSystemTraceStore

from script_build_host.domain.records import Principal
from script_build_host.infrastructure.config import ScriptBuildSettings
from script_build_host.production import (
    ProductionRuntimeDependencies,
    compose_production_host,
)


class _Principals:
    async def current(self, request_context: object | None = None) -> Principal:
        if request_context is None:
            raise PermissionError("request context is required")
        return Principal("owner")


class _Authorizer:
    async def require_source_access(self, _principal: Principal, **_source: int) -> None:
        return None

    async def require_access(self, _principal: Principal, _script_build_id: int) -> None:
        return None


def _runtime(tmp_path: Path) -> ProductionRuntimeDependencies:
    async def llm_call(*_args: object, **_kwargs: object) -> dict[str, str]:
        return {"content": "unused", "finish_reason": "stop"}

    return ProductionRuntimeDependencies(
        runner=AgentRunner(
            trace_store=FileSystemTraceStore(str(tmp_path / "traces")),
            llm_call=llm_call,
        ),
        principal_provider=_Principals(),
        build_authorizer=_Authorizer(),
    )


def _settings(tmp_path: Path, **overrides: object) -> ScriptBuildSettings:
    decode = tmp_path / "decode.json"
    knowledge = tmp_path / "knowledge.json"
    decode.write_text(json.dumps({"items": []}), encoding="utf-8")
    knowledge.write_text(json.dumps({"items": []}), encoding="utf-8")
    values: dict[str, object] = {
        "environment": "test",
        "read_database_url": f"sqlite+aiosqlite:///{tmp_path / 'read.db'}",
        "write_database_url": f"sqlite+aiosqlite:///{tmp_path / 'write.db'}",
        "agent_data_root": tmp_path / "agent-data",
        "persona_root": tmp_path / "persona",
        "section_pattern_root": tmp_path / "sections",
        "decode_index_root": decode,
        "knowledge_root": knowledge,
        "prompt_fallback_root": tmp_path / "prompts",
        "websocket_allowed_origins": ("https://ui.example",),
    }
    values.update(overrides)
    return ScriptBuildSettings(**values)  # type: ignore[arg-type]


@pytest.mark.asyncio
async def test_production_composition_wires_real_stores_and_closes_resources(
    tmp_path: Path,
) -> None:
    host = compose_production_host(_settings(tmp_path), _runtime(tmp_path))
    await host.validate_startup()
    manifest = await host.runtime_manifest.load()
    assert manifest["decode_index"]["file_count"] == 1
    assert manifest["knowledge"]["file_count"] == 1
    paths = set(host.composition.app.openapi()["paths"])
    assert "/api/pattern/script_builds" in paths
    assert any(path.startswith("/api/v2/script-builds/") for path in paths)
    assert (
        host.composition.coordinator.task_store.base_path
        == (tmp_path / "agent-data" / "task-ledger").resolve()
    )
    await host.close()
    await host.close()
    assert host.http_client.is_closed


def test_production_requires_decode_service_and_durable_trace_store(tmp_path: Path) -> None:
    production = _settings(
        tmp_path,
        environment="production",
        read_database_url="mysql+asyncmy://reader:read@db.example/app",
        write_database_url="mysql+asyncmy://writer:write@db.example/app",
    )
    with pytest.raises(RuntimeError, match="DECODE_ENDPOINT"):
        compose_production_host(production, _runtime(tmp_path))

    runtime = _runtime(tmp_path)
    runtime.runner.trace_store = None
    with pytest.raises(RuntimeError, match="durable trace store"):
        compose_production_host(_settings(tmp_path), runtime)