nisse/nisse-dev/tests/test_loop_preflight.py

"""Toolchain preflight in the loop (D3, M4.1): before the driver plans, the
build VM is checked (baseline, clone, marker); without a marker the item's
driver gets the toolchain ready and the harness checks what it reports; later
runs check the versions again; a rebuilt VM is re-cloned and set up again,
and a changed host key waits for the operator."""

import io
import shutil
import uuid
from pathlib import Path
from typing import Any

import pytest
from test_loop_milestones import (
    DON,
    FINISH,
    RESTATE,
    WORK,
    ScriptedModel,
    call,
    say,
)
from test_loop_milestones import item as the_item
from test_loop_milestones import world as world  # noqa: PLC0414 -- the fixture

from nisse_dev.cancel import Cancel
from nisse_dev.canonical import Fact, ToolResult
from nisse_dev.execution import HostKeyChanged
from nisse_dev.loop import preflight
from nisse_dev.loop.agent import LoopAgent, Project
from nisse_dev.loop.driver import Driver
from nisse_dev.loop.items import ANSWER, GO_AHEAD, WAITING, loop_of
from nisse_dev.oplog import OpLog
from nisse_dev.tools import Workspace
from nisse_dev.transport import InMemoryTransport
from nisse_dev.worker import InlineWorker

PYTHON = {"name": "python", "version_command": "python3 --version"}


def ready(**more: Any) -> list[Any]:
    """A preflight that finds Python and reports it."""
    return [
        call("run_command", command="python3 --version"),
        call(
            "toolchain_ready",
            summary="Python is there",
            toolchain=[PYTHON],  # type: ignore[arg-type]
            **more,
        ),
    ]


@pytest.fixture
def project() -> str:
    return f"pf-{uuid.uuid4().hex[:8]}"


def markers(world: dict[str, Any]) -> str:
    """Markers in the test's own directory, not this host's home."""
    return str(Path(world["repo"].root).parent / "markers")


def agent_for(
    world: dict[str, Any], project: str, model: ScriptedModel
) -> tuple[LoopAgent, InMemoryTransport]:
    transport = InMemoryTransport()
    cancel = Cancel()
    loop = LoopAgent(
        world["store"],
        transport,
        world["forge"],
        projects=[
            Project(
                project,
                "don/sandbox",
                Workspace(world["host"], world["repo"].root, timeout_s=30),
                world["milestones"],
            )
        ],
        account="nisse-dev",
        operators={DON},
        gitea_operators={"don"},
        driver=Driver(model, "m", cancel=cancel),  # type: ignore[arg-type]
        worker=InlineWorker(),
        cancel=cancel,
        oplog=OpLog("nisse-dev", io.StringIO()),
        preflight=True,
        preflight_turns=5,
        markers=markers(world),
    )
    transport.on_message(loop.on_message)
    return loop, transport


def facts(world: dict[str, Any]) -> list[str]:
    events = world["store"].transcript(the_item(world).id, "orchestrator").read()
    return [e.event.text for e in events if isinstance(e.event, Fact)]


def said(transport: InMemoryTransport) -> list[str]:
    return [m.text for m in transport.outbound]


def test_preflight_comes_before_planning(world: dict[str, Any], project: str) -> None:
    model = ScriptedModel(ready(), RESTATE)
    agent, transport = agent_for(world, project, model)
    agent.intake()
    item = the_item(world)
    assert (item.position, loop_of(item).wait) == (WAITING, GO_AHEAD)
    known = facts(world)
    toolchain = [
        f for f in known if f.startswith("preflight: toolchain: python Python 3.")
    ]
    assert toolchain, known
    assert known.index(
        "preflight: getting the project's toolchain ready"
    ) < known.index("restatement rev-1 put up for a go-ahead")
    assert loop_of(item).toolchain[0][:2] == ("python", "python3 --version")
    marker = preflight.check(
        world["host"], project, str(world["repo"].root), markers(world)
    ).marker
    assert marker is not None and marker["complete"]
    assert not [t for t in said(transport) if t.startswith("Preflight:")]  # quiet


