"""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)