nisse/nisse-review/tests/review_fakes.py

"""A Gitea and a model to review against, in memory."""

import io
import tarfile
from dataclasses import dataclass, field, replace
from typing import Any

from nisse_core.gitea import (
    Comment,
    GiteaError,
    GiteaNotFound,
    Issue,
    LineComment,
    PullRequest,
    Review,
    ReviewNote,
)
from nisse_core.log.canonical import CanonicalEvent, ToolCall

FILES = {
    "README.md": "# demo\n\nCount words.\n",
    "count.py": "def count(text):\n    return len(text.split())\n",
    "test_count.py": "from count import count\n\n\ndef test():\n    assert count('a b') == 2\n",
}
DIFF = """diff --git a/count.py b/count.py
--- a/count.py
+++ b/count.py
@@ -1,2 +1,2 @@
 def count(text):
-    return len(text)
+    return len(text.split())
"""


def archive(files: dict[str, str], *, top: str = "demo") -> bytes:
    out = io.BytesIO()
    with tarfile.open(fileobj=out, mode="w:gz") as tar:
        for path, text in files.items():
            raw = text.encode()
            info = tarfile.TarInfo(f"{top}/{path}")
            info.size = len(raw)
            tar.addfile(info, io.BytesIO(raw))
    return out.getvalue()


def pr(number: int = 7, **more: Any) -> PullRequest:
    base = PullRequest(
        number=number,
        state="open",
        head="fix-count",
        url=f"https://gitea.test/o/demo/pulls/{number}",
        head_sha="abc123def456",
        title="Count words, not characters",
        body="Fixes #3.",
        author="nisse-dev",
        base="main",
    )
    return replace(base, **more)


