nisse/nisse-dev/tests/test_loop_new_projects.py

"""New projects on the new loop (M2.5, PR B): a conversation proposes one,
the agent lists what it needs, and on "done" reads its configuration again
and checks what was provided (D1, D4, D8) before opening the tracking issue
and starting on the architecture. Real git on the throwaway build host,
with protection enforced by a hook on the fake forge's bare repository."""

import io
import re
from pathlib import Path
from typing import Any

import pytest
import yaml
from test_loop_milestones import DON, Forge, ScriptedModel, call, make_world
from test_loop_projects import Router

from nisse_dev.cancel import Cancel
from nisse_dev.config import _parse_projects
from nisse_dev.execution import SSHHost
from nisse_dev.loop.agent import LoopAgent, Project
from nisse_dev.loop.driver import Driver
from nisse_dev.loop.items import CHAT, GO_AHEAD, MERGE, PROVISION, 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

PROPOSAL = call(
    "propose_project",
    name="tally",
    purpose="A small command-line tally counter.",
    decisions="Python 3.12 with pytest; no services, secrets or extra egress.",
)


@pytest.fixture
def worlds(ssh_host: SSHHost, tmp_path: Path) -> dict[str, Any]:
    sandbox = make_world(ssh_host, tmp_path / "sandbox")
    tally = make_world(
        ssh_host, tmp_path / "tally", store=sandbox["store"], repo_name="don/tally"
    )
    sandbox["forge"].issues.clear()  # only the conversation here
    tally["forge"].issues.clear()
    return {"sandbox": sandbox, "tally": tally}


def agent_for(
    worlds: dict[str, Any], model: ScriptedModel, *, entry_added: list[bool]
) -> tuple[LoopAgent, InMemoryTransport]:
    """An agent that knows only the sandbox; tally's entry appears in its
    configuration once ``entry_added[0]`` is set."""
    transport = InMemoryTransport()
    cancel = Cancel()
    router = Router({w["repo_name"]: w["forge"] for w in worlds.values()})
    for w in worlds.values():
        w["milestones"]._gitea = router

    def runtime(name: str) -> Project:
        w = worlds[name]
        return Project(
            name=name,
            repo=w["repo_name"],
            workspace=Workspace(w["host"], w["repo"].root, timeout_s=30),
            milestones=w["milestones"],
        )

    def reload(known: set[str]) -> list[Project]:
        names = ["sandbox", "tally"] if entry_added[0] else ["sandbox"]
        return [runtime(n) for n in names if n not in known]

    agent = LoopAgent(
        worlds["sandbox"]["store"],
        transport,
        router,  # type: ignore[arg-type]
        projects=[runtime("sandbox")],
        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()),
        reload=reload,
    )
    transport.on_message(agent.on_message)
    return agent, transport


def conversation(worlds: dict[str, Any]) -> Any:
    [chat] = [
        i for i in worlds["sandbox"]["store"].work_items() if i.scope == "general"
    ]
    return chat


def proposed(
    worlds: Any, model: ScriptedModel, added: list[bool]
) -> tuple[Any, Any, str]:
    agent, transport = agent_for(worlds, model, entry_added=added)
    root = transport.receive(DON, "I want a little tally counter, as its own project.")
    assert transport.outbound[-1].text.startswith("rev-1: a new project, `tally`.")
    assert loop_of(conversation(worlds)).wait == GO_AHEAD
    transport.receive(DON, "go", thread=root)
    return agent, transport, root


def test_an_agreed_project_gets_a_checklist_with_its_entry(worlds: Any) -> None:
    _, transport, _ = proposed(worlds, ScriptedModel([PROPOSAL]), [False])
    loop = loop_of(conversation(worlds))
    assert (loop.wait, loop.new_project) == (PROVISION, "tally")
    checklist = transport.outbound[-1].text
    for need in ("repository", "Protection", "collaborator", "build VM", "deploy key"):
        assert need in checklist
    # The entry is valid configuration once its placeholders are filled in.
    entry = re.search(r"```yaml\n(.*?)\n```", checklist, re.DOTALL)
    assert entry is not None
    filled = (
        entry.group(1)
        .replace("<owner>/tally", "don/tally")
        .replace("<the build VM>", "build.test")
        .replace("<my SSH key for it>", "/k")
        .replace("<the deploy key's private half>", "/p")
    )
    errors: list[str] = []
    [project] = _parse_projects(yaml.safe_load(filled)["projects"], {}, errors)
    assert errors == [] and project.name == "tally" and project.repo == "don/tally"


def test_done_before_the_entry_is_added_says_so(worlds: Any) -> None:
    _, transport, root = proposed(worlds, ScriptedModel([PROPOSAL]), [False])
    transport.receive(DON, "done", thread=root)
    assert "I don't see `tally` in my configuration yet" in transport.outbound[-1].text
    assert loop_of(conversation(worlds)).new_project == "tally"


def test_what_isnt_set_up_is_named_and_nothing_starts(worlds: Any) -> None:
    added = [False]
    _, transport, root = proposed(worlds, ScriptedModel([PROPOSAL]), added)
    added[0] = True  # the entry is there, but tally's main isn't protected
    transport.receive(DON, "done", thread=root)
    said = transport.outbound[-1].text
    assert said.startswith("Not ready yet:")
    assert "main isn't protected" in said and "could merge my own PR" in said
    assert loop_of(conversation(worlds)).new_project == "tally"
    assert worlds["tally"]["forge"].issues == []
    # The check's PR was closed and its branch deleted.
    assert all(p["state"] == "closed" for p in worlds["tally"]["forge"].prs.values())


def test_a_checked_project_gets_its_record_and_starts_on_its_design(
    worlds: Any,
) -> None:
    added = [False]
    model = ScriptedModel(
        [PROPOSAL],
        [  # tally's first item: the architecture and the build plan
            call("start_milestone", name="architecture"),
            call("write_file", path="ARCHITECTURE.md", content="# tally\n"),
            call("write_file", path="docs/plans/tally.md", content="# plan\n"),
            call("finish_milestone", summary="architecture and plan", final="no"),
        ],
    )
    agent, transport, root = proposed(worlds, model, added)
    forge: Forge = worlds["tally"]["forge"]
    forge.protect()
    added[0] = True
    transport.receive(DON, "done", thread=root)

    assert "tally" in agent._projects
    [record] = forge.issues
    marker = re.search(r"<!-- nisse-item: ([\w-]+) -->", record.body)
    assert marker and "tally counter" in record.body
    work = worlds["sandbox"]["store"].work_item(marker.group(1))
    assert work.scope == "tally" and loop_of(work).started
    assert "design" in loop_of(work).labels
    assert loop_of(work).wait == MERGE
    pr = max(n for n, p in forge.prs.items() if p["state"] == "open")
    assert sorted(forge.files(pr)) == ["ARCHITECTURE.md", "docs/plans/tally.md"]
    assert f"Part of #{record.number}" in forge.prs[pr]["body"]
    chat = loop_of(conversation(worlds))
    assert chat.became == (f"don/tally#{record.number}",)
    assert (chat.wait, chat.new_project) == (CHAT, "")


def test_a_project_name_that_exists_is_sent_back(worlds: Any) -> None:
    model = ScriptedModel(
        [call("propose_project", name="sandbox", purpose="again", decisions="x.")],
        [call("reply", text="sandbox exists; shall I propose work on it?")],
    )
    agent, transport = agent_for(worlds, model, entry_added=[False])
    transport.receive(DON, "a new sandbox?")
    assert loop_of(conversation(worlds)).wait == CHAT
    assert agent