Marcos
7f37a36722
Repos can pin a free-text notice (e.g. 'this repo is in maintenance mode',
or 'reviewers: focus on the public API only for this quarter') in
.pr-review.json:static_message. The string is stripped and capped at 400
chars (mirror of the existing instructions cap), then rendered as a
Markdown blockquote (> {msg}) directly under the REVIEW_HEADER so it
surfaces on every review without scrolling.
Plumbing: parse_repo_config exposes 'static_message'; format_review_body
accepts a static_message kwarg and inserts the blockquote before the
'### Summary of Changes' section; review_pr threads
config.get('static_message') to both call sites (salvage path + happy
path). Empty / non-string values are silently dropped, mirroring the
parser's 'ignore blank' handling for every other text field.
1702 lines
63 KiB
Python
1702 lines
63 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
|
|
_balanced_json_substring,
|
|
_extract_first_json_object,
|
|
_last_balanced_json,
|
|
_render_collapsible_usage,
|
|
build_user_prompt,
|
|
compute_attribution,
|
|
findings_table,
|
|
format_review_body,
|
|
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
|
|
# New layout: always emits Summary of Changes + Key Risks. Findings table
|
|
# only shows when findings_for_table is passed (callers pass the actual
|
|
# list of finding dicts; plain-string findings arg renders as bullets).
|
|
assert "### Summary of Changes" in body
|
|
assert "### Key Risks & Concerns" in body
|
|
|
|
|
|
def test_format_review_body_empty_findings():
|
|
body = format_review_body("", "glm-5.2:cloud", "abcdef1234567890")
|
|
# No summary → "no summary provided" sentinel; Findings table absent
|
|
# because no findings were passed.
|
|
assert "_No summary provided._" in body
|
|
assert "_None identified._" in body
|
|
assert "### Findings Overview" not in body
|
|
|
|
|
|
def test_format_review_body_whitespace_findings():
|
|
body = format_review_body(" \n ", "glm-5.2:cloud", "abcdef1234567890")
|
|
assert "_No summary provided._" 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)
|
|
# Severity emoji + bracketed label.
|
|
assert "🔴 [HIGH]" in body
|
|
assert "bad" in body
|
|
# Standard ```suggestion fence (Gitea/Forgejo apply-on-click).
|
|
assert "```suggestion\ngood()\n```" in body
|
|
assert "good()" in body
|
|
|
|
|
|
def test_inline_comment_body_suggestion_not_lang_tagged():
|
|
# Per the format spec, the suggestion fence is ALWAYS ```suggestion —
|
|
# never a language-tagged fence (those are reserved for cross-file
|
|
# pattern illustrations, which we don't emit here).
|
|
f = {"severity": "high", "path": "src/Foo.java", "line": 1,
|
|
"problem": "bad", "fix": "swap", "suggestion": "good();"}
|
|
body = inline_comment_body(f)
|
|
assert "```suggestion\ngood();\n```" in body
|
|
assert "```java" not 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_inline_comment_body_severity_emoji_mapping():
|
|
cases = [
|
|
("critical", "🔴 [CRITICAL]"),
|
|
("high", "🔴 [HIGH]"),
|
|
("medium", "🟡 [MEDIUM]"),
|
|
("low", "🔵 [LOW]"),
|
|
("info", "⚪ [INFO]"),
|
|
("nit", "⚪ [INFO]"), # "nit" maps to the INFO label
|
|
("bogus", "⚪ [INFO]"), # unknown severity falls back to INFO
|
|
]
|
|
for sev, badge in cases:
|
|
f = {"severity": sev, "path": "a", "line": 1, "problem": "p", "fix": "",
|
|
"suggestion": "", "reference": ""}
|
|
assert badge in inline_comment_body(f), f"{sev} → {badge}"
|
|
|
|
|
|
def test_inline_comment_body_with_token_attribution():
|
|
# Operator wants per-comment attribution back: every inline comment shows
|
|
# the attributed output tokens + share of total. Hidden only when no
|
|
# attribution data was computed (legacy callers / ollama path without
|
|
# usage metering).
|
|
f = {"severity": "high", "path": "a", "line": 1, "problem": "p",
|
|
"fix": "f", "suggestion": "", "reference": "",
|
|
"_tok_attrib": 1234, "_tok_pct": 0.30}
|
|
body = inline_comment_body(f)
|
|
assert "🪙 ~1234 tok" in body
|
|
assert "30%" in body
|
|
assert "attributed output" 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
|
|
assert "**Fix:** f" in b
|
|
|
|
|
|
def test_summary_bullets_with_reference_link():
|
|
fs = [{"severity": "medium", "path": "x", "line": 1, "problem": "p",
|
|
"fix": "", "suggestion": "", "reference": "https://owasp.org/x"}]
|
|
b = summary_bullets(fs)
|
|
assert "🔗 **Reference:** [owasp.org/x](https://owasp.org/x)" in b
|
|
assert "https://owasp.org/x" in b # URL preserved
|
|
|
|
|
|
def test_findings_table_renders_table():
|
|
fs = [
|
|
{"severity": "high", "path": "a.py", "line": 1, "problem": "bug", "fix": "", "suggestion": "", "reference": ""},
|
|
{"severity": "low", "path": "b.go", "line": 9, "problem": "nit", "fix": "", "suggestion": "", "reference": ""},
|
|
]
|
|
t = findings_table(fs)
|
|
assert t.startswith("| Severity | Location | Finding |")
|
|
assert "|---|---|---|" in t
|
|
assert "🔴 [HIGH]" in t
|
|
assert "🔵 [LOW]" in t
|
|
assert "`a.py:1`" in t
|
|
assert "`b.go:9`" in t
|
|
|
|
|
|
def test_findings_table_escapes_pipes():
|
|
fs = [{"severity": "high", "path": "a", "line": 1,
|
|
"problem": "uses | inside", "fix": "", "suggestion": "", "reference": ""}]
|
|
t = findings_table(fs)
|
|
assert "uses \\| inside" in t
|
|
|
|
|
|
def test_findings_table_empty():
|
|
assert findings_table([]) == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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":" "}') == {}
|
|
|
|
|
|
def test_parse_repo_config_reads_static_message():
|
|
cfg = parse_repo_config(json.dumps({"static_message": " NOTE: this repo is in maintenance mode "}))
|
|
assert cfg == {"static_message": "NOTE: this repo is in maintenance mode"}
|
|
|
|
|
|
def test_parse_repo_config_static_message_caps_length():
|
|
long_text = "x" * 9999
|
|
cfg = parse_repo_config(json.dumps({"static_message": long_text}))
|
|
assert "static_message" in cfg
|
|
assert len(cfg["static_message"]) <= 400
|
|
|
|
|
|
def test_parse_repo_config_static_message_ignores_blank():
|
|
assert "static_message" not in parse_repo_config(json.dumps({"static_message": " "}))
|
|
assert "static_message" not in parse_repo_config(json.dumps({"static_message": ""}))
|
|
assert "static_message" not in parse_repo_config(json.dumps({"static_message": 42}))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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"
|
|
|
|
|
|
def test_parse_findings_fenced_json_with_nested_object():
|
|
# Real-world regression: agent emits a fence whose inner JSON has nested
|
|
# objects. The old regex `\{.*?\}` matched only the first `}`, truncating
|
|
# the JSON. Now we balance braces inside the fence.
|
|
txt = (
|
|
"```json\n"
|
|
'{"summary":"x","findings":[{"severity":"high","path":"a.py","line":1,'
|
|
'"problem":"p","fix":"f","suggestion":"","reference":""}],"meta":{"engine":"opencode"}}\n'
|
|
"```"
|
|
)
|
|
fs = parse_findings(txt)
|
|
assert len(fs) == 1
|
|
assert fs[0]["path"] == "a.py"
|
|
|
|
|
|
def test_parse_findings_unfenced_at_tail():
|
|
# No fence at all. Agent wrote the JSON inline at the very end of its
|
|
# prose. The old first-balanced regex caught the FIRST `{`, not this one.
|
|
txt = (
|
|
"I considered the diff carefully. Two findings stand out:\n"
|
|
"First one is just text.\n"
|
|
'{"findings":[{"severity":"critical","path":"x","line":1,"problem":"p","fix":"f"}]}'
|
|
)
|
|
fs = parse_findings(txt)
|
|
assert len(fs) == 1
|
|
assert fs[0]["severity"] == "critical"
|
|
|
|
|
|
def test_parse_findings_bare_array():
|
|
# Some agents skip the `{"summary":..., "findings":[...]}` wrapper and
|
|
# emit just the array.
|
|
txt = (
|
|
"Here are my findings:\n"
|
|
"```json\n"
|
|
'[{"severity":"low","path":"a","line":1,"problem":"p","fix":"f","suggestion":"","reference":""}]\n'
|
|
"```"
|
|
)
|
|
fs = parse_findings(txt)
|
|
assert len(fs) == 1
|
|
assert fs[0]["path"] == "a"
|
|
|
|
|
|
def test_parse_review_output_unfenced_at_tail():
|
|
# The exact shape canalhandia produced: long prose, JSON at the very end,
|
|
# no fence. Old parser returned ([], salvage) — now we recover findings.
|
|
txt = (
|
|
"Let me refine the fix: should call a dedicated `setPermanent`.\n"
|
|
"Let me finalize. Let me also double-check the `find` thread-safety.\n"
|
|
'{"summary":"Adds void protection; one critical race.","findings":['
|
|
'{"severity":"high","path":"VoidProtection.java","line":162,'
|
|
'"problem":"drop duplication race","fix":"use ItemMeta","suggestion":"","reference":""}]}'
|
|
)
|
|
summary, fs, *_ = parse_review_output(txt)
|
|
assert "void protection" in summary.lower()
|
|
assert len(fs) == 1
|
|
assert fs[0]["path"] == "VoidProtection.java"
|
|
|
|
|
|
def test_parse_review_output_bare_array_at_tail():
|
|
txt = (
|
|
"All wrapped up.\n"
|
|
'[{"severity":"low","path":"a","line":1,"problem":"p","fix":"","suggestion":"","reference":""}]'
|
|
)
|
|
summary, fs, *_ = parse_review_output(txt)
|
|
assert summary == ""
|
|
assert len(fs) == 1
|
|
|
|
|
|
def test_scan_balanced_handles_braces_in_strings():
|
|
# The JSON scanner must not be fooled by `{` or `}` inside string literals.
|
|
s = '{"a":"contains { and }","b":1}'
|
|
obj = _extract_first_json_object(s)
|
|
assert obj == s
|
|
d = json.loads(obj)
|
|
assert d["a"] == "contains { and }"
|
|
|
|
|
|
def test_last_balanced_json_picks_latest():
|
|
s = '{"a":1} some text {"b":2,"nested":{"c":3}} trailing'
|
|
out = _last_balanced_json(s)
|
|
assert out is not None
|
|
d = json.loads(out)
|
|
assert d == {"b": 2, "nested": {"c": 3}}
|
|
|
|
|
|
def test_last_balanced_json_no_json():
|
|
assert _last_balanced_json("nothing here") is None
|
|
assert _last_balanced_json("") is None
|
|
|
|
|
|
def test_balanced_json_substring_skips_leading_prose():
|
|
s = 'preamble {"a":1} more prose {"b":2}'
|
|
out = _balanced_json_substring(s)
|
|
assert out == '{"a":1}'
|
|
|
|
|
|
def test_balanced_json_substring_handles_array():
|
|
s = '[{"a":1},{"b":2}]'
|
|
out = _balanced_json_substring(s)
|
|
assert out == s
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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)
|
|
# Per spec: Markdown hyperlink, not raw URL.
|
|
assert "🔗 **Reference:** [cve.example/X](https://cve.example/X)" in body
|
|
|
|
|
|
def test_reference_non_url_renders_as_plain_text():
|
|
# A CVE id or doc title is not a URL. `[CVE-2024-1](CVE-2024-1)` renders as
|
|
# a broken *relative* link in Gitea, so bare text is the correct fallback.
|
|
assert ai_review._format_reference("CVE-2024-1234") == "CVE-2024-1234"
|
|
assert ai_review._format_reference("see OWASP A03") == "see OWASP A03"
|
|
assert ai_review._format_reference("") == ""
|
|
f = {"severity": "high", "path": "a", "line": 1, "problem": "p", "fix": "",
|
|
"suggestion": "", "reference": "CVE-2024-1234"}
|
|
body = inline_comment_body(f)
|
|
assert "🔗 **Reference:** CVE-2024-1234" in body
|
|
assert "](CVE-" not in body
|
|
|
|
|
|
def test_pr_has_label_reads_the_live_labels(monkeypatch):
|
|
# The AI-USAGE opt-in is read at render time, not from the trigger
|
|
# payload: labelling AI-REVIEW then AI-USAGE is two events, the review
|
|
# claims on the first, and the second is dropped by the in-flight dedupe.
|
|
seen = {}
|
|
|
|
def _get(api, repo, path, token, accept="application/json"):
|
|
seen["path"] = path
|
|
return 200, b'[{"name": "AI-REVIEW"}, {"name": "AI-USAGE"}]'
|
|
|
|
monkeypatch.setattr(ai_review, "gitea_get", _get)
|
|
assert ai_review.pr_has_label("http://api", "o/r", "9", "t", "AI-USAGE") is True
|
|
assert seen["path"] == "issues/9/labels"
|
|
assert ai_review.pr_has_label("http://api", "o/r", "9", "t", "NOPE") is False
|
|
|
|
|
|
def test_pr_has_label_survives_a_broken_api(monkeypatch):
|
|
def _boom(*a, **k):
|
|
raise RuntimeError("gitea down")
|
|
monkeypatch.setattr(ai_review, "gitea_get", _boom)
|
|
assert ai_review.pr_has_label("http://api", "o/r", "9", "t", "AI-USAGE") is False
|
|
|
|
monkeypatch.setattr(ai_review, "gitea_get", lambda *a, **k: (404, b"nope"))
|
|
assert ai_review.pr_has_label("http://api", "o/r", "9", "t", "AI-USAGE") is False
|
|
|
|
monkeypatch.setattr(ai_review, "gitea_get", lambda *a, **k: (200, b'{"not": "a list"}'))
|
|
assert ai_review.pr_has_label("http://api", "o/r", "9", "t", "AI-USAGE") is False
|
|
|
|
|
|
def test_int_env_falls_back_on_garbage(monkeypatch, capsys):
|
|
monkeypatch.setenv("PRAGENT_DIFF_CONTEXT", "two")
|
|
assert ai_review._int_env("PRAGENT_DIFF_CONTEXT", 1) == 1
|
|
assert "ignoring PRAGENT_DIFF_CONTEXT" in capsys.readouterr().err
|
|
monkeypatch.setenv("PRAGENT_DIFF_CONTEXT", " 3 ")
|
|
assert ai_review._int_env("PRAGENT_DIFF_CONTEXT", 1) == 3
|
|
monkeypatch.setenv("PRAGENT_DIFF_CONTEXT", "")
|
|
assert ai_review._int_env("PRAGENT_DIFF_CONTEXT", 1) == 1
|
|
monkeypatch.delenv("PRAGENT_DIFF_CONTEXT")
|
|
assert ai_review._int_env("PRAGENT_DIFF_CONTEXT", -1) == -1
|
|
|
|
|
|
def test_inline_comment_body_no_reference_no_ref_line():
|
|
f = {"severity": "low", "path": "a", "line": 1, "problem": "p", "fix": "",
|
|
"suggestion": "", "reference": ""}
|
|
assert "🔗" not in inline_comment_body(f)
|
|
assert "Reference:" not in inline_comment_body(f)
|
|
|
|
|
|
def test_inline_comment_body_reference_truncates_long_url():
|
|
f = {"severity": "high", "path": "a", "line": 1, "problem": "p", "fix": "",
|
|
"suggestion": "",
|
|
"reference": "https://very-long-domain.example.com/some/very/long/path/that/exceeds/the/sixty/char/limit/x"}
|
|
body = inline_comment_body(f)
|
|
# Visible label is truncated to ≤60 chars (ellipsis added).
|
|
assert "…" in body
|
|
# But the underlying URL is preserved verbatim inside the link target.
|
|
assert "very-long-domain.example.com" in body
|
|
|
|
|
|
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 + usage block + 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():
|
|
# Operator wants per-comment attribution back: every inline comment shows
|
|
# the attributed output tokens + share of total. Hidden only when no
|
|
# attribution data was computed (legacy callers / ollama path without
|
|
# usage metering).
|
|
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_render_collapsible_usage_renders_totals():
|
|
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 = _render_collapsible_usage(usage, "glm-5.2:cloud", config=None)
|
|
assert "🔋 AI Usage & Run Details" in sec
|
|
assert "`glm-5.2:cloud`" in sec
|
|
assert "7 steps" in sec
|
|
assert "142.0s" in sec
|
|
assert "18420 in / 612 out" in sec and "19032 total" in sec
|
|
assert "$0.00" in sec
|
|
assert "Whole-repo checkout" in sec
|
|
assert "attributed" in sec
|
|
|
|
|
|
def test_render_collapsible_usage_none_returns_empty():
|
|
assert _render_collapsible_usage(None, "m", config=None) == ""
|
|
|
|
|
|
def test_render_collapsible_usage_cost_nonzero_drops_free_tier_note():
|
|
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 = _render_collapsible_usage(usage, "m", config=None)
|
|
assert "$0.0123" in sec
|
|
# Was hardcoded "free tier" previously; now says "billed" since cost > 0.
|
|
assert "billed" in sec
|
|
assert "free tier" not in sec
|
|
|
|
|
|
def test_render_collapsible_usage_uses_passed_model_for_free_tier_clause():
|
|
# Regression: the cost parenthetical must reflect the actually-routed model,
|
|
# not a stale hardcoded `headroom glm-5.2:cloud` literal that predates the
|
|
# MiniMax / Anthropic switch.
|
|
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 = _render_collapsible_usage(usage, "MiniMax-M2.7", config=None)
|
|
# The parenthetical clause is "(<model> — free tier)" — a model name MUST
|
|
# sit immediately before "— free tier".
|
|
assert "(MiniMax-M2.7 — free tier)" in sec
|
|
# And the stale hardcoded model name must no longer appear anywhere.
|
|
assert "glm-5.2:cloud" not in sec
|
|
|
|
|
|
def test_render_collapsible_usage_full_provider_prefix_in_display():
|
|
# When the caller has resolved a provider-prefixed model ref (the opencode
|
|
# subprocess path), the parenthetical should mirror that verbatim.
|
|
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 = _render_collapsible_usage(usage, "headroom/MiniMax-M2.7", config=None)
|
|
assert "(headroom/MiniMax-M2.7 — free tier)" in sec
|
|
|
|
|
|
def test_render_collapsible_usage_nonzero_cost_says_billed():
|
|
usage = {"input": 10, "output": 0, "reasoning": 0, "cache_read": 0,
|
|
"cache_write": 0, "total": 10, "cost": 0.123, "steps": 1, "duration_s": 1.0}
|
|
sec = _render_collapsible_usage(usage, "MiniMax-M2.7", config=None)
|
|
assert "(MiniMax-M2.7 — billed)" in sec
|
|
assert "free tier" not in sec
|
|
|
|
|
|
def test_format_review_body_usage_section_below_findings():
|
|
# New layout: header → Summary of Changes → Key Risks → findings → usage.
|
|
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)
|
|
assert body.index("risky.") < body.index("[high]")
|
|
assert body.index("[high]") < body.index("AI usage")
|
|
assert body.index("AI usage") < 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
|
|
|
|
|
|
def test_format_review_body_renders_static_message_banner():
|
|
body = format_review_body(
|
|
"", "glm-5.2:cloud", "abcdef1234567890",
|
|
static_message="NOTE: this repo is in maintenance mode.",
|
|
)
|
|
assert "> NOTE: this repo is in maintenance mode." in body
|
|
# Banner sits under the header and above the rest of the body.
|
|
assert body.index("NOTE") > body.index("🤖")
|
|
assert body.index("NOTE") < body.index("### Summary of Changes")
|
|
|
|
|
|
def test_format_review_body_omits_static_message_when_blank():
|
|
body = format_review_body("", "glm-5.2:cloud", "abcdef1234567890")
|
|
assert "> " not in body
|
|
|
|
|
|
def test_format_review_body_with_summary_changes_and_risks():
|
|
body = format_review_body(
|
|
"", "glm-5.2:cloud", "abcdef1234567890",
|
|
summary_changes=["Adds void-death item rescue.", "Adds chunk-loader pause/rename."],
|
|
risks=["Mob spawn leak in force-loaded chunks.", "Race on /chunkloader tempo -1."],
|
|
findings_for_table=[
|
|
{"severity": "high", "path": "a.py", "line": 1, "problem": "race", "fix": "",
|
|
"suggestion": "", "reference": ""},
|
|
],
|
|
inline_count=1,
|
|
)
|
|
assert "### Summary of Changes" in body
|
|
assert "Adds void-death item rescue." in body
|
|
assert "Adds chunk-loader pause/rename." in body
|
|
assert "### Key Risks & Concerns" in body
|
|
assert "Mob spawn leak in force-loaded chunks." in body
|
|
assert "### Findings Overview" in body
|
|
assert "1 inline comment(s); 1 total." in body
|
|
assert "🔴 [HIGH]" in body
|
|
assert "`a.py:1`" in body
|
|
|
|
|
|
def test_render_collapsible_usage_contains_details():
|
|
usage = {
|
|
"model": "glm-5.2:cloud", "input": 1000, "output": 200, "reasoning": 0,
|
|
"cache_read": 0, "cache_write": 0, "total": 1200, "steps": 5, "duration_s": 12.0,
|
|
"cost": 0.0,
|
|
}
|
|
block = _render_collapsible_usage(usage, "glm-5.2:cloud", config=None)
|
|
assert "<details>" in block
|
|
assert "<summary>🔋 AI Usage & Run Details</summary>" in block
|
|
assert "</details>" in block
|
|
assert "glm-5.2:cloud" in block
|
|
assert "1000 in / 200 out" in block
|
|
|
|
|
|
def test_render_collapsible_usage_empty_when_no_usage():
|
|
assert _render_collapsible_usage(None, "glm-5.2:cloud", config=None) == ""
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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```") == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# equivalent_cost + usage-block equivalent-provider line
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_equivalent_cost_matches_cost_model():
|
|
usage = {"input": 1_000_000, "output": 0, "cache_read": 0, "cache_write": 0}
|
|
eq = ai_review.equivalent_cost(usage, "claude-sonnet-5")
|
|
# Sonnet 5 input is $2/MTok, so 1M input = $2.00 exactly.
|
|
assert abs(eq - 2.0) < 1e-9
|
|
|
|
|
|
def test_equivalent_cost_unknown_key_returns_zero():
|
|
assert ai_review.equivalent_cost({"input": 100}, "bogus") == 0.0
|
|
|
|
|
|
def test_usage_block_shows_equivalent_provider_cost():
|
|
usage = {"input": 200000, "output": 4000, "reasoning": 0,
|
|
"cache_read": 0, "cache_write": 0, "total": 204000,
|
|
"cost": 0.0, "steps": 6, "duration_s": 100.0}
|
|
sec = ai_review._render_collapsible_usage(usage, "glm-5.2:cloud", config=None)
|
|
# Two cost lines now: an equivalent (default Sonnet 5) AND the $0 actual.
|
|
assert "🔋 AI Usage & Run Details" in sec
|
|
assert "**Est. cost on Claude Sonnet 5**" in sec
|
|
assert "**Actual**: $0.00" in sec
|
|
# The "free tier" clause must mention the routed model verbatim, not the
|
|
# stale hardcoded `headroom glm-5.2:cloud` literal.
|
|
assert "free tier" in sec
|
|
assert "glm-5.2:cloud" in sec
|
|
# Equivalent should be > 0 for non-trivial token counts.
|
|
assert "$0.00" in sec # the actual line
|
|
# And a non-zero one for the equivalent.
|
|
import re
|
|
cost_lines = [ln for ln in sec.splitlines() if "cost on" in ln]
|
|
assert len(cost_lines) == 1
|
|
assert re.search(r"\$\d", cost_lines[0]) is not None
|
|
assert "$0.00" not in cost_lines[0]
|
|
|
|
|
|
def test_usage_block_honors_cost_target(monkeypatch):
|
|
monkeypatch.setenv("PRAGENT_PRICE_TARGET", "claude-haiku-4-5")
|
|
usage = {"input": 1000, "output": 100, "reasoning": 0,
|
|
"cache_read": 0, "cache_write": 0, "total": 1100,
|
|
"cost": 0.0, "steps": 1, "duration_s": 5.0}
|
|
sec = ai_review._render_collapsible_usage(usage, "glm-5.2:cloud", config=None)
|
|
assert "Claude Haiku 4.5" in sec
|
|
# 1k * $1/MTok + 100 * $5/MTok = 0.001 + 0.0005 = $0.0015
|
|
assert "$0.0015" in sec
|
|
|
|
|
|
def test_usage_block_respects_repo_config_cost_target(monkeypatch):
|
|
monkeypatch.delenv("PRAGENT_PRICE_TARGET", raising=False)
|
|
usage = {"input": 1000, "output": 100, "reasoning": 0,
|
|
"cache_read": 0, "cache_write": 0, "total": 1100,
|
|
"cost": 0.0, "steps": 1, "duration_s": 5.0}
|
|
sec = ai_review._render_collapsible_usage(
|
|
usage, "glm-5.2:cloud", config={"cost_target": "claude-opus-5"}
|
|
)
|
|
assert "Claude Opus 5" in sec
|
|
# Opus 5 = $5/MTok input + $25/MTok output → 1000*5e-6 + 100*25e-6 = 0.0075
|
|
assert "$0.0075" in sec
|
|
|
|
|
|
def test_usage_block_reports_unknown_price_target():
|
|
usage = {"input": 100, "output": 100, "reasoning": 0,
|
|
"cache_read": 0, "cache_write": 0, "total": 200,
|
|
"cost": 0.0, "steps": 1, "duration_s": 1.0}
|
|
sec = ai_review._render_collapsible_usage(
|
|
usage, "glm-5.2:cloud", config={"cost_target": "bogus-model"}
|
|
)
|
|
# Falls back to default + surfaces the error in the line.
|
|
assert "Claude Sonnet 5" in sec
|
|
assert "unknown price target" in sec
|
|
assert "bogus-model" in sec
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_repo_config — extended schema
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_repo_config_new_fields_all_valid():
|
|
raw = json.dumps({
|
|
"focus": ["security"],
|
|
"style": "strict",
|
|
"severity_threshold": "high",
|
|
"max_findings": 5,
|
|
"exclude_tests": True,
|
|
"require_tests": True,
|
|
"patterns": {"allow": ["src/**"], "deny": ["**/*.test.ts"]},
|
|
"cost_target": "claude-opus-5",
|
|
})
|
|
c = ai_review.parse_repo_config(raw)
|
|
assert c["style"] == "strict"
|
|
assert c["severity_threshold"] == "high"
|
|
assert c["max_findings"] == 5
|
|
assert c["exclude_tests"] is True
|
|
assert c["require_tests"] is True
|
|
assert c["patterns"]["allow"] == ["src/**"]
|
|
assert c["patterns"]["deny"] == ["**/*.test.ts"]
|
|
assert c["cost_target"] == "claude-opus-5"
|
|
|
|
|
|
def test_parse_repo_config_rejects_bad_style_and_threshold():
|
|
c = ai_review.parse_repo_config(json.dumps({"style": "wild", "severity_threshold": "meh"}))
|
|
assert "style" not in c
|
|
assert "severity_threshold" not in c
|
|
|
|
|
|
def test_parse_repo_config_caps_max_findings():
|
|
c1 = ai_review.parse_repo_config(json.dumps({"max_findings": 0}))
|
|
c2 = ai_review.parse_repo_config(json.dumps({"max_findings": 999}))
|
|
c3 = ai_review.parse_repo_config(json.dumps({"max_findings": "12"}))
|
|
assert "max_findings" not in c1 # 0 invalid
|
|
assert "max_findings" not in c2 # > 30 invalid
|
|
assert c3["max_findings"] == 12 # numeric string accepted
|
|
|
|
|
|
def test_parse_repo_config_caps_patterns():
|
|
raw = json.dumps({
|
|
"patterns": {"allow": [f"a{i}" for i in range(20)], "deny": [f"d{i}" for i in range(20)]}
|
|
})
|
|
c = ai_review.parse_repo_config(raw)
|
|
assert len(c["patterns"]["allow"]) == ai_review.CONFIG_MAX_PATTERNS_ITEMS
|
|
assert len(c["patterns"]["deny"]) == ai_review.CONFIG_MAX_PATTERNS_ITEMS
|
|
|
|
|
|
def test_effective_config_applies_style_defaults():
|
|
eff = ai_review.effective_config({"focus": ["security"]})
|
|
assert eff["style"] == "balanced"
|
|
assert eff["max_findings"] == 12
|
|
assert eff["severity_threshold"] == "medium"
|
|
assert eff["focus"] == ["security"]
|
|
|
|
|
|
def test_effective_config_style_overrides_fields():
|
|
eff = ai_review.effective_config({"style": "strict"})
|
|
assert eff["max_findings"] == 5
|
|
assert eff["severity_threshold"] == "high"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# apply_repo_config — filter findings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
_FINDINGS = [
|
|
{"severity": "critical", "path": "src/main.py", "line": 1, "problem": "p", "fix": "f", "suggestion": ""},
|
|
{"severity": "high", "path": "src/main.py", "line": 5, "problem": "p", "fix": "f", "suggestion": ""},
|
|
{"severity": "medium", "path": "src/main.py", "line": 9, "problem": "p", "fix": "f", "suggestion": ""},
|
|
{"severity": "low", "path": "src/main.py", "line": 13, "problem": "p", "fix": "f", "suggestion": ""},
|
|
{"severity": "high", "path": "src/FooTest.java", "line": 22, "problem": "p", "fix": "f", "suggestion": ""},
|
|
{"severity": "medium", "path": "src/app.test.ts", "line": 7, "problem": "p", "fix": "f", "suggestion": ""},
|
|
]
|
|
|
|
|
|
def test_apply_repo_config_severity_threshold():
|
|
kept, dropped = ai_review.apply_repo_config(_FINDINGS, {"severity_threshold": "high"})
|
|
assert len(kept) == 3 # critical + 2 highs (main.py + FooTest.java)
|
|
assert all(f["severity"] in ("critical", "high") for f in kept)
|
|
assert len(dropped) == 3
|
|
|
|
|
|
def test_apply_repo_config_exclude_tests_drops_test_files():
|
|
kept, dropped = ai_review.apply_repo_config(_FINDINGS, {"exclude_tests": True})
|
|
paths = {f["path"] for f in kept}
|
|
assert "src/FooTest.java" not in paths
|
|
assert "src/app.test.ts" not in paths
|
|
|
|
|
|
def test_apply_repo_config_patterns_deny_drops_matching():
|
|
cfg = {"patterns": {"deny": ["src/main.py"]}}
|
|
kept, dropped = ai_review.apply_repo_config(_FINDINGS, cfg)
|
|
paths = {f["path"] for f in kept}
|
|
assert "src/main.py" not in paths
|
|
|
|
|
|
def test_apply_repo_config_patterns_allow_keeps_only_matching():
|
|
cfg = {"patterns": {"allow": ["src/main.py"]}}
|
|
kept, dropped = ai_review.apply_repo_config(_FINDINGS, cfg)
|
|
paths = {f["path"] for f in kept}
|
|
assert paths == {"src/main.py"}
|
|
|
|
|
|
def test_apply_repo_config_max_findings_caps():
|
|
kept, dropped = ai_review.apply_repo_config(_FINDINGS, {"max_findings": 2})
|
|
assert len(kept) == 2
|
|
# Highest-severity first (critical, then high)
|
|
assert kept[0]["severity"] == "critical"
|
|
assert kept[1]["severity"] == "high"
|
|
|
|
|
|
def test_apply_repo_config_exclude_paths_glob():
|
|
cfg = {"exclude_paths": ["src/main.py"]}
|
|
kept, dropped = ai_review.apply_repo_config(_FINDINGS, cfg)
|
|
assert "src/main.py" not in {f["path"] for f in kept}
|
|
|
|
|
|
def test_apply_repo_config_require_tests_synthetic_finding():
|
|
cfg = {"require_tests": True}
|
|
changed = ["src/main.py", "src/lib.ts"]
|
|
kept, dropped = ai_review.apply_repo_config([], cfg, changed_paths=changed)
|
|
assert any(f.get("_config_synthetic") for f in kept)
|
|
|
|
|
|
def test_apply_repo_config_require_tests_no_synthetic_when_tests_present():
|
|
cfg = {"require_tests": True}
|
|
changed = ["src/main.py", "src/main_test.py"]
|
|
kept, dropped = ai_review.apply_repo_config(_FINDINGS, cfg, changed_paths=changed)
|
|
assert not any(f.get("_config_synthetic") for f in kept)
|
|
|
|
|
|
def test_is_test_path_recognises_common_patterns():
|
|
assert ai_review.is_test_path("src/FooTest.java")
|
|
assert ai_review.is_test_path("src/foo.test.ts")
|
|
assert ai_review.is_test_path("tests/foo_test.py")
|
|
assert ai_review.is_test_path("test_foo.py")
|
|
assert ai_review.is_test_path("packages/app/__tests__/foo.js")
|
|
assert not ai_review.is_test_path("src/main.py")
|
|
assert not ai_review.is_test_path("src/testing.py") # "testing" ≠ "test_"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# compact_prior_reviews
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_compact_prior_reviews_drops_prose_keeps_bullets():
|
|
bodies = [
|
|
"🤖 AI Review · m · `abc`\n\nLong prose.\n\n- **[HIGH]** `a.py:1` — bug.\n- **[LOW]** `b.go:2` — nit.\n\n_2 inline comments posted._\n<!-- pragent:sha=abc -->",
|
|
"Just chatter, no findings.",
|
|
]
|
|
out = ai_review.compact_prior_reviews(bodies)
|
|
assert len(out) == 1
|
|
assert "HIGH" in out[0] and "a.py:1" in out[0]
|
|
assert "Long prose." not in out[0]
|
|
assert "inline comments posted" not in out[0]
|
|
|
|
|
|
def test_compact_prior_reviews_empty_and_none():
|
|
assert ai_review.compact_prior_reviews([]) == []
|
|
assert ai_review.compact_prior_reviews(None) == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ADDITIONAL_CONTEXT_URL — env var + per-repo config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_repo_config_accepts_additional_context_urls():
|
|
raw = json.dumps({
|
|
"additional_context_urls": [
|
|
"https://nexus.example.com/raw/context.md",
|
|
" https://other.example/x.md ",
|
|
123, # ignored (non-string)
|
|
"", # ignored (empty after strip)
|
|
]
|
|
})
|
|
cfg = parse_repo_config(raw)
|
|
assert "additional_context_urls" in cfg
|
|
# Non-strings and empty are stripped; whitespace trimmed.
|
|
assert cfg["additional_context_urls"] == [
|
|
"https://nexus.example.com/raw/context.md",
|
|
"https://other.example/x.md",
|
|
]
|
|
|
|
|
|
def test_parse_repo_config_additional_context_urls_capped_at_8():
|
|
urls = [f"https://x.example/{i}.md" for i in range(20)]
|
|
cfg = parse_repo_config(json.dumps({"additional_context_urls": urls}))
|
|
assert len(cfg["additional_context_urls"]) == 8
|
|
|
|
|
|
def test_parse_repo_config_additional_context_urls_absent_when_missing():
|
|
assert "additional_context_urls" not in parse_repo_config("{}")
|
|
|
|
|
|
def test_resolve_additional_context_urls_env_wins_and_dedupes(monkeypatch):
|
|
monkeypatch.setenv(
|
|
"PRAGENT_ADDITIONAL_CONTEXT_URL",
|
|
"https://env.example/a.md, https://env.example/b.md",
|
|
)
|
|
cfg = {"additional_context_urls": [
|
|
"https://env.example/a.md", # dup with env -> dropped from cfg list
|
|
"https://cfg.example/d.md",
|
|
]}
|
|
urls = ai_review._resolve_additional_context_urls(cfg)
|
|
# Env comes first, in declared order; cfg entries that duplicate env are skipped.
|
|
assert urls == [
|
|
"https://env.example/a.md",
|
|
"https://env.example/b.md",
|
|
"https://cfg.example/d.md",
|
|
]
|
|
|
|
|
|
def test_resolve_additional_context_urls_no_env_no_config():
|
|
import os as _os
|
|
_os.environ.pop("PRAGENT_ADDITIONAL_CONTEXT_URL", None)
|
|
assert ai_review._resolve_additional_context_urls(None) == []
|
|
assert ai_review._resolve_additional_context_urls({}) == []
|
|
|
|
|
|
def test_resolve_additional_context_urls_total_cap_is_8(monkeypatch):
|
|
monkeypatch.setenv(
|
|
"PRAGENT_ADDITIONAL_CONTEXT_URL",
|
|
",".join(f"https://e.example/{i}" for i in range(20)),
|
|
)
|
|
urls = ai_review._resolve_additional_context_urls({
|
|
"additional_context_urls": [f"https://c.example/{i}" for i in range(20)]
|
|
})
|
|
assert len(urls) == 8
|
|
|
|
|
|
class _FakeResp:
|
|
"""Minimal stand-in for urllib's HTTP response: context manager + .read(N)."""
|
|
|
|
def __init__(self, body: bytes):
|
|
import io as _io
|
|
self._buf = _io.BytesIO(body)
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
def read(self, n=-1):
|
|
return self._buf.read(n)
|
|
|
|
|
|
def _patch_urlopen(body_for_url):
|
|
"""Replace ai_review.urllib.request.urlopen with a fake that returns the
|
|
configured body for each URL. `body_for_url: dict[str, bytes]`. Records
|
|
every URL it sees in `calls` on the closure."""
|
|
calls: list[str] = []
|
|
|
|
def fake(req, *args, **kwargs):
|
|
url = req.full_url if hasattr(req, "full_url") else str(req)
|
|
calls.append(url)
|
|
return _FakeResp(body_for_url.get(url, b""))
|
|
|
|
import ai_review as _ar
|
|
orig = _ar.urllib.request.urlopen
|
|
_ar.urllib.request.urlopen = fake
|
|
|
|
def restore():
|
|
_ar.urllib.request.urlopen = orig
|
|
|
|
return calls, restore
|
|
|
|
|
|
def test_fetch_additional_context_joins_blocks():
|
|
ai_review._ADDITIONAL_CONTEXT_CACHE.clear()
|
|
calls, restore = _patch_urlopen({
|
|
"https://a/x.md": b"alpha body",
|
|
"https://b/y.md": b"beta body",
|
|
})
|
|
try:
|
|
out = ai_review.fetch_additional_context(["https://a/x.md", "https://b/y.md"])
|
|
finally:
|
|
restore()
|
|
|
|
assert "alpha body" in out and "beta body" in out
|
|
assert calls == ["https://a/x.md", "https://b/y.md"]
|
|
|
|
|
|
def test_fetch_additional_context_caches_by_url():
|
|
ai_review._ADDITIONAL_CONTEXT_CACHE.clear()
|
|
calls, restore = _patch_urlopen({"https://a/x.md": b"cached body"})
|
|
try:
|
|
ai_review.fetch_additional_context(["https://a/x.md"])
|
|
ai_review.fetch_additional_context(["https://a/x.md", "https://a/x.md"])
|
|
finally:
|
|
restore()
|
|
|
|
# Second call hits cache; only one network call despite 3 references.
|
|
assert calls == ["https://a/x.md"]
|
|
|
|
|
|
def test_fetch_additional_context_rejects_non_http_schemes():
|
|
ai_review._ADDITIONAL_CONTEXT_CACHE.clear()
|
|
out = ai_review.fetch_additional_context([
|
|
"file:///etc/passwd",
|
|
"javascript:alert(1)",
|
|
"ftp://x/y",
|
|
])
|
|
# All rejected at scheme check, no network calls.
|
|
assert out == ""
|
|
|
|
|
|
def test_fetch_additional_context_truncates_per_url():
|
|
ai_review._ADDITIONAL_CONTEXT_CACHE.clear()
|
|
cap = ai_review._ADDITIONAL_CONTEXT_MAX_PER_URL_CHARS
|
|
_, restore = _patch_urlopen({"https://a/big.md": b"X" * (cap + 500)})
|
|
try:
|
|
out = ai_review.fetch_additional_context(["https://a/big.md"])
|
|
finally:
|
|
restore()
|
|
|
|
assert "…[truncated]" in out
|
|
# The fetched body is bounded to `cap` chars (the marker + the URL
|
|
# header are appended on top by the joiner, so we count just X's).
|
|
assert out.count("X") == cap
|
|
|
|
|
|
def test_fetch_additional_context_caps_total_chars():
|
|
ai_review._ADDITIONAL_CONTEXT_CACHE.clear()
|
|
cap_total = ai_review._ADDITIONAL_CONTEXT_MAX_TOTAL_CHARS
|
|
# Each block: "### https://a/N.md\n\n" + 3990 X + "\nEND" ≈ 4017 chars.
|
|
big = (b"X" * 3990) + b"\nEND"
|
|
urls = [f"https://a/{i}.md" for i in range(8)]
|
|
_, restore = _patch_urlopen({u: big for u in urls})
|
|
try:
|
|
out = ai_review.fetch_additional_context(urls)
|
|
finally:
|
|
restore()
|
|
|
|
# Total is bounded by the cap plus the truncation marker (if the last
|
|
# block was cut mid-flight).
|
|
assert len(out) <= cap_total + 20, len(out)
|
|
|
|
|
|
def test_fetch_additional_context_empty_returns_empty():
|
|
assert ai_review.fetch_additional_context([]) == ""
|
|
|
|
|
|
def test_build_user_prompt_injects_additional_context():
|
|
prompt = build_user_prompt(
|
|
"T", "B", "diff", config=None, prior_reviews=None,
|
|
additional_context="### https://a/x.md\n\nalpha body",
|
|
)
|
|
assert "## Repo-provided context" in prompt
|
|
assert "alpha body" in prompt
|
|
# URL header preserved so the agent knows which block is which.
|
|
assert "https://a/x.md" in prompt
|
|
|
|
|
|
def test_build_user_prompt_skips_additional_context_when_empty():
|
|
prompt = build_user_prompt("T", "B", "diff")
|
|
assert "## Repo-provided context" not in prompt
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_repo_config: reviewers[] + triage (multi-lens orchestration)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_repo_config_reviewers_array_basic():
|
|
raw = json.dumps({
|
|
"reviewers": [
|
|
{"id": "security", "severity_floor": "high", "max_findings": 10},
|
|
{"id": "docs", "agent_file": ".opencode/agents/docs.md"},
|
|
{"id": "perf", "model": "headroom/glm-5.2:cloud",
|
|
"skip_if_all_changed_paths": "docs/**"},
|
|
]
|
|
})
|
|
cfg = parse_repo_config(raw)
|
|
assert cfg["reviewers"] == [
|
|
{"id": "security", "severity_floor": "high", "max_findings": 10},
|
|
{"id": "docs", "agent_file": ".opencode/agents/docs.md"},
|
|
{"id": "perf", "model": "headroom/glm-5.2:cloud",
|
|
"skip_if_all_changed_paths": "docs/**"},
|
|
]
|
|
|
|
|
|
def test_parse_repo_config_reviewers_rejects_bad_id():
|
|
# Punctuation, leading dash, underscore, empty — all silently dropped.
|
|
cfg = parse_repo_config(json.dumps({
|
|
"reviewers": [
|
|
{"id": "BAD!!!"},
|
|
{"id": "-bad-start"},
|
|
{"id": "ok_under"},
|
|
{"id": ""},
|
|
{"id": "good-one"},
|
|
]
|
|
}))
|
|
assert cfg["reviewers"] == [{"id": "good-one"}]
|
|
|
|
|
|
def test_parse_repo_config_reviewers_caps_at_8():
|
|
cfg = parse_repo_config(json.dumps({
|
|
"reviewers": [{"id": f"l{i}"} for i in range(12)]
|
|
}))
|
|
assert len(cfg["reviewers"]) == 8
|
|
|
|
|
|
def test_parse_repo_config_reviewers_drop_non_dict_entries():
|
|
cfg = parse_repo_config(json.dumps({
|
|
"reviewers": ["not-a-dict", 42, None, {"id": "ok"}]
|
|
}))
|
|
assert cfg["reviewers"] == [{"id": "ok"}]
|
|
|
|
|
|
def test_parse_repo_config_reviewers_absent_yields_no_key():
|
|
cfg = parse_repo_config("{}")
|
|
assert "reviewers" not in cfg
|
|
|
|
|
|
def test_parse_repo_config_reviewers_activation_validated():
|
|
cfg = parse_repo_config(json.dumps({
|
|
"reviewers": [
|
|
{"id": "a", "activation": "auto"},
|
|
{"id": "b", "activation": "always"},
|
|
{"id": "c", "activation": "off"},
|
|
{"id": "d", "activation": "BOGUS"}, # dropped (unsupported)
|
|
]
|
|
}))
|
|
# Only the entries with valid activation carry the key — the BOGUS one
|
|
# just keeps id (the unknown field is silently dropped, not rejected).
|
|
assert [r.get("activation") for r in cfg["reviewers"]] == [
|
|
"auto", "always", "off", None
|
|
]
|
|
|
|
|
|
def test_parse_repo_config_triage_object_full():
|
|
cfg = parse_repo_config(json.dumps({
|
|
"triage": {"enabled": True, "model": "headroom/haiku", "max_lenses": 3}
|
|
}))
|
|
assert cfg["triage"] == {"enabled": True, "model": "headroom/haiku", "max_lenses": 3}
|
|
|
|
|
|
def test_parse_repo_config_triage_disabled():
|
|
cfg = parse_repo_config(json.dumps({"triage": {"enabled": False}}))
|
|
assert cfg["triage"] == {"enabled": False}
|
|
|
|
|
|
def test_parse_repo_config_triage_malformed_yields_disabled():
|
|
# Non-object triage value (string, list, number) should disable, not crash.
|
|
for raw in (
|
|
'{"triage": "off"}',
|
|
'{"triage": []}',
|
|
'{"triage": 42}',
|
|
):
|
|
cfg = parse_repo_config(raw)
|
|
assert cfg.get("triage") == {"enabled": False}, f"failed for {raw}"
|
|
|
|
|
|
def test_parse_repo_config_triage_absent_yields_no_key():
|
|
cfg = parse_repo_config("{}")
|
|
assert "triage" not in cfg
|
|
|
|
|
|
def test_parse_repo_config_triage_max_lenses_capped_at_8():
|
|
cfg = parse_repo_config(json.dumps({"triage": {"max_lenses": 100}}))
|
|
# 100 is out of range; the key is dropped, not clamped. Caller defaults.
|
|
assert "max_lenses" not in cfg.get("triage", {})
|
|
|
|
|
|
def test_render_collapsible_usage_shows_lenses_when_multi():
|
|
usage = {
|
|
"input": 100, "output": 50, "reasoning": 0,
|
|
"cache_read": 0, "cache_write": 0, "total": 150,
|
|
"steps": 12, "duration_s": 8.4,
|
|
"lenses": ["security", "docs", "tests"],
|
|
"lens_steps": 12,
|
|
}
|
|
out = _render_collapsible_usage(usage, "headroom/glm-5.2:cloud", None)
|
|
assert "Lenses" in out
|
|
# All three lens ids are shown in backticks.
|
|
assert "`security`" in out
|
|
assert "`docs`" in out
|
|
assert "`tests`" in out
|
|
# Step count is surfaced.
|
|
assert "12" in out
|
|
|
|
|
|
def test_render_collapsible_usage_omits_lenses_when_single_primary():
|
|
usage = {
|
|
"input": 100, "output": 50, "reasoning": 0,
|
|
"cache_read": 0, "cache_write": 0, "total": 150,
|
|
"steps": 4, "duration_s": 2.0,
|
|
}
|
|
out = _render_collapsible_usage(usage, "headroom/glm-5.2:cloud", None)
|
|
assert "Lenses" not in out
|
|
|