agentci/tests/test_development.py

import asyncio
from pathlib import Path

import pytest

from agentci.integrations.development import (
    DevelopmentEnvironment,
    DevelopmentEnvironmentError,
)


def environment(
    tmp_path: Path, scripts: list[str], *, timeout_seconds: int = 5
) -> DevelopmentEnvironment:
    scripts_dir = tmp_path / "scripts"
    scripts_dir.mkdir()
    return DevelopmentEnvironment(
        scripts=scripts,
        scripts_dir=scripts_dir,
        tools_dir=tmp_path / "tools",
        timeout_seconds=timeout_seconds,
        python_version="3.13",
        dotnet_channel="10.0",
    )


def script(path: Path, body: str) -> None:
    path.write_text(f"#!/bin/sh\nset -eu\n{body}\n")
    path.chmod(0o755)


async def test_runs_custom_scripts_in_order_with_sanitized_environment(
    tmp_path, monkeypatch
) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    development = environment(tmp_path, ["first", "second"])
    script(
        development.scripts_dir / "first",
        'printf \'first:%s:%s\\n\' "$DEV_TOOLS_DIR" "${AGENTCI_SECRET-unset}" >> order',
    )
    script(development.scripts_dir / "second", "printf 'second\\n' >> order")
    monkeypatch.setenv("AGENTCI_SECRET", "must-not-leak")

    await development.prepare(workspace)

    assert (workspace / "order").read_text().splitlines() == [
        f"first:{tmp_path / 'tools'}:unset",
        "second",
    ]
    assert (tmp_path / "tools" / "bin").is_dir()


async def test_supplied_script_names_use_the_same_directory(tmp_path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    development = environment(tmp_path, ["python"])
    script(development.scripts_dir / "python", "printf 'python\\n' > selected")

    await development.prepare(workspace)

    assert (workspace / "selected").read_text() == "python\n"


async def test_runs_non_executable_shell_script_from_bind_mount(tmp_path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    development = environment(tmp_path, ["mounted"])
    mounted = development.scripts_dir / "mounted"
    mounted.write_text("#!/bin/sh\nprintf 'mounted\\n' > selected\n")
    mounted.chmod(0o644)

    await development.prepare(workspace)

    assert (workspace / "selected").read_text() == "mounted\n"


async def test_serializes_concurrent_preparation(tmp_path, monkeypatch) -> None:
    development = environment(tmp_path, ["shared"])
    script(development.scripts_dir / "shared", "true")
    started = asyncio.Event()
    release = asyncio.Event()
    active = 0
    maximum_active = 0

    async def run(*_args) -> None:
        nonlocal active, maximum_active
        active += 1
        maximum_active = max(maximum_active, active)
        started.set()
        await release.wait()
        active -= 1

    monkeypatch.setattr(development, "_run", run)
    first = asyncio.create_task(development.prepare(tmp_path / "first"))
    await started.wait()
    second = asyncio.create_task(development.prepare(tmp_path / "second"))
    await asyncio.sleep(0)
    release.set()
    await asyncio.gather(first, second)

    assert maximum_active == 1


async def test_reports_script_failure_output(tmp_path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    development = environment(tmp_path, ["broken"])
    script(development.scripts_dir / "broken", "printf 'failed detail' >&2; exit 7")

    with pytest.raises(DevelopmentEnvironmentError, match="exited with 7: failed detail"):
        await development.prepare(workspace)


async def test_reports_missing_install_script(tmp_path: Path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    development = environment(tmp_path, ["missing"])

    with pytest.raises(
        DevelopmentEnvironmentError,
        match=r"Install script 'missing' was not found",
    ):
        await development.prepare(workspace)


async def test_stops_before_second_script_after_failure(tmp_path: Path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    development = environment(tmp_path, ["first", "second"])
    script(development.scripts_dir / "first", "printf 'first\n' > first-ran; exit 3")
    script(development.scripts_dir / "second", "printf 'second\n' > second-ran")

    with pytest.raises(DevelopmentEnvironmentError, match="exited with 3"):
        await development.prepare(workspace)

    assert (workspace / "first-ran").read_text() == "first\n"
    assert not (workspace / "second-ran").exists()


async def test_failure_output_keeps_only_bounded_tail(tmp_path: Path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    development = environment(tmp_path, ["verbose"])
    script(
        development.scripts_dir / "verbose",
        "printf 'discarded-prefix' >&2; "
        'i=0; while [ "$i" -lt 2100 ]; do printf x >&2; i=$((i + 1)); done; '
        "printf 'useful-tail' >&2; exit 9",
    )

    with pytest.raises(DevelopmentEnvironmentError) as raised:
        await development.prepare(workspace)

    message = str(raised.value)
    assert "discarded-prefix" not in message
    assert message.endswith("useful-tail")


async def test_wraps_subprocess_start_error(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    development = environment(tmp_path, ["broken"])
    script(development.scripts_dir / "broken", "true")

    async def create_subprocess_exec(*_args, **_kwargs):
        raise OSError("exec unavailable")

    monkeypatch.setattr(
        "agentci.integrations.development.asyncio.create_subprocess_exec",
        create_subprocess_exec,
    )

    with pytest.raises(
        DevelopmentEnvironmentError,
        match="Could not run install script 'broken': exec unavailable",
    ) as raised:
        await development.prepare(workspace)

    assert isinstance(raised.value.__cause__, OSError)


async def test_times_out_install_script(tmp_path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    development = environment(tmp_path, ["slow"], timeout_seconds=1)
    script(development.scripts_dir / "slow", "exec sleep 10")

    with pytest.raises(DevelopmentEnvironmentError, match="exceeded 1 seconds"):
        await development.prepare(workspace)