def test_preflight_runs_once_then_versions_are_checked(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    model = ScriptedModel(ready(), RESTATE, [*WORK, *FINISH])
    agent, transport = agent_for(world, project, model)
    agent.intake()
    real = preflight.versions

    def newer(host: Any, tools: list[Any]) -> list[Any]:
        return [
            preflight.Tool(t.name, t.command, "Python 3.99.0")
            for t in real(host, tools)
        ]

    monkeypatch.setattr(preflight, "versions", newer)
    say(transport, world, "go")  # no second preflight: the script has none
    changed = [f for f in facts(world) if f.startswith("the toolchain changed")]
    assert changed and "→ Python 3.99.0" in changed[0]
    assert any("toolchain changed" in t for t in said(transport))
    [pr] = [n for n in world["forge"].prs]
    assert "Tested with: python Python 3.99.0" in world["forge"].prs[pr]["body"]


def test_only_version_commands_are_accepted(
    world: dict[str, Any], project: str
) -> None:
    bad = call(
        "toolchain_ready",
        summary="s",
        toolchain=[{"name": "py", "version_command": "python3 -c 'print(1)'"}],  # type: ignore[arg-type]
    )
    model = ScriptedModel([bad], ready(), RESTATE)
    agent, _ = agent_for(world, project, model)
    agent.intake()
    results = [
        e.event.result
        for e in world["store"].transcript(the_item(world).id, "driver").read()
        if isinstance(e.event, ToolResult)
    ]
    assert any("plain version commands" in r for r in results)
    assert loop_of(the_item(world)).toolchain


def test_what_was_installed_or_is_missing_is_said(
    world: dict[str, Any], project: str
) -> None:
    model = ScriptedModel(
        ready(
            installed=["libfoo-dev"],  # type: ignore[arg-type]
            missing=[{"name": "protoc", "why": "no package for this OS"}],  # type: ignore[arg-type]
        ),
        RESTATE,
    )
    agent, transport = agent_for(world, project, model)
    agent.intake()
    [line] = [t for t in said(transport) if t.startswith("Preflight:")]
    assert "installed: libfoo-dev" in line
    assert "missing: protoc (no package for this OS)" in line


def test_an_unfinished_preflight_is_said_and_work_goes_on(
    world: dict[str, Any], project: str
) -> None:
    from nisse_dev.canonical import AssistantText

    prose = [AssistantText(text="hmm", model="m")]
    model = ScriptedModel(prose, prose, RESTATE)
    agent, transport = agent_for(world, project, model)
    agent.intake()
    assert loop_of(the_item(world)).wait == GO_AHEAD  # it planned anyway
    assert any("the preflight didn't finish" in t for t in said(transport))
    marker = preflight.check(
        world["host"], project, str(world["repo"].root), markers(world)
    ).marker
    assert marker is not None and not marker["complete"]


def test_a_wiped_vm_is_cloned_again_and_set_up(
    world: dict[str, Any], project: str
) -> None:
    model = ScriptedModel(ready(), RESTATE, ready(), [*WORK, *FINISH])
    agent, transport = agent_for(world, project, model)
    agent.intake()
    # The VM is rebuilt: the clone and the marker are gone.
    shutil.rmtree(world["repo"].root)
    world["host"].execute(f'rm -f "{preflight.marker_path(project, markers(world))}"')
    say(transport, world, "go")
    assert any("cloned the workspace again" in f for f in facts(world))
    assert (Path(world["repo"].root) / ".git").exists()
    assert sum("getting the project's toolchain ready" in f for f in facts(world)) == 2
    assert world["forge"].prs  # and the work went on to a PR


def test_a_missing_baseline_waits_on_the_operator(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(preflight, "BASELINE", ("git", "no-such-tool-xyz"))
    agent, _ = agent_for(world, project, ScriptedModel())
    agent.intake()
    loop = loop_of(the_item(world))
    assert loop.wait == ANSWER and "lacks no-such-tool-xyz" in loop.note


def test_a_changed_host_key_waits_for_the_operators_word(
    world: dict[str, Any], project: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    real = preflight.check
    changed = {"on": True}

    def check(host: Any, name: str, root: str, *rest: Any) -> Any:
        if changed["on"]:
            raise HostKeyChanged("the key changed")
        return real(host, name, root, *rest)

    forgotten: list[str] = []
    monkeypatch.setattr(preflight, "check", check)
    monkeypatch.setattr(
        type(world["host"]), "forget_host_key", lambda self: forgotten.append("yes")
    )
    agent, transport = agent_for(world, project, ScriptedModel(ready(), RESTATE))
    agent.intake()
    loop = loop_of(the_item(world))
    assert loop.host_key_changed and "trust new host key" in loop.note
    say(transport, world, "it's fine")  # an answer isn't the operator's word
    still = loop_of(the_item(world))
    assert (still.wait, still.host_key_changed, forgotten) == (ANSWER, True, [])
    changed["on"] = False
    say(transport, world, "trust new host key")
    assert forgotten == ["yes"]
    assert loop_of(the_item(world)).wait == GO_AHEAD  # set up, then planned