nisse/tests/integration/test_end_to_end_loop.py

"""The end-to-end test on the new loop (Development Plan M2.1, PR D): the
whole slice, from an issue assigned in Gitea to a merged PR and a closed work
item, with real models, the Gitea sandbox and a real build host.

A scripted operator plays the operator's part: in Gitea through its own test
account (NISSE_OPERATOR_TOKEN; an operator only in nisse-test.yaml), and in
the item's Matrix thread through the in-memory transport. Run with
scripts/e2e.sh after each milestone.
"""

import contextlib
import os
import time
import uuid
from collections.abc import Iterator
from dataclasses import replace
from fnmatch import fnmatch
from pathlib import Path
from typing import Any

import pytest

from nisse_dev.config import AgentConfig
from nisse_dev.execution import SSHHost
from nisse_dev.gitea import GiteaClient, GiteaError
from nisse_dev.loop.build import RunningLoop, build_loop_agent
from nisse_dev.loop.items import ANSWER, GO_AHEAD, MERGE, WAITING, loop_of
from nisse_dev.transport import InMemoryTransport

pytestmark = [pytest.mark.integration, pytest.mark.e2e]

OPERATOR = "@dongar:dongar.net"  # an allowlisted Matrix operator in nisse-test.yaml
TASK = (
    "Add a function count_vowels(text) to the sandbox package that returns how "
    "many of a, e, i, o, u (either case) the text contains, with tests."
)
DEADLINE_S = 1200


def operator_gitea(config: AgentConfig) -> GiteaClient:
    token = os.environ.get("NISSE_OPERATOR_TOKEN", "")
    assert token.strip(), (
        "NISSE_OPERATOR_TOKEN is not set: the test operator's Gitea token"
    )
    return GiteaClient(config.gitea_url, token)


@pytest.fixture
def running(
    live_config: AgentConfig,
    ssh_host: SSHHost,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> Iterator[tuple[RunningLoop, InMemoryTransport]]:
    from test_end_to_end import build_host

    run = uuid.uuid4().hex[:8]
    config = replace(
        live_config,
        loop="new",
        state_db=tmp_path / "state.db",
        workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"),
        build_host=build_host(ssh_host),
    )
    transport = InMemoryTransport()
    monkeypatch.setattr("nisse_dev.loop.build.MatrixTransport", lambda cfg: transport)
    agent = None
    try:
        agent = build_loop_agent(config)
        # The test drives intake and PR watching itself, not on a timer.
        agent.loop._interval = 3600
        agent.start()
        yield agent, transport
    finally:
        if agent is not None:
            agent.stop()
        host = SSHHost(
            config.build_host.address,
            key=config.build_host.key,
            user=config.build_host.user,
            port=config.build_host.port,
            state_dir=tmp_path / "ssh",
        )
        host.execute(f"rm -rf /tmp/nisse-e2e-{run}")


def is_artifact(path: str) -> bool:
    parts = path.split("/")
    return (
        "__pycache__" in parts
        or ".pytest_cache" in parts
        or fnmatch(parts[-1], "*.py[co]")
    )


class Operator:
    def __init__(self, agent: RunningLoop, transport: InMemoryTransport) -> None:
        self.agent = agent
        self.transport = transport

    def settle(self) -> None:
        self.agent.worker.call(lambda: None, timeout=DEADLINE_S)

    def tick(self) -> None:
        """Intake and PR watching, as the agent's timer would do."""
        self.agent.worker.call(self.agent.loop.intake, timeout=DEADLINE_S)

    def item(self) -> Any:
        items = self.agent.worker.call(self.agent.store.work_items)
        return items[0] if items else None

    def say(self, text: str) -> None:
        self.transport.receive(OPERATOR, text, thread=loop_of(self.item()).thread)
        self.settle()

    def drive_until(self, wait: str) -> Any:
        """Answer whatever the agent asks until it waits for ``wait``."""
        deadline = time.monotonic() + DEADLINE_S
        while time.monotonic() < deadline:
            item = self.item()
            assert item is not None, "no work item"
            loop = loop_of(item)
            if item.position == WAITING and loop.wait == wait:
                return item
            if item.status == "closed":
                raise AssertionError(f"closed early: {self.last()}")
            if item.position == WAITING and loop.wait == GO_AHEAD:
                self.say("go")
            elif item.position == WAITING and loop.wait == ANSWER:
                self.say("Use your best judgement and proceed.")
            else:
                self.settle()
        raise AssertionError(f"never waited for {wait}: {self.last()}")

    def last(self) -> str:
        return self.transport.outbound[-1].text if self.transport.outbound else ""


def test_the_slice_end_to_end(running: tuple[RunningLoop, InMemoryTransport]) -> None:
    agent, transport = running
    operator = Operator(agent, transport)
    gitea, repo = agent.gitea, agent.config.gitea_repo
    as_operator = operator_gitea(agent.config)
    issue = None
    pr = 0
    try:
        # --- the operator files an issue and assigns it to the agent (D1) ---
        issue = as_operator.create_issue(
            repo, f"count_vowels ({uuid.uuid4().hex[:6]})", TASK
        )
        as_operator._request(
            "PATCH",
            f"/repos/{repo}/issues/{issue}",
            json={"assignees": [agent.account]},
        )
        operator.tick()
        item = operator.item()
        assert item is not None and loop_of(item).issue == issue

        # --- restatement, go-ahead, milestone, PR (L3, D4) ---
        item = operator.drive_until(MERGE)
        pr = loop_of(item).pr
        files = gitea._request("GET", f"/repos/{repo}/pulls/{pr}/files")
        changed = [f["filename"] for f in files]
        assert changed and not [p for p in changed if is_artifact(p)], changed
        body = gitea._request("GET", f"/repos/{repo}/pulls/{pr}")["body"]
        assert f"Closes #{issue}" in body

        # --- a review comment is answered with a fix (D5) ---
        before = gitea._request("GET", f"/repos/{repo}/pulls/{pr}")["head"]["sha"]
        as_operator.post_comment(
            repo, pr, "Please add a one-line docstring to count_vowels."
        )
        operator.tick()
        operator.drive_until(MERGE)
        after = gitea._request("GET", f"/repos/{repo}/pulls/{pr}")["head"]["sha"]
        assert after != before, "no fix was pushed for the review comment"

        # --- the operator approves and merges; the agent closes the item ---
        with contextlib.suppress(GiteaError):  # approval may not be required
            as_operator._request(
                "POST",
                f"/repos/{repo}/pulls/{pr}/reviews",
                json={"event": "APPROVED", "body": "looks good"},
            )
        as_operator._request(
            "POST", f"/repos/{repo}/pulls/{pr}/merge", json={"Do": "merge"}
        )
        operator.tick()
        deadline = time.monotonic() + DEADLINE_S
        while operator.item().status != "closed" and time.monotonic() < deadline:
            if loop_of(operator.item()).wait == ANSWER:
                operator.say("It's merged; please close the item.")
            operator.settle()
        assert operator.item().outcome == "completed"
        state = gitea._request("GET", f"/repos/{repo}/issues/{issue}")["state"]
        assert state == "closed"
    finally:
        # Leave the sandbox tidy: whatever is still open gets closed.
        item = operator.item()
        if item is not None and item.status == "open":
            operator.say("abandon")
        if issue is not None:
            with contextlib.suppress(GiteaError):
                as_operator.close_issue(repo, issue)
        if pr:
            with contextlib.suppress(GiteaError):
                head = gitea._request("GET", f"/repos/{repo}/pulls/{pr}")["head"]["ref"]
                as_operator._request("DELETE", f"/repos/{repo}/branches/{head}")