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

from __future__ import annotations

import httpx
import pytest

from script_build_host.internal_e2e import _wait_for_root


@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,
            )