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

from __future__ import annotations

from pathlib import Path

from sqlalchemy import insert
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from script_build_host.application.legacy_api import LegacyScriptBuildApiService
from script_build_host.application.legacy_projection import LegacyDetailProjectionService
from script_build_host.domain.records import Principal
from script_build_host.infrastructure.legacy_tables import (
    legacy_metadata,
    script_build_element,
    script_build_paragraph,
    script_build_paragraph_element,
    script_build_record,
    script_build_task_plan_step,
)
from script_build_host.infrastructure.raw_artifacts import FileRawArtifactStore
from script_build_host.infrastructure.tables import metadata


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


async def test_ten_table_fresh_database_supports_detail_and_log_without_history_tables() -> None:
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    legacy_tables = (
        script_build_record,
        script_build_paragraph,
        script_build_element,
        script_build_paragraph_element,
        script_build_task_plan_step,
    )
    async with engine.begin() as connection:
        await connection.run_sync(metadata.create_all)
        await connection.run_sync(
            lambda sync_connection: legacy_metadata.create_all(
                sync_connection,
                tables=legacy_tables,
            )
        )

    sessions = async_sessionmaker(engine, expire_on_commit=False)
    async with sessions() as session, session.begin():
        await session.execute(
            insert(script_build_record).values(
                id=1,
                execution_id=11,
                topic_build_id=12,
                topic_id=13,
                status="success",
                reson_trace_id="root-1",
                is_deleted=False,
                is_favorited=False,
                summary="fresh output",
            )
        )

    authorizer = _Authorizer()
    projection = LegacyDetailProjectionService(
        sessions,
        authorizer,  # type: ignore[arg-type]
        include_legacy_decision_traces=False,
    )
    detail = await projection.project_authorized(1, principal=Principal("owner"))
    assert detail["build"]["id"] == 1
    assert detail["decision_traces"] == []

    legacy_api = LegacyScriptBuildApiService(
        sessions,
        sessions,
        authorizer,  # type: ignore[arg-type]
        include_legacy_logs=False,
    )
    log = await legacy_api.log(1, Principal("owner"))
    assert log["log_content"] == "fresh output"

    await engine.dispose()


async def test_raw_artifact_store_persists_and_revalidates_content(tmp_path: Path) -> None:
    store = FileRawArtifactStore(tmp_path)
    reference = await store.freeze_bytes(b"fresh raw output", media_type="text/plain")

    assert await store.read_bytes(reference) == b"fresh raw output"