A repository per test run instead of the shared sandbox

Every live test that touches a repository gets one of its own from
e2e_support.fresh_repo, made by the test operator as an operator makes a
project's (D4, D8): seeded from tests/integration/seed/ (the sandbox's
package, without past runs' leftovers), the agent's account a collaborator
with write access, main protected (pushes only by the owner, an approval to
merge), and a deploy key generated for the run. It is deleted afterwards
(NISSE_E2E_KEEP_REPOS keeps it); scripts/e2e-cleanup.sh deletes leftovers.

- Both e2e fixtures use it; per-run base branches are gone. The merge
  helper approves first, as the protection now requires.
- test_gitea_live uses it, so CI needs NISSE_OPERATOR_TOKEN and no longer
  writes the sandbox deploy key (NISSE_PUSH_KEY).
- A live test of the helper: seeded, shared with the agent, protected
  against it, deleted.
- nisse-test.yaml's project entry is a placeholder; pytest and ruff skip
  the seed.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE

tests/integration/e2e_support.py

@@ -1,10 +1,14 @@
-"""Shared by the end-to-end tests: each run gets work that can't already be
-done, and a base branch of its own, so runs never depend on one another or
-on what earlier runs merged."""
+"""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
 
@@ -34,18 +38,97 @@ def operator_gitea(url: str) -> GiteaClient:
     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 base_branch(operator: GiteaClient, repo: str, run: str) -> Iterator[str]:
-    """A branch off main for this run's PRs to target and merge into, so
-    nothing accumulates in main; deleted afterwards."""
-    name = f"e2e-base-{run}"
+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",
-        f"/repos/{repo}/branches",
-        json={"new_branch_name": name, "old_branch_name": "main"},
+        "/user/repos",
+        json={
+            "name": name,
+            "private": True,
+            "auto_init": True,
+            "default_branch": "main",
+        },
     )
     try:
-        yield name
+        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:
-        with contextlib.suppress(GiteaError):
-            operator._request("DELETE", f"/repos/{repo}/branches/{name}")
+        if not os.environ.get("NISSE_E2E_KEEP_REPOS"):
+            with contextlib.suppress(GiteaError):
+                operator._request("DELETE", f"/repos/{full}")

tests/integration/nisse-test.yaml

@@ -13,17 +13,16 @@ state_db: /tmp/nisse-test/state.db
 
 projects:
   nisse-dev-testing:
-    repo: dongar/nisse-dev-testing-sandbox    # the sandbox, never a real project
-    # The live tests replace these with a throwaway local sshd (the `ssh_host`
-    # fixture) or the bring-up build VM, in temporary directories.
+    # Placeholders: every live test that touches a repository makes one of
+    # its own (tests/integration/e2e_support.py, fresh_repo), owned by the
+    # test operator and deleted afterwards, with its own deploy key; and runs
+    # on a throwaway local sshd (the `ssh_host` fixture) or the bring-up build
+    # VM, in temporary directories.
+    repo: nisse-testing-operator/made-per-run
     workspace: /tmp/nisse-test/work
     build_host:
       address: 127.0.0.1
       key: /tmp/nisse-test/unused-key
-    # What pushes go by (D4): the sandbox's deploy key, whose private half CI
-    # writes to this path from the NISSE_PUSH_KEY secret (locally: copy it
-    # there). An HTTPS push token works too: push_user plus push_token.
-    push_key: /tmp/nisse-test/push_key
 
 llm:
   base_url: https://litellm.dongar.net

tests/integration/seed/README.md

@@ -0,0 +1,4 @@
+# sandbox
+
+A small Python package for nisse-dev's end-to-end tests: each run gets a
+fresh repository seeded with this. Run its tests with `python3 -m pytest`.

tests/integration/seed/sandbox/__init__.py

@@ -0,0 +1,2 @@
+from .text import truncate
+from .mathutils import factorial

tests/integration/seed/sandbox/mathutils.py

@@ -0,0 +1,16 @@
+"""Math utilities."""
+
+
+def factorial(n: int) -> int:
+    """Return n! for a non-negative integer n.
+
+    Raises TypeError if n is not an int (bool is rejected), ValueError if n < 0.
+    """
+    if isinstance(n, bool) or not isinstance(n, int):
+        raise TypeError(f"n must be an int, got {type(n).__name__}")
+    if n < 0:
+        raise ValueError(f"n must be >= 0, got {n}")
+    result = 1
+    for i in range(2, n + 1):
+        result *= i
+    return result

tests/integration/seed/sandbox/slugify.py

@@ -0,0 +1,10 @@
+import re
+import unicodedata
+
+_NON_ALNUM = re.compile(r"[^a-z0-9]+")
+
+
+def slugify(text: str) -> str:
+    """Lowercase; runs of non-alphanumerics -> single '-'; no leading/trailing '-'."""
+    ascii_text = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode("ascii")
+    return _NON_ALNUM.sub("-", ascii_text.lower()).strip("-")

tests/integration/seed/sandbox/text.py

@@ -0,0 +1,16 @@
+"""Text utilities."""
+
+ELLIPSIS = "\u2026"
+
+
+def truncate(text: str, n: int) -> str:
+    """Return text unchanged if len(text) <= n, else its first n-1 chars followed by '…'.
+
+    Raises ValueError if n < 1.
+    """
+    if n < 1:
+        raise ValueError(f"n must be >= 1, got {n}")
+    if len(text) <= n:
+        return text
+    return text[: n - 1] + ELLIPSIS
+

tests/integration/seed/tests/test_factorial.py

@@ -0,0 +1,42 @@
+import math
+
+import pytest
+
+from sandbox import factorial
+
+
+@pytest.mark.parametrize(
+    "n, expected",
+    [
+        (0, 1),
+        (1, 1),
+        (2, 2),
+        (3, 6),
+        (5, 120),
+        (10, 3628800),
+        (20, 2432902008176640000),
+    ],
+)
+def test_factorial_known_values(n, expected):
+    assert factorial(n) == expected
+
+
+def test_factorial_matches_math_factorial():
+    for n in range(0, 101):
+        assert factorial(n) == math.factorial(n)
+
+
+def test_factorial_large_input_no_recursion_limit():
+    assert factorial(2000) == math.factorial(2000)
+
+
+@pytest.mark.parametrize("n", [-1, -5, -100])
+def test_factorial_negative_raises_value_error(n):
+    with pytest.raises(ValueError):
+        factorial(n)
+
+
+@pytest.mark.parametrize("n", [5.0, 2.5, "5", None, True, False])
+def test_factorial_non_int_raises_type_error(n):
+    with pytest.raises(TypeError):
+        factorial(n)

tests/integration/seed/tests/test_slugify.py

@@ -0,0 +1,25 @@
+import pytest
+
+from sandbox.slugify import slugify
+
+
+@pytest.mark.parametrize(
+    "text, expected",
+    [
+        ("Hello World", "hello-world"),
+        ("Hello, World!", "hello-world"),
+        ("  leading and trailing  ", "leading-and-trailing"),
+        ("multiple   spaces", "multiple-spaces"),
+        ("a--b__c..d", "a-b-c-d"),
+        ("!!!", ""),
+        ("", ""),
+        ("-already-slugged-", "already-slugged"),
+        ("MiXeD CaSe 123", "mixed-case-123"),
+        ("tab\tand\nnewline", "tab-and-newline"),
+        ("Café Déjà Vu", "cafe-deja-vu"),
+        ("emoji 🎉 party", "emoji-party"),
+        ("Foo Bar", "foo-bar"),
+    ],
+)
+def test_slugify(text, expected):
+    assert slugify(text) == expected

tests/integration/seed/tests/test_truncate.py

@@ -0,0 +1,37 @@
+import pytest
+
+from sandbox import truncate
+
+
+@pytest.mark.parametrize(
+    "text, n, expected",
+    [
+        ("", 1, ""),
+        ("abc", 3, "abc"),
+        ("abc", 10, "abc"),
+        ("abcd", 3, "ab\u2026"),
+        ("hello world", 5, "hell\u2026"),
+        ("ab", 1, "\u2026"),
+        ("a", 1, "a"),
+        ("h\u00e9llo w\u00f6rld", 4, "h\u00e9l\u2026"),
+    ],
+)
+def test_truncate_cases(text, n, expected):
+    assert truncate(text, n) == expected
+
+
+@pytest.mark.parametrize("n", range(1, 16))
+def test_result_never_exceeds_n(n):
+    assert len(truncate("hello world!!", n)) <= n
+
+
+def test_uses_single_ellipsis_char():
+    result = truncate("hello world", 5)
+    assert result.endswith("\u2026")
+    assert not result.endswith("...")
+
+
+@pytest.mark.parametrize("n", [0, -5])
+def test_invalid_n_raises(n):
+    with pytest.raises(ValueError):
+        truncate("abc", n)

tests/integration/test_end_to_end.py

@@ -25,7 +25,7 @@ from pathlib import Path
 from typing import Any
 
 import pytest
-from e2e_support import base_branch, operator_gitea
+from e2e_support import fresh_repo, operator_gitea
 from e2e_support import task as make_task
 
 from nisse_dev.agent import Agent, build_agent
@@ -38,7 +38,7 @@ from nisse_dev.control import (
     PR_OPEN,
 )
 from nisse_dev.execution import SSHHost
-from nisse_dev.gitea import GiteaError
+from nisse_dev.gitea import GiteaClient, GiteaError
 from nisse_dev.transport import InMemoryTransport
 
 pytestmark = [pytest.mark.integration, pytest.mark.e2e]
@@ -72,10 +72,15 @@ def agent(
     run = uuid.uuid4().hex[:8]
     stack = contextlib.ExitStack()
     operator = operator_gitea(live_config.gitea_url)
-    base = stack.enter_context(base_branch(operator, live_config.gitea_repo, run))
+    account = GiteaClient(live_config.gitea_url, live_config.gitea_token).whoami()
+    fresh = stack.enter_context(
+        fresh_repo(operator, run, tmp_path / "keys", agent=account)
+    )
     config = replace(
         live_config.with_project(
-            base_branch=base,
+            repo=fresh.name,
+            base_branch="main",
+            push_key=fresh.push_key,
             workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"),
             build_host=build_host(ssh_host),
         ),
@@ -100,7 +105,7 @@ def agent(
             state_dir=tmp_path / "ssh",
         )
         host.execute(f"rm -rf /tmp/nisse-e2e-{run}")
-        stack.close()  # the run's base branch
+        stack.close()  # the run's repository
 
 
 class Operator:

tests/integration/test_end_to_end_loop.py

@@ -21,7 +21,7 @@ from pathlib import Path
 from typing import Any
 
 import pytest
-from e2e_support import base_branch, dependent_task, operator_gitea, task
+from e2e_support import dependent_task, fresh_repo, operator_gitea, task
 
 from nisse_dev.config import AgentConfig, LoopTuning
 from nisse_dev.execution import SSHHost
@@ -63,10 +63,13 @@ def running(
     keep.mkdir(parents=True, exist_ok=True)
     print(f"\ne2e run {run}: state store in {keep}")
     operator = operator_gitea(live_config.gitea_url)
-    base = stack.enter_context(base_branch(operator, live_config.gitea_repo, run))
+    account = GiteaClient(live_config.gitea_url, live_config.gitea_token).whoami()
+    fresh = stack.enter_context(fresh_repo(operator, run, keep / "keys", agent=account))
     config = replace(
         live_config.with_project(
-            base_branch=base,
+            repo=fresh.name,
+            base_branch="main",
+            push_key=fresh.push_key,
             workspace=Path(f"/tmp/nisse-e2e-{run}/sandbox"),
             build_host=build_host(ssh_host),
         ),
@@ -95,7 +98,7 @@ def running(
             state_dir=tmp_path / "ssh",
         )
         host.execute(f"rm -rf /tmp/nisse-e2e-{run}")
-        stack.close()  # the run's base branch
+        stack.close()  # the run's repository
 
 
 def is_artifact(path: str) -> bool:
@@ -108,8 +111,14 @@ def is_artifact(path: str) -> bool:
 
 
 def merge(gitea: GiteaClient, repo: str, pr: int, wait_s: float = 60) -> None:
-    """Merge, waiting out Gitea's "Please try again later": it answers that
-    while it is still checking a freshly pushed PR can merge."""
+    """Approve and merge, waiting out Gitea's "Please try again later": it
+    answers that while it is still checking a freshly pushed PR can merge."""
+    with contextlib.suppress(GiteaError):  # approved already
+        gitea._request(
+            "POST",
+            f"/repos/{repo}/pulls/{pr}/reviews",
+            json={"event": "APPROVED", "body": "looks good"},
+        )
     deadline = time.monotonic() + wait_s
     while True:
         try:

tests/integration/test_fresh_repo_live.py

@@ -0,0 +1,33 @@
+"""The per-run test repository (e2e_support.fresh_repo): what the operator
+provides for a project (D4, D8), made and deleted by the test operator."""
+
+import uuid
+from pathlib import Path
+
+import pytest
+from e2e_support import fresh_repo, operator_gitea
+
+from nisse_dev.config import AgentConfig
+from nisse_dev.gitea import GiteaClient, GiteaError
+
+pytestmark = pytest.mark.integration
+
+
+def test_a_fresh_repository_is_seeded_shared_and_protected(
+    tmp_path: Path, live_config: AgentConfig, gitea: GiteaClient
+) -> None:
+    operator = operator_gitea(live_config.gitea_url)
+    run = uuid.uuid4().hex[:8]
+    with fresh_repo(operator, run, tmp_path / "keys", agent=gitea.whoami()) as repo:
+        # Seeded: the package and its tests, and the README replaced.
+        text = gitea._request("GET", f"/repos/{repo.name}/contents/sandbox/text.py")
+        assert text["name"] == "text.py"
+        assert gitea._request("GET", f"/repos/{repo.name}/contents/tests")
+        # The agent sees it as a collaborator that may push, and main as
+        # protected against it (what D4's verification reads).
+        main = gitea._request("GET", f"/repos/{repo.name}/branches/main")
+        assert main["protected"] and not main["user_can_push"]
+        assert gitea._request("GET", f"/repos/{repo.name}")["permissions"]["push"]
+        assert repo.push_key.exists()
+    with pytest.raises(GiteaError, match="404"):
+        gitea._request("GET", f"/repos/{repo.name}")

tests/integration/test_gitea_live.py

@@ -1,19 +1,18 @@
-"""Live Gitea round trip against the sandbox repo: push a branch, open a PR,
-comment, and prove the L8 reconciler reads external truth correctly.
-
-Cleans up after itself (closes the PR, deletes the branch) so repeated runs
-do not silt up the sandbox.
+"""Live Gitea round trip in a repository of the run's own: push a branch,
+open a PR, comment, and prove the L8 reconciler reads external truth
+correctly, and that protection holds against the agent (D4).
 """
 
 import uuid
 from pathlib import Path
 
 import pytest
