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

from __future__ import annotations

import asyncio
from types import SimpleNamespace

import pytest

from script_build_host.application.mission_service import (
    BuildTransitionGate,
    DirectionReconciler,
    ScriptMissionService,
    StartScriptBuildCommand,
)
from script_build_host.domain.errors import InputRelationMismatch, ProtocolViolation
from script_build_host.domain.records import BuildStatus, Principal


class _RejectingInputs:
    async def validate_source(self, **_: object) -> None:
        raise InputRelationMismatch()


class _SourceAuthorizer:
    async def require_source_access(self, *_args: object, **_kwargs: object) -> None:
        return None

    async def require_access(self, *_args: object, **_kwargs: object) -> None:
        return None


class _LegacyState:
    def __init__(self) -> None:
        self.created = 0

    async def create(self, **_: object) -> int:
        self.created += 1
        return 1


@pytest.mark.asyncio
async def test_relation_mismatch_produces_no_build_or_binding_write() -> None:
    legacy = _LegacyState()
    bindings = SimpleNamespace(created=0)
    service = ScriptMissionService(
        runner=SimpleNamespace(),
        coordinator=SimpleNamespace(),
        factory=SimpleNamespace(),
        input_snapshots=_RejectingInputs(),  # type: ignore[arg-type]
        bindings=bindings,
        legacy_state=legacy,  # type: ignore[arg-type]
        authorizer=_SourceAuthorizer(),
        direction_reconciler=SimpleNamespace(),
    )
    with pytest.raises(InputRelationMismatch):
        await service.start(
            StartScriptBuildCommand(
                execution_id=1,
                topic_build_id=2,
                topic_id=3,
                principal=Principal("tester"),
            )
        )
    assert legacy.created == 0
    assert bindings.created == 0


@pytest.mark.asyncio
async def test_source_authorization_runs_before_input_or_build_access() -> None:
    legacy = _LegacyState()
    inputs = SimpleNamespace(validated=False)

    class Denied:
        async def require_source_access(self, *_args: object, **_kwargs: object) -> None:
            raise PermissionError("source forbidden")

    service = ScriptMissionService(
        runner=SimpleNamespace(),
        coordinator=SimpleNamespace(),
        factory=SimpleNamespace(),
        input_snapshots=inputs,
        bindings=SimpleNamespace(),
        legacy_state=legacy,  # type: ignore[arg-type]
        authorizer=Denied(),  # type: ignore[arg-type]
        direction_reconciler=SimpleNamespace(),
    )
    with pytest.raises(PermissionError, match="source forbidden"):
        await service.start(StartScriptBuildCommand(1, 2, 3, Principal("intruder")))
    assert legacy.created == 0
    assert inputs.validated is False


class _StopState:
    def __init__(self, status: BuildStatus) -> None:
        self.status = status

    async def get_status(self, _build: int) -> BuildStatus:
        return self.status

    async def set_status(self, _build: int, status: BuildStatus, **_: object) -> None:
        self.status = status


@pytest.mark.asyncio
async def test_stop_waits_for_planner_even_when_there_are_no_operations() -> None:
    state = _StopState(BuildStatus.RUNNING)
    pending: asyncio.Future[None] = asyncio.get_running_loop().create_future()
    trace = SimpleNamespace(status="running")
    runner = SimpleNamespace(
        stop=lambda _root: _async_value(True),
        trace_store=SimpleNamespace(get_trace=lambda _root: _async_value(trace)),
    )
    coordinator = SimpleNamespace(
        task_store=SimpleNamespace(load=lambda _root: _async_value(SimpleNamespace(operations={})))
    )
    service = ScriptMissionService(
        runner=runner,
        coordinator=coordinator,
        factory=SimpleNamespace(),
        input_snapshots=SimpleNamespace(),
        bindings=SimpleNamespace(
            get_by_build=lambda _build: _async_value(SimpleNamespace(root_trace_id="root"))
        ),
        legacy_state=state,
        authorizer=SimpleNamespace(require_access=lambda _principal, _build: _async_value(None)),
        direction_reconciler=SimpleNamespace(),
        stop_timeout_seconds=0.01,
    )
    service._runs[7] = pending  # type: ignore[assignment]
    result = await service.stop(7, Principal("owner"))
    assert result.status == BuildStatus.STOPPING
    assert state.status == BuildStatus.STOPPING
    pending.cancel()


@pytest.mark.asyncio
async def test_stop_is_idempotent_and_publication_is_gated_by_durable_status() -> None:
    state = _StopState(BuildStatus.STOPPED)
    service = ScriptMissionService(
        runner=SimpleNamespace(),
        coordinator=SimpleNamespace(),
        factory=SimpleNamespace(),
        input_snapshots=SimpleNamespace(),
        bindings=SimpleNamespace(),
        legacy_state=state,
        authorizer=SimpleNamespace(require_access=lambda _principal, _build: _async_value(None)),
        direction_reconciler=SimpleNamespace(),
    )
    result = await service.stop(7, Principal("owner"))
    assert result.status == BuildStatus.STOPPED

    state.status = BuildStatus.STOPPING
    publications = SimpleNamespace(prepared=False)
    reconciler = DirectionReconciler(
        coordinator=SimpleNamespace(),
        bindings=SimpleNamespace(),
        artifacts=SimpleNamespace(),
        publications=publications,
        legacy_state=state,
    )
    with pytest.raises(ProtocolViolation, match="forbidden while stopping"):
        await reconciler.reconcile(7, "root")
    assert publications.prepared is False


@pytest.mark.asyncio
async def test_stop_intent_and_direction_reconcile_are_serialized() -> None:
    gate = BuildTransitionGate()
    entered = asyncio.Event()
    release = asyncio.Event()
    state = _StopState(BuildStatus.RUNNING)

    class BlockingReconciler:
        transition_gate = gate

        async def reconcile(self, script_build_id: int, _root: str) -> int:
            async with gate.hold(script_build_id):
                entered.set()
                await release.wait()
                assert state.status is BuildStatus.RUNNING
                return 1

    runner = SimpleNamespace(
        stop=lambda _root: _async_value(True),
        trace_store=SimpleNamespace(get_trace=lambda _root: _async_value(None)),
    )
    coordinator = SimpleNamespace(
        task_store=SimpleNamespace(load=lambda _root: _async_value(SimpleNamespace(operations={})))
    )
    service = ScriptMissionService(
        runner=runner,
        coordinator=coordinator,
        factory=SimpleNamespace(),
        input_snapshots=SimpleNamespace(),
        bindings=SimpleNamespace(
            get_by_build=lambda _build: _async_value(SimpleNamespace(root_trace_id="root"))
        ),
        legacy_state=state,
        authorizer=SimpleNamespace(require_access=lambda _principal, _build: _async_value(None)),
        direction_reconciler=BlockingReconciler(),  # type: ignore[arg-type]
        transition_gate=gate,
    )
    reconcile = asyncio.create_task(service.direction_reconciler.reconcile(7, "root"))
    await entered.wait()
    stop = asyncio.create_task(service.stop(7, Principal("owner")))
    await asyncio.sleep(0)
    assert state.status is BuildStatus.RUNNING
    release.set()
    assert await reconcile == 1
    assert (await stop).status is BuildStatus.STOPPED


async def _async_value(value):
    return value