@dataclass
class FakeGitea:
    account: str = "nisse-review"
    repos: dict[str, list[PullRequest]] = field(default_factory=dict)
    merged: set[tuple[str, int]] = field(default_factory=set)
    permissions: dict[str, dict[str, bool]] = field(default_factory=dict)
    files: dict[str, str] = field(default_factory=lambda: dict(FILES))
    diff: str = DIFF
    posted: list[dict[str, Any]] = field(default_factory=list)
    comments: list[tuple[str, int, str]] = field(default_factory=list)
    refuse_off_diff: bool = False
    down: bool = False
    # Readable but not given (public, no grant): missing from /user/repos.
    public: set[str] = field(default_factory=set)
    unlisted: set[str] = field(default_factory=set)  # its PRs can't be listed
    ci: dict[str, str] = field(default_factory=dict)  # head -> checks; success
    # What others said: top-level comments and comments on lines (M7.5).
    thread: list[Comment] = field(
        default_factory=lambda: [Comment(1, "don", "Please keep it simple.", "t")]
    )
    notes: list[ReviewNote] = field(default_factory=list)
    since: list[str] = field(default_factory=list)  # commits_between's answer
    reactions: list[tuple[int, str]] = field(default_factory=list)
    _ids: int = 1000

    def _up(self) -> None:
        if self.down:
            raise GiteaError("gitea is down")

    def whoami(self) -> str:
        return self.account

    def readable_repositories(self) -> list[str]:
        self._up()
        return [r for r in self.repos if r not in self.public]

    def visible_repositories(self) -> list[str]:
        self._up()
        return list(self.repos)

    def repository(self, repo: str) -> dict[str, Any]:
        if repo not in self.repos:
            raise GiteaNotFound(f"no repository {repo}")
        return {
            "permissions": self.permissions.get(
                repo, {"pull": True, "push": False, "admin": False}
            )
        }

    def open_pull_requests(self, repo: str) -> list[PullRequest]:
        self._up()
        if repo in self.unlisted:
            raise GiteaNotFound(f"gitea GET /repos/{repo}/pulls returned 404")
        return list(self.repos[repo])

    def pull_request(self, repo: str, index: int) -> PullRequest:
        self._up()
        for found in self.repos.get(repo, []):
            if found.number == index:
                return found
        return pr(index, state="closed", merged=(repo, index) in self.merged)

    def pull_request_commits(self, repo: str, index: int) -> list[str]:
        return ["abc123d Count words"]

    def pull_request_diff(self, repo: str, index: int) -> str:
        return self.diff

    def _id(self) -> int:
        self._ids += 1
        return self._ids

    def say(self, author: str, body: str) -> int:
        """Someone else's top-level comment on the PR."""
        n = self._id()
        self.thread.append(Comment(n, author, body, "t"))
        return n

    def note(self, author: str, body: str, where: str) -> str:
        """Someone else's comment on a line."""
        key = f"line-{self._id()}"
        self.notes.append(ReviewNote(key, author, body, where))
        return key

    def list_comments(self, repo: str, index: int, since: str | None = None):
        self._up()
        return list(self.thread)

    def review_notes(self, repo: str, index: int) -> list[ReviewNote]:
        self._up()
        mine: list[ReviewNote] = []
        for n, p in enumerate(self.posted, 1):
            if (p["repo"], p["pr"]) != (repo, index):
                continue
            if p["body"].strip():
                mine.append(ReviewNote(f"review-{n}", self.account, p["body"], ""))
            mine += [
                ReviewNote(f"line-{n}-{i}", self.account, c.body, f"{c.path}:{c.line}")
                for i, c in enumerate(p["comments"])
            ]
        return mine + list(self.notes)

    def commits_between(self, repo: str, base: str, head: str) -> list[str]:
        return list(self.since)

    def commit_diff(self, repo: str, sha: str) -> str:
        return f"diff of {sha}\n"

    def react(self, repo: str, comment_id: int, content: str) -> None:
        self.reactions.append((comment_id, content))

    def checks(self, repo: str, sha: str) -> str:
        return self.ci.get(sha, "success")

    def archive(self, repo: str, ref: str) -> bytes:
        return archive(self.files)

    def issue(self, repo: str, index: int) -> Issue:
        return Issue(index, "Counting is wrong", f"in {repo}", "don", ())

    def search_issues(self, repo: str, text: str) -> list[Issue]:
        return [Issue(3, "Counting is wrong", "", "don", ())]

    def reviews(self, repo: str, index: int) -> list[Review]:
        self._up()
        return [
            Review(n, self.account, "COMMENT", p["commit"], p["body"])
            for n, p in enumerate(self.posted, 1)
            if (p["repo"], p["pr"]) == (repo, index)
        ]

    def post_review(
        self,
        repo: str,
        index: int,
        *,
        body: str,
        commit: str,
        comments: list[LineComment] = (),  # type: ignore[assignment]
    ) -> int:
        self._up()
        if self.refuse_off_diff and any(c.path != "count.py" for c in comments):
            raise GiteaError("422: line is not part of the diff")
        self.posted.append(
            {
                "repo": repo,
                "pr": index,
                "body": body,
                "commit": commit,
                "comments": list(comments),
            }
        )
        return len(self.posted)

    def post_comment(self, repo: str, index: int, body: str) -> Comment:
        self.comments.append((repo, index, body))
        said = Comment(self._id(), self.account, body, "t")
        self.thread.append(said)
        return said

    def close(self) -> None:
        pass


def call(tool: str, **arguments: str) -> ToolCall:
    call.n += 1  # type: ignore[attr-defined]
    return ToolCall(call_id=f"c{call.n}", tool=tool, arguments=arguments)  # type: ignore[attr-defined]


call.n = 0  # type: ignore[attr-defined]


class Scripted:
    """Replies with each turn in order; remembers what it was offered."""

    def __init__(self, *turns: list[CanonicalEvent]) -> None:
        self._turns = list(turns)
        self.offered: list[set[str]] = []
        self.systems: list[str] = []
        self.seen: list[list[Any]] = []

    def complete(self, events, tools, model, system=None) -> list[CanonicalEvent]:
        self.offered.append({t["function"]["name"] for t in tools or []})
        self.systems.append(system or "")
        self.seen.append(list(events))
        if not self._turns:
            return []
        return self._turns.pop(0)