nisse/nisse-review/tests/test_clones.py

"""The reviewer's code as kept git clones (R3): cloned once, fetched after,
checked out exactly at the PR's head, only the latest repository kept, and
git walled off from the code it fetches. A local bare repository stands in
for Gitea (``file://``: the configured URL's protocol is the one allowed)."""

import io
import json
import os
import subprocess
import threading
import time
from pathlib import Path
from typing import Any

import pytest

from nisse_core.models.interruptible import Stopped
from nisse_core.oplog import OpLog
from nisse_core.state.cancel import Cancel
from nisse_review.clones import CloneError, Clones
from nisse_review.reader import Repository

TOKEN = "s3cret-t0ken-value"
GIT_ENV = {
    "GIT_CONFIG_GLOBAL": os.devnull,
    "GIT_CONFIG_NOSYSTEM": "1",
    "GIT_AUTHOR_NAME": "t",
    "GIT_AUTHOR_EMAIL": "t@t",
    "GIT_COMMITTER_NAME": "t",
    "GIT_COMMITTER_EMAIL": "t@t",
    "PATH": os.environ["PATH"],
}


def git(cwd: Path, *args: str) -> str:
    return subprocess.run(
        ["git", *args], cwd=cwd, env=GIT_ENV, check=True, capture_output=True, text=True
    ).stdout.strip()


class Remote:
    """A Gitea stand-in: bare repositories under ``root``, and a work tree
    per repository to make commits in."""

    def __init__(self, tmp: Path) -> None:
        self.root = tmp / "gitea"
        self.url = f"file://{self.root}"
        self._tmp = tmp

    def repo(self, name: str, files: dict[str, str]) -> Path:
        bare = self.root / f"{name}.git"
        bare.mkdir(parents=True)
        git(bare, "init", "-q", "--bare", "-b", "main")
        work = self._tmp / "work" / name
        work.mkdir(parents=True)
        git(work, "init", "-q", "-b", "main")
        git(work, "remote", "add", "origin", str(bare))
        self.commit(work, files, "first")
        git(work, "push", "-q", "origin", "main")
        return work

    def commit(self, work: Path, files: dict[str, str | None], message: str) -> str:
        for path, text in files.items():
            if text is None:
                git(work, "rm", "-q", path)
                continue
            (work / path).parent.mkdir(parents=True, exist_ok=True)
            (work / path).write_text(text)
            git(work, "add", path)
        git(work, "commit", "-q", "-m", message)
        return git(work, "rev-parse", "HEAD")

    def pull(self, work: Path, number: int, files: dict[str, str | None]) -> str:
        """A PR's new head, as Gitea keeps it: refs/pull/N/head."""
        git(work, "checkout", "-q", "-B", f"pr{number}", "main")
        sha = self.commit(work, files, f"pr {number}")
        git(work, "push", "-q", "-f", "origin", f"HEAD:refs/pull/{number}/head")
        git(work, "checkout", "-q", "main")
        return sha


def setup(tmp_path: Path, **more: Any) -> tuple[Remote, Clones, io.StringIO]:
    out = io.StringIO()
    remote = Remote(tmp_path)
    clones = Clones(
        tmp_path / "repos", remote.url, TOKEN, oplog=OpLog("t", out), **more
    )
    return remote, clones, out


def events(out: io.StringIO, name: str) -> list[dict[str, Any]]:
    found = (json.loads(line) for line in out.getvalue().splitlines())
    return [e for e in found if e["event"] == name]


def test_a_first_read_clones_and_checks_out_the_head(tmp_path: Path) -> None:
    remote, clones, out = setup(tmp_path)
    work = remote.repo("o/demo", {"count.py": "old\n", "gone.py": "x\n"})
    head = remote.pull(work, 7, {"count.py": "new\n", "gone.py": None})
    tree = clones.open("o/demo", 7, head, "main").tree("head")
    assert tree == tmp_path / "repos" / "o" / "demo" / "head"
    assert (tree / "count.py").read_text() == "new\n"
    assert not (tree / "gone.py").exists()
    assert git(tree, "rev-parse", "HEAD") == head
    [cloned] = events(out, "review.clone")
    assert cloned["cause"] == "o/demo" and "seconds" in cloned
    assert not (tree.parent / "base").exists()  # only made when read


def test_a_later_run_fetches_and_leaves_nothing_of_the_last(tmp_path: Path) -> None:
    remote, clones, out = setup(tmp_path)
    work = remote.repo("o/demo", {"count.py": "one\n"})
    first = remote.pull(work, 7, {"extra.py": "x\n"})
    tree = clones.open("o/demo", 7, first, "main").tree("head")
    (tree / "junk.txt").write_text("left by the last run\n")
    (tree / "count.py").write_text("edited in place\n")
    second = remote.pull(work, 8, {"count.py": "two\n"})
    again = clones.open("o/demo", 8, second, "main").tree("head")
    assert again == tree
    assert (tree / "count.py").read_text() == "two\n"
    assert not (tree / "junk.txt").exists() and not (tree / "extra.py").exists()
    assert len(events(out, "review.clone")) == 1
    assert len(events(out, "review.fetch")) == 1