+from e2e_support import fresh_repo, operator_gitea
 
-from nisse_dev.agent import push_credential, remote_url
+from nisse_dev.agent import remote_url
 from nisse_dev.config import AgentConfig
 from nisse_dev.execution import SSHHost
-from nisse_dev.git import GitError, GitRepo
+from nisse_dev.git import GitError, GitRepo, SSHKey
 from nisse_dev.gitea import GiteaClient, pull_request_reconciler
 from nisse_dev.state import Intent
 
@@ -34,9 +33,21 @@ def intent_for(branch: str) -> Intent:
 def test_push_open_pr_and_reconcile(
     tmp_path: Path, live_config: AgentConfig, gitea: GiteaClient, ssh_host: SSHHost
 ) -> None:
-    repo_path = live_config.gitea_repo  # the sandbox
     client = gitea
+    with fresh_repo(
+        operator_gitea(live_config.gitea_url),
+        uuid.uuid4().hex[:8],
+        tmp_path / "keys",
+        agent=client.whoami(),
+    ) as fresh:
+        _round_trip(
+            tmp_path, client, ssh_host, fresh.name, SSHKey(fresh.push_key.read_text())
+        )
 
+
+def _round_trip(
+    tmp_path: Path, client: GiteaClient, ssh_host: SSHHost, repo_path: str, key: SSHKey
+) -> None:
     account = client.whoami()
     assert "testing" in account, (
         f"live tests must run as a testing account, not {account}"
@@ -49,12 +60,12 @@ def test_push_open_pr_and_reconcile(
     assert reconcile(intent_for(branch)) is False
 
     # Pushed the way the agent does: from the build VM (here the throwaway
-    # sshd), with the push credential the config names, never the API token.
-    key = push_credential(live_config)
-    assert key is not None, "nisse-test.yaml names no push credential (D4)"
+    # sshd), with the repository's deploy key, never the API token.
     url = remote_url(client, repo_path, key)
     repo = GitRepo(ssh_host, tmp_path / "work", known_hosts=str(tmp_path / "kh"))
     repo.init()
+    repo._git("fetch", "-q", url, "main", credential=key)
+    repo._git("checkout", "-q", "-B", "work", "FETCH_HEAD")
     repo.configure_identity(account, f"{account}@example.test")
     (repo.root / f"{branch}.txt").write_text("live test artifact\n")
     assert repo.commit_all("Live test commit") is not None
@@ -93,5 +104,4 @@ def test_push_open_pr_and_reconcile(
     finally:
         if pr is not None:
             client.close_pull_request(repo_path, pr.number)
-        repo.delete_remote_branch(url, branch, credential=key)
         client.close()