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

from __future__ import annotations

from typing import Any

import pytest

from script_build_host.application.input_snapshot_service import (
    AssembleInputRequest,
    ScriptInputSnapshotService,
)
from script_build_host.domain.input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
from script_build_host.domain.ports import PersonaInput
from script_build_host.domain.records import Principal


class FakeLegacyInput:
    def __init__(self) -> None:
        self.calls: list[tuple[int, int, int]] = []

    async def read_topic_graph(
        self, *, execution_id: int, topic_build_id: int, topic_id: int
    ) -> dict[str, Any]:
        self.calls.append((execution_id, topic_build_id, topic_id))
        return {
            "build_record": {
                "personal_config": {"account_name": "account"},
                "strategies_config": {"always_on": [7], "on_demand": [8]},
                "agent_config": {"api_key": "hidden", "model": "fake"},
            },
            "topic": {"id": topic_id},
            "points": [],
            "composition_items": [],
            "relations": [],
            "sources": [],
        }


class FakePersona:
    async def load(self, account_name: str) -> PersonaInput:
        return PersonaInput(account_name, account_name, ({"point": "one"},), (), {"p": "sha256:x"})


class FakeStrategies:
    async def load(
        self, *, always_on: tuple[Any, ...], on_demand: tuple[Any, ...]
    ) -> tuple[dict[str, Any], ...]:
        return ({"always_on": list(always_on), "on_demand": list(on_demand)},)


class FakePrompts:
    async def load(self, requests: tuple[Any, ...]) -> tuple[dict[str, Any], ...]:
        return ({"source": "fixture", "count": len(requests)},)


class FakeSnapshots:
    def __init__(self) -> None:
        self.frozen: ScriptBuildInput | None = None

    async def freeze(
        self, assembled: ScriptBuildInput, *, version: int = 1
    ) -> ScriptBuildInputSnapshotV1:
        self.frozen = assembled
        raise NotImplementedError

    async def get(self, snapshot_id: str, *, script_build_id: int) -> ScriptBuildInputSnapshotV1:
        raise NotImplementedError


@pytest.mark.asyncio
async def test_validate_source_is_read_only_and_assemble_redacts_secrets() -> None:
    legacy = FakeLegacyInput()
    service = ScriptInputSnapshotService(
        legacy_input=legacy,
        persona_source=FakePersona(),
        strategy_source=FakeStrategies(),
        prompt_source=FakePrompts(),
        snapshots=FakeSnapshots(),
    )
    await service.validate_source(execution_id=10, topic_build_id=20, topic_id=30)
    assembled = await service.assemble(
        AssembleInputRequest(
            script_build_id=1,
            execution_id=10,
            topic_build_id=20,
            topic_id=30,
            principal=Principal("user-1"),
            datasource_manifest={"endpoint": "https://api.example?q=ok&token=hidden"},
            model_manifest={"planner": {"model": "fake", "api_key": "hidden"}},
        )
    )
    assert legacy.calls == [(10, 20, 30), (10, 20, 30)]
    assert "hidden" not in str(assembled.canonical_payload())
    assert assembled.strategies[0] == {"always_on": [7], "on_demand": [8]}