dongarandClaude Opus 5.5
3813ad55d6
Archives were downloaded and unpacked afresh for every run, which doesn't scale to very large repositories. Each repository is now cloned once, with its whole history, under scratch/repos/<owner>/<name>/ (head/ the clone, base/ a worktree when read). A run fetches every branch and the PR's refs/pull/N/head, checks out the head exactly, and fails if that head is gone. Only the latest repository is kept for now. Git runs walled off: none of the host's git configuration, links checked out as plain files, no hooks, submodules or LFS, only the Gitea URL's protocol, and the token as an http.extraHeader for that URL in git's environment only. A stop ends a git command; a cut-short first clone is removed. The reader never shows .git. GiteaClient.archive is gone. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk
269 lines
8.9 KiB
Python
269 lines
8.9 KiB
Python
"""A Gitea and a model to review against, in memory."""
|
|
|
|
from dataclasses import dataclass, field, replace
|
|
from pathlib import Path
|
|
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())
|
|
"""
|
|
|
|
|
|
class FakeSources:
|
|
"""The repository's code as plain folders: ``gitea.files`` at head,
|
|
and at base unless ``base`` is given."""
|
|
|
|
def __init__(self, gitea: "FakeGitea", root: Path) -> None:
|
|
self._gitea = gitea
|
|
self._root = root
|
|
self.base: dict[str, str] | None = None
|
|
self.opened: list[tuple[str, int, str, str]] = []
|
|
|
|
def open(self, repo: str, number: int, head: str, base: str) -> "FakeTrees":
|
|
self.opened.append((repo, number, head, base))
|
|
return FakeTrees(self, len(self.opened))
|
|
|
|
|
|
class FakeTrees:
|
|
def __init__(self, sources: FakeSources, n: int) -> None:
|
|
self._sources = sources
|
|
self._n = n
|
|
self.made: list[str] = []
|
|
|
|
def tree(self, ref: str) -> Path:
|
|
root = self._sources._root / f"run{self._n}" / ref
|
|
if not root.exists():
|
|
self.made.append(ref)
|
|
files = self._sources._gitea.files
|
|
if ref == "base" and self._sources.base is not None:
|
|
files = self._sources.base
|
|
for path, text in files.items():
|
|
(root / path).parent.mkdir(parents=True, exist_ok=True)
|
|
(root / path).write_text(text)
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
return root
|
|
|
|
|
|
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 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)
|