nisse/tests/integration/e2e_support.py

"""Shared by the live tests: each run gets a repository of its own, made by
the test operator and deleted afterwards, so runs never depend on one
another or on what earlier runs merged."""

import base64
import contextlib
import os
import subprocess
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path

from nisse_dev.gitea import GiteaClient, GiteaError
from nisse_dev.llm import RoleMap

# The end-to-end tests run every role on a local model, which costs nothing
# per call; NISSE_E2E_MODEL overrides it (e.g. claude-haiku-4-5 for a quick,
# paid run). The no-thinking alias (LiteLLM: chat_template_kwargs
# enable_thinking false): with thinking, some answers ran to 6-8k tokens,
# minutes each, and stretched a test from 3.5 to 20 minutes.
E2E_MODEL = "qwen3.8-27b-nothink"


def e2e_roles() -> RoleMap:
    model = os.environ.get("NISSE_E2E_MODEL", "").strip() or E2E_MODEL
    return RoleMap(executor=model, planner=model, helper=model, expert=model)


def task(run: str) -> str:
    """A request no earlier run can have fulfilled."""
    return (
        f"Add a function vowels_{run}(text) to the sandbox package that returns "
        "how many of a, e, i, o, u (either case) the text contains, with tests."
    )


def dependent_task(run: str) -> str:
    """A request that builds on task(run): it can only be done after it."""
    return (
        f"Add a function consonants_{run}(text) to the sandbox package that "
        "returns how many letters in the text are not vowels, using "
        f"vowels_{run}, with tests."
    )


def operator_gitea(url: str) -> 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(url, token)


OWNER = "nisse-testing-operator"  # the test operator: owns every test repository
SEED = Path(__file__).parent / "seed"


@dataclass(frozen=True)
class FreshRepo:
    name: str  # owner/repo
    push_key: Path  # this repository's deploy key, private half


def make_repo(
    operator: GiteaClient, name: str, keys: Path, *, agent: str, seed: bool = True
) -> FreshRepo:
    """A repository as the operator would make a project's (D4, D8): seeded
    with a small Python package (or only a README), the ``agent`` account a
    collaborator with write access, the default branch protected (pushes only
    by the owner, an approval to merge), and a deploy key of its own, written
    under ``keys``. A failure part-way deletes what was made."""
    full = f"{OWNER}/{name}"
    operator._request(
        "POST",
        "/user/repos",
        json={
            "name": name,
            "private": True,
            "auto_init": True,
            "default_branch": "main",
        },
    )
    try:
        files = [
            {
                "operation": "create",
                "path": str(path.relative_to(SEED)),
                "content": base64.b64encode(path.read_bytes()).decode(),
            }
            for path in sorted(SEED.rglob("*"))
            if seed and path.is_file() and path.name != "README.md"
        ]
        if files:
            operator._request(
                "POST",
                f"/repos/{full}/contents",
                json={
                    "branch": "main",
                    "message": "Seed the test project",
                    "files": files,
                },
            )
        readme = operator._request("GET", f"/repos/{full}/contents/README.md")
        # A new project's README says only what it is; a seeded one, the seed's.
        text = (SEED / "README.md").read_bytes() if seed else f"# {name}\n".encode()
        operator._request(
            "PUT",
            f"/repos/{full}/contents/README.md",
            json={
                "branch": "main",
                "message": "README",
                "sha": readme["sha"],
                "content": base64.b64encode(text).decode(),
            },
        )
        operator._request(
            "PUT", f"/repos/{full}/collaborators/{agent}", json={"permission": "write"}
        )
        keys.mkdir(parents=True, exist_ok=True)
        key = keys / f"{name}.key"
        key.unlink(missing_ok=True)
        subprocess.run(
            ["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", name, "-f", str(key)],
            check=True,
        )
        operator._request(
            "POST",
            f"/repos/{full}/keys",
            json={
                "title": f"{name} push",
                "key": key.with_suffix(".key.pub").read_text().strip(),
                "read_only": False,
            },
        )
        operator._request(
            "POST",
            f"/repos/{full}/branch_protections",
            # Pushes only by the operator (who sets up fixtures, as an owner
            # can); the agent's are refused, and merging needs an approval.
            json={
                "branch_name": "main",
                "enable_push": True,
                "enable_push_whitelist": True,
                "push_whitelist_usernames": [OWNER],
                "required_approvals": 1,
            },
        )
        # The labels the agent reads (D1), as a project would have them.
        for label, color in (
            ("bug", "#ee0701"),
            ("discuss", "#0e8a16"),
            ("design", "#5319e7"),
        ):
            operator._request(
                "POST", f"/repos/{full}/labels", json={"name": label, "color": color}
            )
    except BaseException:
        delete_repo(operator, full)
        raise
    return FreshRepo(full, key)


def delete_repo(operator: GiteaClient, full: str) -> None:
    with contextlib.suppress(GiteaError):
        operator._request("DELETE", f"/repos/{full}")


@contextlib.contextmanager
def fresh_repo(
    operator: GiteaClient, run: str, keys: Path, *, agent: str, seed: bool = True
) -> Iterator[FreshRepo]:
    """A repository of its own for one test run (make_repo), deleted
    afterwards unless NISSE_E2E_KEEP_REPOS is set."""
    repo = make_repo(operator, f"nisse-e2e-{run}", keys, agent=agent, seed=seed)
    try:
        yield repo
    finally:
        if not os.environ.get("NISSE_E2E_KEEP_REPOS"):
            delete_repo(operator, repo.name)