claude-code
d778c113e9
werkator build succeeded after 01:14
werkrevisor 0.13.0: mechanical file changes skip the model. - A file is mechanical only if every changed line is a pure rename, a whitespace change (not in whitespace-significant files) or an import line of a known language; it goes neither into the diff nor into the full files for the model. - A PR of only mechanical files is GREEN-eligible without a model call; hard rules still apply. - Additional changes: the example host configs match mih09 (`mode = "combined"`, `token_count_factor = 1.55`); an unknown change kind in a changed `.werkrevisor.toml` gives YELLOW, not RED. Details: the PR document under `docs/prs/`. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Reviewed-on: #11 Co-authored-by: claude-code <claude-code@javagil.de>
272 lines
11 KiB
Python
272 lines
11 KiB
Python
"""Runs the real PR-Agent 0.47.0 review code with a fake Gitea provider and a fake model, fully offline."""
|
||
|
||
import unittest
|
||
from types import SimpleNamespace
|
||
|
||
from tests.fakes import HEAD, config, pr_ref
|
||
|
||
try:
|
||
import pr_agent.git_providers as providers
|
||
from pr_agent.algo.ai_handlers.base_ai_handler import BaseAiHandler
|
||
from pr_agent.algo.file_filter import filter_ignored
|
||
from pr_agent.algo.run_details import record_ai_call
|
||
from pr_agent.algo.types import EDIT_TYPE, FilePatchInfo
|
||
from pr_agent.git_providers.git_provider import GitProvider, IncrementalPR
|
||
from pr_agent.git_providers.gitea_provider import GiteaProvider, _GiteaCommitAdapter
|
||
HAVE_PR_AGENT = True
|
||
except ImportError:
|
||
HAVE_PR_AGENT = False
|
||
|
||
ANSWER = """```yaml
|
||
review:
|
||
estimated_effort_to_review_[1-5]: |
|
||
1
|
||
risk_level: |
|
||
low
|
||
merge_recommendation: |
|
||
changes_required
|
||
relevant_tests: |
|
||
No
|
||
key_issues_to_review:
|
||
- relevant_file: |
|
||
src/a.py
|
||
issue_header: |
|
||
Possible Bug
|
||
issue_content: |
|
||
Division by zero when n is 0.
|
||
start_line: 2
|
||
end_line: 2
|
||
security_concerns: |
|
||
No
|
||
```"""
|
||
|
||
PATCH = "@@ -1,1 +1,2 @@\n def f(n):\n+ return 1 / n\n"
|
||
|
||
|
||
def _patch_info(name, patch):
|
||
return FilePatchInfo(base_file="def f(n):\n", head_file="def f(n):\n return 1 / n\n", patch=patch,
|
||
filename=name, edit_type=EDIT_TYPE.MODIFIED, num_plus_lines=1, num_minus_lines=0)
|
||
|
||
|
||
def fake_provider_class(files):
|
||
class FakeGiteaProvider(GiteaProvider):
|
||
def __init__(self, url=None):
|
||
GitProvider.__init__(self)
|
||
self.pr_url = url
|
||
self.base_url = "https://git.example.org"
|
||
self.owner, self.repo, self.pr_number = "team", "app", 7
|
||
self.pr = SimpleNamespace(title="Add f", body="Adds f.", labels=[], html_url=url,
|
||
head=SimpleNamespace(sha=HEAD, ref="feature"),
|
||
base=SimpleNamespace(sha="b" * 40, ref="main"))
|
||
self.sha = HEAD
|
||
self.git_files = [{"filename": f.filename, "status": "modified"} for f in files]
|
||
self.diff_files = list(files)
|
||
self.incremental = IncrementalPR(False)
|
||
self.last_commit = self.last_commit_id = _GiteaCommitAdapter({"sha": HEAD})
|
||
self.enabled_pr, self.enabled_issue = True, False
|
||
self.max_comment_chars = 65000
|
||
|
||
def get_languages(self):
|
||
return {"Python": 100}
|
||
|
||
def get_commit_messages(self):
|
||
return "Add f"
|
||
|
||
def get_diff_files(self):
|
||
# Like GiteaProvider.get_diff_files, which applies the [ignore] settings to the Gitea file list.
|
||
kept = {f["filename"] for f in filter_ignored(self.git_files, platform="gitea")}
|
||
return [f for f in self.diff_files if f.filename in kept]
|
||
|
||
def publish_comment(self, *args, **kwargs):
|
||
raise AssertionError("publish_output is false, nothing may be published")
|
||
|
||
return FakeGiteaProvider
|
||
|
||
|
||
def fake_handler_class(answer, prompts):
|
||
class FakeHandler(BaseAiHandler):
|
||
def __init__(self):
|
||
pass
|
||
|
||
@property
|
||
def deployment_id(self):
|
||
return None
|
||
|
||
async def chat_completion(self, model, system, user, temperature=0.2, img_path=None):
|
||
prompts.append((model, system, user))
|
||
record_ai_call({"prompt_tokens": 1234, "completion_tokens": 210, "total_tokens": 1444}, model=model)
|
||
return answer, "stop"
|
||
|
||
return FakeHandler
|
||
|
||
|
||
@unittest.skipUnless(HAVE_PR_AGENT, "runs inside the PR-Agent image")
|
||
class PrAgentRunTest(unittest.TestCase):
|
||
def setUp(self):
|
||
from werkrevisor.redact import Redactor
|
||
from werkrevisor.review import configure_logging
|
||
configure_logging(Redactor(), "CRITICAL")
|
||
|
||
def run_review(self, files, answer=ANSWER, section="", admit=None, prompts=None, excluded_paths=(),
|
||
**config_overrides):
|
||
from werkrevisor.review import run_model_review
|
||
prompts = [] if prompts is None else prompts
|
||
original = providers._GIT_PROVIDERS["gitea"]
|
||
providers._GIT_PROVIDERS["gitea"] = fake_provider_class(files)
|
||
try:
|
||
run = run_model_review(config(**config_overrides), pr_ref(), "t" * 20, "k" * 20, "Check X.", section,
|
||
admit=admit, ai_handler=fake_handler_class(answer, prompts),
|
||
excluded_paths=excluded_paths)
|
||
finally:
|
||
providers._GIT_PROVIDERS["gitea"] = original
|
||
return run, prompts
|
||
|
||
def test_review_result_usage_and_prompt(self):
|
||
run, prompts = self.run_review([_patch_info("src/a.py", PATCH)])
|
||
self.assertEqual(len(prompts), 1)
|
||
model, system, user = prompts[0]
|
||
self.assertEqual(model, config().model.id)
|
||
self.assertIn("Check X.", system)
|
||
self.assertIn("merge_recommendation", system)
|
||
self.assertIn("## File: 'src/a.py'", user)
|
||
self.assertEqual(run.prompt_files, {"src/a.py"})
|
||
self.assertEqual(run.reviewed_head_sha, HEAD)
|
||
self.assertEqual(run.data["review"]["merge_recommendation"].strip(), "changes_required")
|
||
self.assertEqual(run.details["num_ai_calls"], 1)
|
||
self.assertEqual((run.details["prompt_tokens"], run.details["completion_tokens"]), (1234, 210))
|
||
self.assertEqual(run.details["cost_status"], "unavailable")
|
||
|
||
def test_full_files_follow_the_diff_in_the_user_prompt(self):
|
||
from werkrevisor import fullfiles
|
||
section = fullfiles.render_section([fullfiles.render_file("src/a.py", "def f(n):\n return 1 / n\n")], HEAD)
|
||
run, prompts = self.run_review([_patch_info("src/a.py", PATCH)], section=section)
|
||
_, system, user = prompts[0]
|
||
self.assertNotIn("Full content of the changed files", system)
|
||
diff_at = user.index("## File: 'src/a.py'")
|
||
files_at = user.index("## Full content of the changed files")
|
||
self.assertLess(diff_at, files_at)
|
||
self.assertLess(files_at, user.index(fullfiles.SECTION_END))
|
||
self.assertEqual(run.prompt_files, {"src/a.py"})
|
||
|
||
def test_excluded_files_never_reach_the_prompt(self):
|
||
names = ("src/a.py", "src/b.py", "src/b.pyx", "x/src/b.py")
|
||
run, prompts = self.run_review([_patch_info(name, PATCH) for name in names], excluded_paths=("src/b.py",))
|
||
_, _, user = prompts[0]
|
||
self.assertNotIn("## File: 'src/b.py'", user)
|
||
self.assertEqual(run.prompt_files, {"src/a.py", "src/b.pyx", "x/src/b.py"})
|
||
|
||
def test_admit_sees_the_exact_prompt_before_the_model_call(self):
|
||
events = []
|
||
prompts = RecordingList(events)
|
||
|
||
def admit(system, user):
|
||
events.append(("admit", system, user))
|
||
|
||
self.run_review([_patch_info("src/a.py", PATCH)], admit=admit, prompts=prompts)
|
||
self.assertEqual([e[0] for e in events], ["admit", "model"])
|
||
self.assertEqual(events[0][1:], events[1][1:])
|
||
self.assertIn("## File: 'src/a.py'", events[0][2])
|
||
|
||
def test_too_long_prompt_never_reaches_the_model(self):
|
||
from werkrevisor import budget
|
||
cfg = config(model={"max_context_tokens": 3000})
|
||
|
||
def admit(system, user):
|
||
count = budget.count_prompt(system, user, cfg.model, lambda: self.fail("the tokenizer must load"))
|
||
reason = budget.precheck(count, cfg.model, cfg.budget)
|
||
if reason:
|
||
raise budget.NoModelCall(reason, "skipped (too large)")
|
||
|
||
section = "\n" + "word " * 3000
|
||
prompts = []
|
||
with self.assertRaises(budget.NoModelCall) as caught:
|
||
self.run_review([_patch_info("src/a.py", PATCH)], section=section, admit=admit, prompts=prompts,
|
||
model={"max_context_tokens": 3000})
|
||
self.assertEqual(prompts, [])
|
||
self.assertRegex(caught.exception.reason.text,
|
||
r"^no AI review: the PR needs \d+ input tokens \(\d+ counted with o200k_base × "
|
||
r"\[model\] token_count_factor = 1.55\), more than \[model\] max_context_tokens = 3000$")
|
||
|
||
def test_pruned_prompt_never_reaches_the_model(self):
|
||
from werkrevisor.budget import NoModelCall
|
||
big = "@@ -1,1 +1,8001 @@\n def f(n):\n" + "".join(f"+ x{i} = {i} * n\n" for i in range(8000))
|
||
files = [_patch_info("src/a.py", PATCH), _patch_info("src/big.py", big)]
|
||
admitted, prompts = [], []
|
||
with self.assertRaises(NoModelCall) as caught:
|
||
self.run_review(files, admit=lambda s, u: admitted.append(u), prompts=prompts,
|
||
model={"max_context_tokens": 3000})
|
||
self.assertEqual((admitted, prompts), ([], []))
|
||
self.assertEqual(caught.exception.reason.code, "too_large_tokens")
|
||
self.assertIn("PR-Agent's window of 10000 tokens", caught.exception.reason.text)
|
||
|
||
def test_prompt_refused_by_pr_agent_never_reaches_the_model(self):
|
||
from werkrevisor.budget import NoModelCall
|
||
admitted, prompts = [], []
|
||
with self.assertRaises(NoModelCall) as caught:
|
||
self.run_review([_patch_info("src/a.py", PATCH)], section="\n" + "word " * 12000,
|
||
admit=lambda s, u: admitted.append(u), prompts=prompts, model={"max_context_tokens": 3000})
|
||
self.assertEqual((admitted, prompts), ([], []))
|
||
self.assertIn("PR-Agent's window of 10000 tokens", caught.exception.reason.text)
|
||
|
||
def test_stop_carries_the_run_details_of_pr_agent(self):
|
||
from werkrevisor.budget import NoModelCall
|
||
from werkrevisor.verdict import Reason
|
||
|
||
def admit(system, user):
|
||
raise NoModelCall(Reason("budget", "budget: test"), "skipped (budget)")
|
||
|
||
prompts = []
|
||
with self.assertRaises(NoModelCall) as caught:
|
||
self.run_review([_patch_info("src/a.py", PATCH)], admit=admit, prompts=prompts)
|
||
self.assertEqual(prompts, [])
|
||
self.assertEqual(caught.exception.details["num_ai_calls"], 0)
|
||
|
||
|
||
class RefusalTrackingTest(unittest.TestCase):
|
||
"""A refusal is told apart from a model error per attempt, so a retry or fallback attempt is judged on its own."""
|
||
|
||
class Eligible(Exception):
|
||
pass
|
||
|
||
def attempts(self, outcomes):
|
||
import asyncio
|
||
from werkrevisor.review import _refusal_as_shortened
|
||
called = []
|
||
|
||
async def prepare(model):
|
||
calls, error = outcomes.pop(0)
|
||
called.extend([True] * calls)
|
||
if error:
|
||
raise self.Eligible(error)
|
||
return "ok"
|
||
|
||
tracked = _refusal_as_shortened(prepare, called, self.Eligible)
|
||
results = []
|
||
for _ in range(2):
|
||
try:
|
||
results.append(asyncio.run(tracked("m")))
|
||
except Exception as e:
|
||
results.append(type(e).__name__)
|
||
return results
|
||
|
||
def test_refusal_after_an_earlier_call_is_still_a_refusal(self):
|
||
self.assertEqual(self.attempts([(1, "invalid answer"), (0, "does not fit")]), ["Eligible", "PromptShortened"])
|
||
|
||
def test_eligible_error_after_a_call_stays_a_model_error(self):
|
||
self.assertEqual(self.attempts([(0, None), (1, "invalid answer")]), ["ok", "Eligible"])
|
||
|
||
|
||
class RecordingList(list):
|
||
def __init__(self, events):
|
||
super().__init__()
|
||
self.events = events
|
||
|
||
def append(self, item):
|
||
_, system, user = item
|
||
self.events.append(("model", system, user))
|
||
super().append(item)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|