"""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_unquoted_no_is_no_security_concern(self): """Scenario#71.02""" from werkrevisor.verdict import parse_model_review answer = ANSWER.replace("security_concerns: |\n No", "security_concerns: No") self.assertNotEqual(answer, ANSWER) run, _ = self.run_review([_patch_info("src/a.py", PATCH)], answer=answer) self.assertIs(run.data["review"]["security_concerns"], False) self.assertEqual(parse_model_review(run.data).security_concerns, "") 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_gradle_files_reach_the_prompt(self): gradle = "@@ -1,1 +1,2 @@\n rootProject.name = 'app'\n+include 'lib'\n" run, prompts = self.run_review([_patch_info("src/a.py", PATCH), _patch_info("settings.gradle", gradle)]) _, _, user = prompts[0] self.assertIn("## File: 'settings.gradle'", user) self.assertEqual(run.prompt_files, {"src/a.py", "settings.gradle"}) 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()