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

import json
from pathlib import Path

import httpx
import pytest

from script_build_host.domain.records import BuildStatus
from script_build_host.internal_e2e import _wait_for_root


def test_recovery_checkpoint_fixtures_cover_all_segment_boundaries() -> None:
    path = Path(__file__).parent / "fixtures" / "recovery_checkpoints.json"
    payload = json.loads(path.read_text(encoding="utf-8"))
    assert set(payload["checkpoints"]) == {"direction", "structure", "portfolio", "root"}
    assert all(
        item["input_snapshot_digest"].startswith("sha256:")
        for item in payload["checkpoints"].values()
    )


@pytest.mark.asyncio
async def test_mission_poll_allows_only_bounded_startup_failures() -> None:
    calls = 0

    def handler(_request: httpx.Request) -> httpx.Response:
        nonlocal calls
        calls += 1
        if calls <= 5:
            return httpx.Response(500, json={"error_code": "LEDGER_NOT_READY"})
        return httpx.Response(
            200,
            json={
                "root_task_id": "root",
                "tasks": [
                    {
                        "task_id": "root",
                        "status": "blocked",
                        "blocked_reason": "BOUNDARY",
                    }
                ],
            },
        )

    async with httpx.AsyncClient(
        transport=httpx.MockTransport(handler), base_url="http://internal"
    ) as client:
        result = await _wait_for_root(
            client,
            7,
            expected_status="blocked",
            expected_reason="BOUNDARY",
            timeout_seconds=1,
            poll_seconds=0,
        )

    assert result["task_id"] == "root"
    assert calls == 6


@pytest.mark.asyncio
async def test_mission_poll_surfaces_persistent_server_failure() -> None:
    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(500, json={"error_code": "InternalServerError"})

    async with httpx.AsyncClient(
        transport=httpx.MockTransport(handler), base_url="http://internal"
    ) as client:
        with pytest.raises(RuntimeError, match="mission poll failed with HTTP 500"):
            await _wait_for_root(
                client,
                7,
                expected_status="blocked",
                expected_reason="BOUNDARY",
                timeout_seconds=1,
                poll_seconds=0,
            )


@pytest.mark.asyncio
async def test_mission_poll_stops_when_build_is_terminal() -> None:
    class FailedState:
        async def get_status(self, _script_build_id):
            return BuildStatus.FAILED

    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            json={
                "root_task_id": "root",
                "tasks": [{"task_id": "root", "status": "needs_replan"}],
            },
        )

    async with httpx.AsyncClient(
        transport=httpx.MockTransport(handler), base_url="http://internal"
    ) as client:
        with pytest.raises(RuntimeError, match="terminal status=failed"):
            await _wait_for_root(
                client,
                7,
                legacy_state=FailedState(),
                expected_status="blocked",
                expected_reason="BOUNDARY",
                timeout_seconds=1,
                poll_seconds=0,
            )