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


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


@contextlib.contextmanager
def fresh_repo(
    operator: GiteaClient, run: str, keys: Path, *, agent: str
) -> Iterator[FreshRepo]:
    """A repository of its own for one test run, as the operator would make a
    project's (D4, D8): seeded with a small Python package, 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``. Deleted afterwards, unless NISSE_E2E_KEEP_REPOS is set."""
    name = f"nisse-e2e-{run}"
    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 path.is_file() and path.name != "README.md"
        ]
        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")
        operator._request(
            "PUT",
            f"/repos/{full}/contents/README.md",
            json={
                "branch": "main",
                "message": "README",
                "sha": readme["sha"],
                "content": base64.b64encode((SEED / "README.md").read_bytes()).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,
            },
        )
        yield FreshRepo(full, key)
    finally:
        if not os.environ.get("NISSE_E2E_KEEP_REPOS"):
            with contextlib.suppress(GiteaError):
                operator._request("DELETE", f"/repos/{full}")