Marcos
5d28f97080
Found by running the reviewer against this branch. The second review of PR #7 ran a full agent loop — 330s of a 540s budget, no timeout — and ended without a parseable ```json block. The code discarded the entire run and posted "AI review produced no parseable output.", losing minutes of work and millions of tokens for a message that tells the maintainer nothing and gives me nothing to debug. Three changes on that path: - salvage_summary() keeps the agent's prose (fenced blocks stripped, tail kept because the conclusion is written last) and posts it under an explicit banner saying it is unstructured and its line numbers were never validated against the diff. A partial review honestly labelled beats no review. - The raw output's length and last 600 chars go to stderr, so the next occurrence is diagnosable from pod logs instead of invisible. - The AI-USAGE section is still rendered. The label asked for it and the tokens were spent either way; dropping the measurement on the failure path is how the cost model stops getting calibration data exactly when it is most interesting. Not fixed here: why the agent went off-format. The likely cause is the 40-step cap in the agent definition being reached on a larger diff (the successful run used 28), which wants either a higher cap or a step-budget warning in the prompt. Needs the next occurrence's stderr to confirm rather than guess. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01B11e8TZZxJyzHW7jj7KWUN
782 lines
27 KiB
Python
782 lines
27 KiB
Python
"""Unit tests for pragent pilot pure helpers. No network."""
|
|
import base64
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
# Allow running without install: add repo root to path.
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT = os.path.abspath(os.path.join(HERE, "..", ".."))
|
|
sys.path.insert(0, os.path.join(ROOT, "pilot"))
|
|
|
|
import ai_review # noqa: E402
|
|
from ai_review import ( # noqa: E402
|
|
build_user_prompt,
|
|
compute_attribution,
|
|
format_review_body,
|
|
format_usage_section,
|
|
inline_comment_body,
|
|
parse_diff_anchors,
|
|
parse_findings,
|
|
parse_repo_config,
|
|
parse_review_output,
|
|
parse_text_blocks,
|
|
prior_review_bodies,
|
|
reviewed_shas,
|
|
split_findings,
|
|
summary_bullets,
|
|
truncate_diff,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# truncate_diff
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_truncate_diff_short():
|
|
text, truncated, n = truncate_diff("abc", 100)
|
|
assert text == "abc"
|
|
assert truncated is False
|
|
assert n == 3
|
|
|
|
|
|
def test_truncate_diff_exact_boundary():
|
|
text, truncated, n = truncate_diff("x" * 100, 100)
|
|
assert truncated is False
|
|
assert n == 100
|
|
assert text == "x" * 100
|
|
|
|
|
|
def test_truncate_diff_over_cap():
|
|
text, truncated, n = truncate_diff("x" * 250, 100)
|
|
assert truncated is True
|
|
assert n == 250
|
|
assert text.startswith("x" * 100)
|
|
assert "[diff truncated at 100 characters]" in text
|
|
|
|
|
|
def test_truncate_diff_none():
|
|
text, truncated, n = truncate_diff(None, 100) # type: ignore[arg-type]
|
|
assert text == ""
|
|
assert truncated is False
|
|
assert n == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_text_blocks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_text_blocks_text_only():
|
|
content = [{"type": "text", "text": "hello"}, {"type": "text", "text": "world"}]
|
|
assert parse_text_blocks(content) == "hello\nworld"
|
|
|
|
|
|
def test_parse_text_blocks_drops_thinking():
|
|
content = [
|
|
{"type": "thinking", "thinking": "reasoning here"},
|
|
{"type": "text", "text": "- [high] a.go:3 — bug. fix."},
|
|
]
|
|
assert parse_text_blocks(content) == "- [high] a.go:3 — bug. fix."
|
|
|
|
|
|
def test_parse_text_blocks_empty_and_malformed():
|
|
assert parse_text_blocks([]) == ""
|
|
assert parse_text_blocks(None) == "" # type: ignore[arg-type]
|
|
assert parse_text_blocks([{"type": "text"}, "garbage", 5]) == ""
|
|
|
|
|
|
def test_parse_text_blocks_real_glm_shape():
|
|
# Captured from glm-5.2:cloud via headroom 8789.
|
|
content = [
|
|
{"type": "thinking", "thinking": "Analyze the request..."},
|
|
{"type": "text", "text": "- [critical] auth.py:12 — token compared with `==`. Use hmac.compare_digest."},
|
|
]
|
|
assert "compare_digest" in parse_text_blocks(content)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# format_review_body
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_format_review_body_findings():
|
|
body = format_review_body("- [high] x:1 — bug. fix.", "glm-5.2:cloud", "abcdef1234567890")
|
|
assert "pragent pilot" in body
|
|
assert "glm-5.2:cloud" in body
|
|
assert "`abcdef12`" in body # 8-char sha
|
|
assert "- [high] x:1" in body
|
|
|
|
|
|
def test_format_review_body_empty_findings():
|
|
body = format_review_body("", "glm-5.2:cloud", "abcdef1234567890")
|
|
assert "No issues found." in body
|
|
|
|
|
|
def test_format_review_body_whitespace_findings():
|
|
body = format_review_body(" \n ", "glm-5.2:cloud", "abcdef1234567890")
|
|
assert "No issues found." in body
|
|
|
|
|
|
def test_format_review_body_no_sha():
|
|
body = format_review_body("- [low] y:2 — nit", "glm-5.2:cloud", "")
|
|
assert "`unknown`" in body
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# build_user_prompt
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_user_prompt_includes_title_and_diff():
|
|
p = build_user_prompt("Fix login", "Closes #1", "diff --git a/x b/x")
|
|
assert "Fix login" in p
|
|
assert "Closes #1" in p
|
|
assert "diff --git a/x b/x" in p
|
|
|
|
|
|
def test_build_user_prompt_truncates_long_body():
|
|
long_body = "B" * 6000
|
|
p = build_user_prompt("t", long_body, "d")
|
|
assert "[PR body truncated]" in p
|
|
assert p.count("B") < 6000
|
|
|
|
|
|
def test_build_user_prompt_no_body():
|
|
p = build_user_prompt("t", "", "d")
|
|
assert "Description:" not in p
|
|
|
|
|
|
def test_build_user_prompt_with_config_and_prior():
|
|
cfg = {"focus": ["security"], "instructions": "Use Result<T,E>."}
|
|
prior = ["🤖 **AI Review** …\n- [high] x:1 — old."]
|
|
p = build_user_prompt("t", "b", "diff --git a/x b/x", config=cfg, prior_reviews=prior)
|
|
assert "## Repo review config" in p
|
|
assert "security" in p
|
|
assert "Result<T,E>" in p
|
|
assert "## PREVIOUS REVIEWS" in p
|
|
assert "old." in p
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_diff_anchors
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
_DIFF = """\
|
|
diff --git a/src/a.py b/src/a.py
|
|
index 1..2 100644
|
|
--- a/src/a.py
|
|
+++ b/src/a.py
|
|
@@ -1,4 +1,5 @@
|
|
context
|
|
-removed
|
|
+added
|
|
context2
|
|
@@ -10,3 +10,4 @@
|
|
keep
|
|
+new
|
|
last
|
|
diff --git a/binary.bin b/binary.bin
|
|
new file mode 100644
|
|
index 0..1
|
|
Binary files differ
|
|
"""
|
|
|
|
|
|
def test_parse_diff_anchors_context_and_added():
|
|
a = parse_diff_anchors(_DIFF)
|
|
# context(1), +added(2), context2(3) | keep(10), +new(11), last(12)
|
|
assert a["src/a.py"] == {1, 2, 3, 10, 11, 12}
|
|
# removed line (-removed, old line 2) has no new-line anchor
|
|
assert 2 in a["src/a.py"] # 2 here is the +added line, not the removed one
|
|
|
|
|
|
def test_parse_diff_anchors_binary_file_present_no_lines():
|
|
a = parse_diff_anchors(_DIFF)
|
|
assert "binary.bin" in a
|
|
assert a["binary.bin"] == set()
|
|
|
|
|
|
def test_parse_diff_anchors_empty():
|
|
assert parse_diff_anchors("") == {}
|
|
assert parse_diff_anchors(None) == {} # type: ignore[arg-type]
|
|
|
|
|
|
def test_parse_diff_anchors_new_file():
|
|
diff = "diff --git a/new.ts b/new.ts\nnew file mode 100644\n--- /dev/null\n+++ b/new.ts\n@@ -0,0 +1,3 @@\n+a\n+b\n+c\n"
|
|
a = parse_diff_anchors(diff)
|
|
assert a["new.ts"] == {1, 2, 3}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_findings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_findings_clean_json():
|
|
txt = '{"findings":[{"severity":"high","path":"a.py","line":3,"problem":"x","fix":"y","suggestion":"z"}]}'
|
|
fs = parse_findings(txt)
|
|
assert len(fs) == 1
|
|
assert fs[0]["severity"] == "high"
|
|
assert fs[0]["path"] == "a.py"
|
|
assert fs[0]["line"] == 3
|
|
|
|
|
|
def test_parse_findings_fenced_json():
|
|
txt = '```json\n{"findings":[{"severity":"low","path":"b.go","line":1,"problem":"p","fix":"","suggestion":""}]}\n```'
|
|
fs = parse_findings(txt)
|
|
assert len(fs) == 1
|
|
assert fs[0]["path"] == "b.go"
|
|
|
|
|
|
def test_parse_findings_json_in_prose():
|
|
txt = 'Here is my review: {"findings":[{"severity":"critical","path":"c","line":9,"problem":"q"}]} thanks!'
|
|
fs = parse_findings(txt)
|
|
assert len(fs) == 1
|
|
assert fs[0]["severity"] == "critical"
|
|
|
|
|
|
def test_parse_findings_empty():
|
|
assert parse_findings('{"findings":[]}') == []
|
|
assert parse_findings("") == []
|
|
assert parse_findings("not json at all") == []
|
|
|
|
|
|
def test_parse_findings_drops_bad_entries():
|
|
# missing path, bad line, unknown severity (normalised)
|
|
txt = '{"findings":[{"line":1},{"path":"x","line":-1},{"path":"x","line":2,"severity":"bogus","problem":"p"}]}'
|
|
fs = parse_findings(txt)
|
|
assert len(fs) == 1
|
|
assert fs[0]["severity"] == "medium"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# split_findings + inline_comment_body + summary_bullets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_split_findings_by_anchor():
|
|
anchors = {"a.py": {1, 3, 4}}
|
|
fs = [
|
|
{"severity": "high", "path": "a.py", "line": 3, "problem": "p", "fix": "f", "suggestion": ""},
|
|
{"severity": "low", "path": "a.py", "line": 99, "problem": "off", "fix": "", "suggestion": ""},
|
|
{"severity": "medium", "path": "other.go", "line": 1, "problem": "x", "fix": "", "suggestion": ""},
|
|
]
|
|
anchored, unanchored = split_findings(fs, anchors)
|
|
assert [f["line"] for f in anchored] == [3]
|
|
assert len(unanchored) == 2
|
|
|
|
|
|
def test_inline_comment_body_with_suggestion():
|
|
f = {"severity": "high", "path": "a", "line": 1, "problem": "bad", "fix": "swap", "suggestion": "good()"}
|
|
body = inline_comment_body(f)
|
|
assert "**[HIGH]**" in body
|
|
assert "bad" in body
|
|
# no extension → bare fence (Gitea 1.26.x has no apply-suggestion; we tag
|
|
# with the file language for highlighting instead of ```suggestion)
|
|
assert "```\ngood()\n```" in body
|
|
assert "good()" in body
|
|
|
|
|
|
def test_inline_comment_body_suggestion_lang_tagged():
|
|
f = {"severity": "high", "path": "src/Foo.java", "line": 1,
|
|
"problem": "bad", "fix": "swap", "suggestion": "good();"}
|
|
body = inline_comment_body(f)
|
|
assert "```java\ngood();\n```" in body
|
|
|
|
|
|
def test_inline_comment_body_no_suggestion():
|
|
f = {"severity": "low", "path": "a", "line": 1, "problem": "p", "fix": "f", "suggestion": ""}
|
|
body = inline_comment_body(f)
|
|
assert "```" not in body
|
|
assert "Fix: f" in body
|
|
|
|
|
|
def test_summary_bullets_format():
|
|
fs = [{"severity": "high", "path": "a.py", "line": 7, "problem": "p", "fix": "f", "suggestion": ""}]
|
|
b = summary_bullets(fs)
|
|
assert "- **[HIGH]**" in b
|
|
assert "`a.py:7`" in b
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# repo config parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_repo_config_full():
|
|
raw = '{"focus":["security","perf"],"exclude_paths":["vendor/**"],"languages":["go"],"instructions":"be strict"}'
|
|
c = parse_repo_config(raw)
|
|
assert c["focus"] == ["security", "perf"]
|
|
assert c["exclude_paths"] == ["vendor/**"]
|
|
assert c["instructions"] == "be strict"
|
|
|
|
|
|
def test_parse_repo_config_partial_and_bad():
|
|
assert parse_repo_config('{"focus":"not-a-list"}') == {}
|
|
assert parse_repo_config('{"focus":["ok"]}') == {"focus": ["ok"]}
|
|
assert parse_repo_config("") == {}
|
|
assert parse_repo_config("not json") == {}
|
|
assert parse_repo_config('{"instructions":" "}') == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# dedupe / prior-context parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_reviewed_shas_extracts_marker():
|
|
reviews = [
|
|
{"body": "🤖 AI Review · glm · `abcdef12`\n\n<!-- pragent:sha=abcdef1234567890 -->"},
|
|
{"body": "human comment, no marker"},
|
|
{"body": "<!-- pragent:sha=0987654 -->"},
|
|
]
|
|
shas = reviewed_shas(reviews)
|
|
assert "abcdef1234567890" in shas
|
|
assert "0987654" in shas
|
|
|
|
|
|
def test_reviewed_shas_empty():
|
|
assert reviewed_shas([]) == set()
|
|
assert reviewed_shas([{"body": "no marker"}]) == set()
|
|
|
|
|
|
def test_prior_review_bodies_skips_current_sha():
|
|
reviews = [
|
|
{"body": "r1\n<!-- pragent:sha=1111111 -->"},
|
|
{"body": "r2\n<!-- pragent:sha=2222222 -->"},
|
|
{"body": "no marker here"},
|
|
]
|
|
prior = prior_review_bodies(reviews, current_sha="2222222")
|
|
assert len(prior) == 1
|
|
assert "r1" in prior[0]
|
|
|
|
|
|
def test_format_review_body_has_sha_marker():
|
|
body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890")
|
|
assert "<!-- pragent:sha=abcdef1234567890 -->" in body
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_review_output (opencode engine: {summary, findings} + reference)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_review_output_summary_and_findings():
|
|
txt = (
|
|
"This PR adds an eval helper — risky. See findings.\n\n"
|
|
"```json\n"
|
|
'{"summary":"Adds eval() — security risk.","findings":['
|
|
'{"severity":"critical","path":"src/x.ts","line":4,"problem":"eval on user input",'
|
|
'"fix":"parse explicitly","suggestion":"const n = Number(s)","reference":"https://owasp.org/x"}'
|
|
"]}",
|
|
"\n```",
|
|
)
|
|
summary, fs = parse_review_output("".join(txt))
|
|
assert "eval()" in summary
|
|
assert len(fs) == 1
|
|
assert fs[0]["severity"] == "critical"
|
|
assert fs[0]["reference"] == "https://owasp.org/x"
|
|
assert fs[0]["suggestion"] == "const n = Number(s)"
|
|
|
|
|
|
def test_parse_review_output_bare_findings_no_summary():
|
|
txt = '```json\n{"findings":[{"severity":"low","path":"a","line":1,"problem":"p"}]}\n```'
|
|
summary, fs = parse_review_output(txt)
|
|
assert summary == ""
|
|
assert len(fs) == 1
|
|
assert fs[0]["reference"] == "" # default
|
|
|
|
|
|
def test_parse_review_output_empty_and_bogus():
|
|
assert parse_review_output("") == ("", [])
|
|
assert parse_review_output("no json here") == ("", [])
|
|
assert parse_review_output('{"findings":[]}') == ("", [])
|
|
|
|
|
|
def test_parse_review_output_uses_last_json_block():
|
|
# Agent emits a stray json-ish block first, then the real one last.
|
|
txt = (
|
|
"```json\n{\"findings\":[{\"path\":\"x\",\"line\":1,\"severity\":\"low\"}]}\n```\n"
|
|
"more prose\n"
|
|
"```json\n{\"summary\":\"real\",\"findings\":[{\"path\":\"y\",\"line\":2,\"severity\":\"high\"}]}\n```"
|
|
)
|
|
summary, fs = parse_review_output(txt)
|
|
assert summary == "real"
|
|
assert len(fs) == 1
|
|
assert fs[0]["path"] == "y"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# reference rendering in inline_comment_body + summary_bullets + summary section
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_inline_comment_body_renders_reference():
|
|
f = {"severity": "high", "path": "a", "line": 1, "problem": "p", "fix": "f",
|
|
"suggestion": "", "reference": "https://cve.example/X"}
|
|
body = inline_comment_body(f)
|
|
assert "📎 ref: https://cve.example/X" in body
|
|
|
|
|
|
def test_inline_comment_body_no_reference_no_ref_line():
|
|
f = {"severity": "low", "path": "a", "line": 1, "problem": "p", "fix": "",
|
|
"suggestion": "", "reference": ""}
|
|
assert "📎 ref" not in inline_comment_body(f)
|
|
|
|
|
|
def test_summary_bullets_renders_reference():
|
|
fs = [{"severity": "high", "path": "a.py", "line": 7, "problem": "p", "fix": "f",
|
|
"suggestion": "", "reference": "https://r.example"}]
|
|
b = summary_bullets(fs)
|
|
assert "https://r.example" in b
|
|
assert "`a.py:7`" in b
|
|
|
|
|
|
def test_format_review_body_with_summary_section():
|
|
body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890",
|
|
summary="This PR adds a risky helper.")
|
|
assert "This PR adds a risky helper." in body
|
|
assert "- [high] x:1" in body
|
|
assert "<!-- pragent:sha=abcdef1234567890 -->" in body
|
|
# summary appears before the findings bullets
|
|
assert body.index("risky helper") < body.index("[high]")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AI-USAGE: compute_attribution + format_usage_section + inline 🪙 line
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_compute_attribution_weighted_split():
|
|
# weights 1 (problem="a") and 3 (problem="aaa"), output 100 → 25 / 75
|
|
fs = [
|
|
{"severity": "high", "path": "x", "line": 1, "problem": "a", "fix": "", "suggestion": ""},
|
|
{"severity": "low", "path": "x", "line": 2, "problem": "aaa", "fix": "", "suggestion": ""},
|
|
]
|
|
compute_attribution(fs, 100)
|
|
assert fs[0]["_tok_attrib"] == 25
|
|
assert fs[1]["_tok_attrib"] == 75
|
|
assert abs(fs[0]["_tok_pct"] - 0.25) < 1e-9
|
|
assert abs(fs[1]["_tok_pct"] - 0.75) < 1e-9
|
|
|
|
|
|
def test_compute_attribution_zero_weights_splits_evenly():
|
|
fs = [
|
|
{"severity": "high", "path": "x", "line": 1, "problem": "", "fix": "", "suggestion": ""},
|
|
{"severity": "low", "path": "x", "line": 2, "problem": "", "fix": "", "suggestion": ""},
|
|
]
|
|
compute_attribution(fs, 80)
|
|
assert fs[0]["_tok_attrib"] == 40
|
|
assert fs[1]["_tok_attrib"] == 40
|
|
assert abs(fs[0]["_tok_pct"] - 0.5) < 1e-9
|
|
|
|
|
|
def test_compute_attribution_noop_on_empty_or_zero_budget():
|
|
fs = [{"severity": "high", "path": "x", "line": 1, "problem": "a", "fix": "", "suggestion": ""}]
|
|
compute_attribution([], 100)
|
|
compute_attribution(fs, 0)
|
|
assert "_tok_attrib" not in fs[0]
|
|
|
|
|
|
def test_inline_comment_body_with_attribution_line():
|
|
f = {"severity": "high", "path": "a", "line": 1, "problem": "bad", "fix": "swap",
|
|
"suggestion": "", "_tok_attrib": 180, "_tok_pct": 0.29}
|
|
body = inline_comment_body(f)
|
|
assert "🪙 ~180 tok" in body
|
|
assert "29%" in body
|
|
assert "attributed output" in body
|
|
|
|
|
|
def test_inline_comment_body_no_attribution_no_coin_line():
|
|
f = {"severity": "high", "path": "a", "line": 1, "problem": "bad", "fix": "swap",
|
|
"suggestion": ""}
|
|
assert "🪙" not in inline_comment_body(f)
|
|
|
|
|
|
def test_format_usage_section_renders_totals_and_table():
|
|
fs = [
|
|
{"severity": "critical", "path": "src/Foo.java", "line": 98,
|
|
"problem": "p"*10, "fix": "f", "suggestion": "", "_tok_attrib": 180, "_tok_pct": 0.29},
|
|
]
|
|
usage = {"input": 18420, "output": 612, "reasoning": 0, "cache_read": 15210,
|
|
"cache_write": 0, "total": 19032, "cost": 0.0, "steps": 7, "duration_s": 142.0}
|
|
sec = format_usage_section(usage, fs, "glm-5.2:cloud")
|
|
assert "## 🔋 AI usage" in sec
|
|
assert "`glm-5.2:cloud`" in sec
|
|
assert "agent steps: 7" in sec
|
|
assert "duration: 142.0s" in sec
|
|
assert "18420 in" in sec and "612 out" in sec and "19032 total" in sec
|
|
assert "$0.00" in sec
|
|
assert "whole-repo checkout" in sec
|
|
assert "attributed" in sec
|
|
# table
|
|
assert "| severity | location | ≈out tok | % |" in sec
|
|
assert "CRITICAL" in sec
|
|
assert "`src/Foo.java:98`" in sec
|
|
assert "180" in sec and "29%" in sec
|
|
|
|
|
|
def test_format_usage_section_omits_table_when_no_attributed_rows():
|
|
usage = {"input": 10, "output": 0, "reasoning": 0, "cache_read": 0,
|
|
"cache_write": 0, "total": 10, "cost": 0.0, "steps": 1, "duration_s": 1.0}
|
|
sec = format_usage_section(usage, [], "glm-5.2:cloud")
|
|
assert "## 🔋 AI usage" in sec
|
|
assert "severity | location" not in sec # no rows → no table
|
|
|
|
|
|
def test_format_usage_section_none_returns_empty():
|
|
assert format_usage_section(None, [], "m") == ""
|
|
|
|
|
|
def test_format_usage_section_cost_nonzero():
|
|
usage = {"input": 10, "output": 0, "reasoning": 0, "cache_read": 0,
|
|
"cache_write": 0, "total": 10, "cost": 0.0123, "steps": 1, "duration_s": 1.0}
|
|
sec = format_usage_section(usage, [], "m")
|
|
assert "$0.0123" in sec
|
|
assert "billed by provider" in sec
|
|
|
|
|
|
def test_format_review_body_usage_section_between_summary_and_findings():
|
|
usage_sec = "## 🔋 AI usage\n\n- model: `m`"
|
|
body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890",
|
|
summary="This PR is risky.", usage_section=usage_sec)
|
|
# order: header < summary < usage < findings < marker
|
|
assert body.index("risky.") < body.index("AI usage")
|
|
assert body.index("AI usage") < body.index("[high]")
|
|
assert body.index("[high]") < body.index("<!-- pragent:sha=")
|
|
assert "## 🔋 AI usage" in body
|
|
|
|
|
|
def test_format_review_body_no_usage_section_omitted():
|
|
body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890")
|
|
assert "AI usage" not in body
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_diff_anchors — empty context lines
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_diff_anchors_counts_empty_context_line():
|
|
# A context line that is *blank* arrives as "" when trailing whitespace was
|
|
# stripped somewhere upstream. If it isn't counted, every later line in the
|
|
# hunk is off by one.
|
|
diff = (
|
|
"diff --git a/x.py b/x.py\n"
|
|
"--- a/x.py\n"
|
|
"+++ b/x.py\n"
|
|
"@@ -1,4 +1,5 @@\n"
|
|
" import os\n"
|
|
"\n" # blank context line, whitespace stripped
|
|
" def f():\n"
|
|
"+ return 1\n"
|
|
" # tail\n"
|
|
)
|
|
anchors = parse_diff_anchors(diff)
|
|
assert anchors["x.py"] == {1, 2, 3, 4, 5}
|
|
|
|
|
|
def test_parse_diff_anchors_space_prefixed_blank_line_still_counts():
|
|
diff = (
|
|
"+++ b/y.py\n"
|
|
"@@ -1,3 +1,4 @@\n"
|
|
" a\n"
|
|
" \n" # properly space-prefixed blank context line
|
|
"+b\n"
|
|
" c\n"
|
|
)
|
|
assert parse_diff_anchors(diff)["y.py"] == {1, 2, 3, 4}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_repo_config — caps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_repo_config_caps_instructions_length():
|
|
cfg = parse_repo_config(json.dumps({"instructions": "x" * 99999}))
|
|
assert len(cfg["instructions"]) == ai_review.CONFIG_MAX_INSTRUCTIONS_CHARS
|
|
|
|
|
|
def test_parse_repo_config_caps_list_length_and_items():
|
|
cfg = parse_repo_config(json.dumps({
|
|
"focus": ["a" * 9999] * 500,
|
|
"exclude_paths": ["vendor/**"],
|
|
}))
|
|
assert len(cfg["focus"]) == ai_review.CONFIG_MAX_LIST_ITEMS
|
|
assert all(len(x) == ai_review.CONFIG_MAX_ITEM_CHARS for x in cfg["focus"])
|
|
assert cfg["exclude_paths"] == ["vendor/**"]
|
|
|
|
|
|
def test_parse_repo_config_still_accepts_normal_config():
|
|
cfg = parse_repo_config(json.dumps({
|
|
"focus": ["security"], "languages": ["go"], "instructions": "No bare throw.",
|
|
}))
|
|
assert cfg == {"focus": ["security"], "languages": ["go"], "instructions": "No bare throw."}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# fetch_repo_config — reads the BASE ref, never the PR head
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _stub_config_response(payload: dict):
|
|
blob = base64.b64encode(json.dumps(payload).encode()).decode()
|
|
return 200, json.dumps({"content": blob}).encode()
|
|
|
|
|
|
def test_fetch_repo_config_uses_given_base_ref(monkeypatch):
|
|
seen = {}
|
|
|
|
def fake_get(api, repo, path, token, accept="application/json"):
|
|
seen["path"] = path
|
|
return _stub_config_response({"focus": ["security"]})
|
|
|
|
monkeypatch.setattr(ai_review, "gitea_get", fake_get)
|
|
cfg = ai_review.fetch_repo_config("http://g", "o/r", "tok", ref="main")
|
|
assert cfg == {"focus": ["security"]}
|
|
assert seen["path"] == "contents/.pr-review.json?ref=main"
|
|
|
|
|
|
def test_fetch_repo_config_without_ref_omits_ref_param(monkeypatch):
|
|
seen = {}
|
|
|
|
def fake_get(api, repo, path, token, accept="application/json"):
|
|
seen["path"] = path
|
|
return _stub_config_response({})
|
|
|
|
monkeypatch.setattr(ai_review, "gitea_get", fake_get)
|
|
ai_review.fetch_repo_config("http://g", "o/r", "tok")
|
|
assert "?ref=" not in seen["path"]
|
|
|
|
|
|
def test_fetch_repo_config_quotes_ref_with_slashes(monkeypatch):
|
|
seen = {}
|
|
|
|
def fake_get(api, repo, path, token, accept="application/json"):
|
|
seen["path"] = path
|
|
return _stub_config_response({})
|
|
|
|
monkeypatch.setattr(ai_review, "gitea_get", fake_get)
|
|
ai_review.fetch_repo_config("http://g", "o/r", "tok", ref="release/v1 x")
|
|
assert "release%2Fv1%20x" in seen["path"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# post_inline_review — degraded fallback must not lose findings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_post_inline_review_fallback_keeps_anchored_findings(monkeypatch):
|
|
posted = []
|
|
|
|
def fake_post(api, repo, path, token, body):
|
|
posted.append((path, body))
|
|
# Reject the inline review, accept the plain one.
|
|
if "reviews" in path and body.get("comments"):
|
|
return 422, b"bad line"
|
|
return 201, b"{}"
|
|
|
|
monkeypatch.setattr(ai_review, "gitea_post", fake_post)
|
|
anchored = [{
|
|
"severity": "high", "path": "a.py", "line": 7,
|
|
"problem": "off-by-one", "fix": "use <=", "suggestion": "", "reference": "",
|
|
}]
|
|
ai_review.post_inline_review("http://g", "o/r", "1", "tok", "SUMMARY", anchored)
|
|
|
|
final_body = posted[-1][1]["body"]
|
|
assert "off-by-one" in final_body
|
|
assert "a.py:7" in final_body
|
|
assert "SUMMARY" in final_body
|
|
|
|
|
|
def test_post_inline_review_success_posts_no_fallback(monkeypatch):
|
|
posted = []
|
|
|
|
def fake_post(api, repo, path, token, body):
|
|
posted.append(path)
|
|
return 201, b"{}"
|
|
|
|
monkeypatch.setattr(ai_review, "gitea_post", fake_post)
|
|
ai_review.post_inline_review("http://g", "o/r", "1", "tok", "S", [])
|
|
assert len(posted) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# fetch_pr_diff — files-endpoint fallback
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_fetch_pr_diff_fallback_emits_git_style_prefixes(monkeypatch):
|
|
def fake_get(api, repo, path, token, accept="application/json"):
|
|
if path.endswith(".diff"):
|
|
return 404, b"nope"
|
|
return 200, json.dumps([
|
|
{"filename": "src/a.py", "patch": "@@ -1 +1,2 @@\n a\n+b"},
|
|
]).encode()
|
|
|
|
monkeypatch.setattr(ai_review, "gitea_get", fake_get)
|
|
diff, truncated, _ = ai_review.fetch_pr_diff("http://g", "o/r", "1", "tok", 10000)
|
|
assert "--- a/src/a.py" in diff
|
|
assert "+++ b/src/a.py" in diff
|
|
assert truncated is False
|
|
# And the synthesized diff must actually anchor.
|
|
assert parse_diff_anchors(diff)["src/a.py"] == {1, 2}
|
|
|
|
|
|
def test_fetch_pr_diff_error_reports_both_statuses(monkeypatch):
|
|
def fake_get(api, repo, path, token, accept="application/json"):
|
|
return (404, b"") if path.endswith(".diff") else (500, b"")
|
|
|
|
monkeypatch.setattr(ai_review, "gitea_get", fake_get)
|
|
try:
|
|
ai_review.fetch_pr_diff("http://g", "o/r", "1", "tok", 10000)
|
|
except RuntimeError as e:
|
|
assert ".diff=404" in str(e)
|
|
assert "files=500" in str(e)
|
|
else:
|
|
raise AssertionError("expected RuntimeError")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# salvage_summary — don't discard an expensive run over a missing JSON block
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_salvage_summary_keeps_the_prose():
|
|
text = "I reviewed the diff. The retry loop in worker.py never terminates."
|
|
out = ai_review.salvage_summary(text)
|
|
assert "never terminates" in out
|
|
assert "unverified" in out
|
|
|
|
|
|
def test_salvage_summary_drops_fenced_blocks():
|
|
text = 'Analysis here.\n\n```json\n{"findings": [ truncated...\n'
|
|
out = ai_review.salvage_summary(text)
|
|
assert "Analysis here." in out
|
|
# The half-written JSON blob is gone (the banner legitimately says
|
|
# "findings", so assert on the blob's own content instead).
|
|
assert "truncated..." not in out
|
|
assert "[" not in out.split("_\n\n", 1)[1]
|
|
|
|
|
|
def test_salvage_summary_drops_complete_fences_too():
|
|
text = "Before.\n```python\nprint(1)\n```\nAfter."
|
|
out = ai_review.salvage_summary(text)
|
|
assert "Before." in out and "After." in out
|
|
assert "print(1)" not in out
|
|
|
|
|
|
def test_salvage_summary_keeps_the_tail_when_long():
|
|
text = "x" * 9000 + " FINAL CONCLUSION"
|
|
out = ai_review.salvage_summary(text, max_chars=1000)
|
|
assert "FINAL CONCLUSION" in out # the conclusion is written last
|
|
assert len(out) < 1600
|
|
|
|
|
|
def test_salvage_summary_empty_when_nothing_to_salvage():
|
|
assert ai_review.salvage_summary("") == ""
|
|
assert ai_review.salvage_summary(" \n ") == ""
|
|
assert ai_review.salvage_summary("```json\n{}\n```") == ""
|