def test_another_repository_replaces_the_last(tmp_path: Path) -> None:
    remote, clones, out = setup(tmp_path)
    a = remote.repo("o/a", {"a.py": "a\n"})
    b = remote.repo("p/b", {"b.py": "b\n"})
    clones.open("o/a", 1, remote.pull(a, 1, {"a2.py": "a\n"}), "main").tree("head")
    clones.open("p/b", 2, remote.pull(b, 2, {"b2.py": "b\n"}), "main").tree("head")
    assert sorted(p.name for p in (tmp_path / "repos").iterdir()) == ["p"]
    assert (tmp_path / "repos" / "p" / "b" / "head" / "b2.py").is_file()
    [removed] = events(out, "review.clone_removed")
    assert removed["path"].endswith("o/a")


def test_the_base_is_a_worktree_at_its_branch(tmp_path: Path) -> None:
    remote, clones, _ = setup(tmp_path)
    work = remote.repo("o/demo", {"count.py": "base\n"})
    head = remote.pull(work, 7, {"count.py": "head\n"})
    checkout = clones.open("o/demo", 7, head, "main")
    base = checkout.tree("base")
    assert base == tmp_path / "repos" / "o" / "demo" / "base"
    assert (base / "count.py").read_text() == "base\n"
    assert (checkout.tree("head") / "count.py").read_text() == "head\n"
    remote.commit(work, {"count.py": "base moved\n"}, "on main")
    git(work, "push", "-q", "origin", "main")
    later = clones.open("o/demo", 7, head, "main")
    assert (later.tree("base") / "count.py").read_text() == "base moved\n"


def test_a_link_arrives_as_a_plain_file(tmp_path: Path) -> None:
    remote, clones, _ = setup(tmp_path)
    work = remote.repo("o/demo", {"ok.py": "x\n"})
    git(work, "checkout", "-q", "-B", "pr7")
    os.symlink("/etc/passwd", work / "link")
    git(work, "add", "link")
    git(work, "commit", "-q", "-m", "a link")
    git(work, "push", "-q", "-f", "origin", "HEAD:refs/pull/7/head")
    head = git(work, "rev-parse", "HEAD")
    tree = clones.open("o/demo", 7, head, "main").tree("head")
    assert not (tree / "link").is_symlink()
    assert (tree / "link").read_text() == "/etc/passwd"


def test_a_head_no_longer_there_fails_the_run(tmp_path: Path) -> None:
    remote, clones, _ = setup(tmp_path)
    work = remote.repo("o/demo", {"ok.py": "x\n"})
    remote.pull(work, 7, {"a.py": "a\n"})
    with pytest.raises(CloneError, match="force-pushed"):
        clones.open("o/demo", 7, "0" * 40, "main").tree("head")


def test_the_token_is_never_on_a_command_line_or_on_disk(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    remote, clones, _ = setup(tmp_path)
    work = remote.repo("o/demo", {"ok.py": "x\n"})
    head = remote.pull(work, 7, {"a.py": "a\n"})
    seen: list[tuple[list[str], dict[str, str]]] = []
    real = subprocess.Popen

    def watched(args: list[str], **kwargs: Any) -> Any:
        seen.append((list(args), dict(kwargs["env"])))
        return real(args, **kwargs)

    monkeypatch.setattr(subprocess, "Popen", watched)
    checkout = clones.open("o/demo", 7, head, "main")
    checkout.tree("head")
    checkout.tree("base")
    assert seen and all(TOKEN not in " ".join(args) for args, _ in seen)
    headers = [v for _, env in seen for v in env.values() if TOKEN in v]
    assert headers and all(v == f"Authorization: token {TOKEN}" for v in headers)
    for path in (tmp_path / "repos").rglob("*"):
        if path.is_file():
            assert TOKEN.encode() not in path.read_bytes(), path


def test_a_stop_ends_a_clone_and_removes_it(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    # A git whose fetch hangs, as a huge first clone would.
    fake = tmp_path / "bin"
    fake.mkdir()
    real_git = subprocess.run(
        ["which", "git"], capture_output=True, text=True, check=True
    ).stdout.strip()
    (fake / "git").write_text(
        f'#!/bin/sh\n[ "$1" = fetch ] && exec sleep 30\nexec {real_git} "$@"\n'
    )
    (fake / "git").chmod(0o755)
    cancel = Cancel()
    remote, clones, _ = setup(tmp_path, cancel=cancel)
    work = remote.repo("o/demo", {"ok.py": "x\n"})
    head = remote.pull(work, 7, {"a.py": "a\n"})
    monkeypatch.setenv("PATH", f"{fake}:{os.environ['PATH']}")
    threading.Timer(0.3, cancel.set).start()
    started = time.monotonic()
    with pytest.raises(Stopped, match="git fetch"):
        clones.open("o/demo", 7, head, "main").tree("head")
    assert time.monotonic() - started < 5
    assert not (tmp_path / "repos" / "o" / "demo").exists()  # made afresh next


def test_the_reader_never_shows_gits_own_folder(tmp_path: Path) -> None:
    remote, clones, _ = setup(tmp_path)
    work = remote.repo("o/demo", {"count.py": "split here\n"})
    head = remote.pull(work, 7, {"a.py": "a\n"})
    repo = Repository(None, "o/demo", clones.open("o/demo", 7, head, "main"))  # type: ignore[arg-type]
    read = repo.dispatch("read_file", {"path": ".git/config"})
    assert not read.success
    listed = repo.dispatch("list_files", {"path": ""})
    assert listed.success and ".git" not in listed.result
    assert "a.py" in listed.result and "count.py" in listed.result
    found = repo.dispatch("search_code", {"text": "origin"})
    assert "nothing" in found.result  # .git/config names it; never searched