"""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 _CONFIDENCE_BADGE, _SEVERITY_EMOJI, _balanced_json_substring, _extract_first_json_object, _last_balanced_json, _normalize_finding, _render_collapsible_usage, _severity_badge, build_user_prompt, compute_attribution, findings_table, fmt_tokens, format_review_body, inline_comment_body, merge_confidence, parse_diff_anchors, parse_findings, parse_repo_config, parse_review_output, parse_text_blocks, prior_review_bodies, reviewed_shas, REVIEW_HEADER, SEVERITIES, SEVERITY_RANK, 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."} 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" 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", "⚪ [NIT]"), # legacy alias — renders with its own name ("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 "🪙 ~1,234 (1.2K) 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"}') == {"enabled": False} assert parse_repo_config('{"focus":["ok"]}') == {"focus": ["ok"], "enabled": False} assert parse_repo_config("") == {} assert parse_repo_config("not json") == {} assert parse_repo_config('{"instructions":" "}') == {"enabled": False} 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.get("static_message") == "NOTE: this repo is in maintenance mode" assert cfg.get("enabled") is False 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})) def test_parse_repo_config_reads_model_override(): # Per-repo override is validated against cost_model.PRICES. Only keys # the cost model knows about can override the review engine. cfg = parse_repo_config(json.dumps({"model": "claude-sonnet-5"})) assert cfg.get("model") == "claude-sonnet-5" def test_parse_repo_config_rejects_unknown_model(capsys): cfg = parse_repo_config(json.dumps({"model": "not-in-prices"})) assert "model" not in cfg # Repos that pin a typo should get a stderr hint pointing at the valid set. err = capsys.readouterr().err assert "model" in err.lower() or "prices" in err.lower() or "unknown" in err.lower() def test_parse_repo_config_model_must_be_string(): assert "model" not in parse_repo_config(json.dumps({"model": 42})) assert "model" not in parse_repo_config(json.dumps({"model": []})) assert "model" not in parse_repo_config(json.dumps({"model": None})) def test_resolve_display_model_precedence(monkeypatch): # Order is OPENCODE_MODEL env > config['model'] > headroom/{base}. monkeypatch.delenv("OPENCODE_MODEL", raising=False) # 1. No env, no config → headroom/ assert ai_review._resolve_display_model("MiniMax-M2.7", None) == "headroom/MiniMax-M2.7" assert ai_review._resolve_display_model("MiniMax-M2.7", {}) == "headroom/MiniMax-M2.7" # 2. No env, config has model → use config model as-is (already a known key) assert ( ai_review._resolve_display_model("MiniMax-M2.7", {"model": "claude-sonnet-5"}) == "claude-sonnet-5" ) # 3. Env wins over config monkeypatch.setenv("OPENCODE_MODEL", "headroom/MiniMax-M2.7") assert ( ai_review._resolve_display_model("MiniMax-M2.7", {"model": "claude-sonnet-5"}) == "headroom/MiniMax-M2.7" ) # 4. Env alone, no config monkeypatch.delenv("OPENCODE_MODEL") assert ai_review._resolve_display_model("x", {}) == "headroom/x" def test_format_review_body_uses_override_for_cost_paren(): # End-to-end sanity: when the caller passes the resolved override as the # `model` arg to format_review_body, both the header AND the cost line # show the override — i.e. callers DO substitute the resolved display # name into both the opencode subprocess ref and the review body. body = format_review_body( "- [high] x:1 — bug. fix.", "claude-sonnet-5", "abcdef1234567890", ) assert "claude-sonnet-5" in body assert "MiniMax-M2.7" not in body # the base didn't leak through assert "🤖" in body # header rendered # --------------------------------------------------------------------------- # dedupe / prior-context parsing # --------------------------------------------------------------------------- def test_reviewed_shas_extracts_marker(): reviews = [ {"body": "🤖 AI Review · glm · `abcdef12`\n\n"}, {"body": "human comment, no marker"}, {"body": ""}, ] 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"}, {"body": "r2\n"}, {"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 "" 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_parse_review_output_extracts_walkthrough_risk_tests(): # The 7-tuple shape carries three new top-level fields: # walkthrough (list[str]), risk_verdict (str), test_coverage (str). txt = ( "```json\n" "{\n" ' "summary": "x",\n' ' "summary_changes": [],\n' ' "risks": [],\n' ' "walkthrough": ["a.py: adds X", "b.py: refactors Y"],\n' ' "risk_verdict": "Low risk.",\n' ' "test_coverage": "No tests for behavioral change in a.py.",\n' ' "findings": []\n' "}\n" "```" ) summary, findings, _changes, _risks, walkthrough, risk_verdict, test_coverage = ( parse_review_output(txt) ) assert summary == "x" assert findings == [] assert walkthrough == ["a.py: adds X", "b.py: refactors Y"] assert risk_verdict == "Low risk." assert test_coverage == "No tests for behavioral change in a.py." def test_parse_review_output_missing_fields_default_empty(): # Backward-compatible: the 4-tuple shape still parses fine; the new # fields default to empty list / empty string. out = parse_review_output('{"summary":"x","findings":[]}') summary, findings, _changes, _risks, walkthrough, risk_verdict, test_coverage = out assert summary == "x" assert findings == [] assert walkthrough == [] assert risk_verdict == "" assert test_coverage == "" 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_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 "" 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 "18,420 (18.4K) in / 612 out" in sec and "19,032 (19.0K) 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 "( — 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("", "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 # --------------------------------------------------------------------------- # fmt_tokens # --------------------------------------------------------------------------- def test_fmt_tokens_zero(): assert fmt_tokens(0) == "0" def test_fmt_tokens_small_no_short(): assert fmt_tokens(42) == "42" assert fmt_tokens(999) == "999" def test_fmt_tokens_thousands(): assert fmt_tokens(1000) == "1,000 (1.0K)" assert fmt_tokens(1234) == "1,234 (1.2K)" assert fmt_tokens(9999) == "9,999 (10.0K)" def test_fmt_tokens_millions(): assert fmt_tokens(1_000_000) == "1,000,000 (1.0M)" assert fmt_tokens(2_071_025) == "2,071,025 (2.1M)" assert fmt_tokens(1_234_567) == "1,234,567 (1.2M)" def test_fmt_tokens_billions(): assert fmt_tokens(1_234_567_890) == "1,234,567,890 (1.2B)" def test_fmt_tokens_none(): assert fmt_tokens(None) == "?" def test_fmt_tokens_negative(): assert fmt_tokens(-1) == "?" # --------------------------------------------------------------------------- # fmt_tokens — applied in usage + inline comment bodies (Task 3) # --------------------------------------------------------------------------- def test_collapsible_usage_renders_humanized_tokens(): usage = {"input": 2_071_025, "output": 17303, "reasoning": 0, "cache_read": 0, "cache_write": 0, "total": 2_088_328, "cost": 0.0, "steps": 1, "duration_s": 10.0} block = _render_collapsible_usage(usage, "glm-5.2:cloud", config={}) assert "2,071,025 (2.1M) in" in block assert "17,303 (17.3K) out" in block # --------------------------------------------------------------------------- # Multi-provider equivalent-cost table — Task 10 # --------------------------------------------------------------------------- def test_collapsible_usage_renders_multi_provider_table(): usage = {"input": 1_000_000, "output": 1000, "reasoning": 0, "cache_read": 0, "cache_write": 0, "total": 1_001_000, "cost": 0.0, "steps": 1, "duration_s": 10.0} block = _render_collapsible_usage(usage, "glm-5.2:cloud", config={"compare_against": ["claude-sonnet-5", "gpt-5"]}) assert "Claude Sonnet 5" in block assert "GPT-5" in block assert "| Provider | Cost |" in block def test_collapsible_usage_uses_default_compare_against_when_absent(): usage = {"input": 1_000_000, "output": 0, "reasoning": 0, "cache_read": 0, "cache_write": 0, "total": 1_000_000, "cost": 0.0, "steps": 1, "duration_s": 5.0} block = _render_collapsible_usage(usage, "glm-5.2:cloud", config={}) assert "Claude Sonnet 5" in block assert "GPT-5" in block assert "Gemini 2.5 Pro" in block assert "Grok 4.5" in block def test_collapsible_usage_bolds_cost_target_row(): usage = {"input": 1_000_000, "output": 0, "reasoning": 0, "cache_read": 0, "cache_write": 0, "total": 1_000_000, "cost": 0.0, "steps": 1, "duration_s": 5.0} block = _render_collapsible_usage(usage, "glm-5.2:cloud", config={"cost_target": "gpt-5"}) assert "**GPT-5**" in block assert "Claude Sonnet 5" in block # still in default compare set def test_collapsible_usage_skips_zero_cost_rows(): usage = {"input": 0, "output": 0, "reasoning": 0, "cache_read": 0, "cache_write": 0, "total": 0, "cost": 0.0, "steps": 1, "duration_s": 1.0} block = _render_collapsible_usage(usage, "glm-5.2:cloud", config={}) # With zero tokens, all costs are $0 — skip the entire table. assert "| Provider | Cost |" not in block def test_inline_comment_body_humanized_tokens(): # Value chosen > 1000 so fmt_tokens actually adds the comma + short suffix; # the plan's 362 would render identically with or without fmt_tokens. f = {"severity": "medium", "path": "x.py", "line": 1, "problem": "p", "fix": "", "suggestion": "", "reference": "", "_tok_attrib": 17303, "_tok_pct": 0.11} body = inline_comment_body(f) assert "🪙 ~17,303 (17.3K) tok" in body # --------------------------------------------------------------------------- # Severity levels — Task 4 (add trivial + info) # --------------------------------------------------------------------------- def test_severities_includes_trivial_and_info(): assert "trivial" in SEVERITIES assert "info" in SEVERITIES def test_severity_rank_orders_new_levels(): assert SEVERITY_RANK["info"] < SEVERITY_RANK["trivial"] < SEVERITY_RANK["low"] def test_threshold_medium_keeps_low_below_trivial_below_info(): # medium+ threshold: # medium (rank 2) → kept # low (rank 1) → DROPPED # trivial (rank 0) → DROPPED # info (rank -1) → DROPPED cfg = {"style": "lenient", "severity_threshold": "medium"} findings = [ {"severity": "info", "path": "a", "line": 1, "problem": "", "fix": "", "suggestion": "", "reference": ""}, {"severity": "trivial", "path": "b", "line": 1, "problem": "", "fix": "", "suggestion": "", "reference": ""}, {"severity": "low", "path": "c", "line": 1, "problem": "", "fix": "", "suggestion": "", "reference": ""}, {"severity": "medium", "path": "d", "line": 1, "problem": "", "fix": "", "suggestion": "", "reference": ""}, ] kept, dropped = ai_review.apply_repo_config(findings, cfg, changed_paths=["x.py"]) sev_kept = [f["severity"] for f in kept] sev_dropped = [f["severity"] for f in dropped] assert "info" in sev_dropped assert "trivial" in sev_dropped assert "low" in sev_dropped assert "medium" in sev_kept # and nothing else assert len(kept) == 1 def test_unknown_severity_still_normalizes_to_medium(): # Backward compat n = _normalize_finding({"severity": "emergency", "path": "x", "line": 1, "problem": "p"}) assert n["severity"] == "medium" def test_emoji_for_trivial_and_info_is_neutral(): # The plan's emoji table maps trivial/info to ⚪ assert _SEVERITY_EMOJI["trivial"] == "⚪" assert _SEVERITY_EMOJI["info"] == "⚪" def test_severity_badge_labels_each_known_severity(): # Trivial and info (and legacy nit) should render with their own name, # not fall back to "INFO". for sev in ("critical", "high", "medium", "low", "trivial", "info", "nit"): badge = _severity_badge(sev) assert f"[{sev.upper()}]" in badge, (sev, badge) # --------------------------------------------------------------------------- # parse_repo_config — `enabled` (kill-switch) + `compare_against` (cost roster) # --------------------------------------------------------------------------- def test_parse_repo_config_enabled_true(): cfg = parse_repo_config('{"enabled": true}') assert cfg.get("enabled") is True def test_parse_repo_config_enabled_false_explicit(): cfg = parse_repo_config('{"enabled": false}') assert cfg.get("enabled") is False def test_parse_repo_config_enabled_missing_defaults_false(): cfg = parse_repo_config('{}') assert cfg.get("enabled") is False def test_parse_repo_config_enabled_wrong_type_ignored(): cfg = parse_repo_config('{"enabled": "yes"}') assert cfg.get("enabled") is False def test_parse_repo_config_compare_against_default_absent(): # absent in returned cfg; defaults applied in render, not parse_repo_config cfg = parse_repo_config('{}') assert "compare_against" not in cfg def test_parse_repo_config_compare_against_valid(): cfg = parse_repo_config( '{"compare_against": ["claude-sonnet-5", "gpt-5", "gemini-2.5-pro"]}') assert cfg["compare_against"] == ["claude-sonnet-5", "gpt-5", "gemini-2.5-pro"] def test_parse_repo_config_compare_against_drops_unknown_keys(capfd): cfg = parse_repo_config( '{"compare_against": ["claude-sonnet-5", "bogus-1", "gpt-5"]}') assert "bogus-1" not in cfg["compare_against"] assert "claude-sonnet-5" in cfg["compare_against"] captured = capfd.readouterr() assert "bogus-1" in captured.err def test_parse_repo_config_compare_against_caps_at_12(monkeypatch): """13+ valid keys must be truncated to the first 12; invalid keys are dropped and do not count. Inject a 13th PRICES entry via monkeypatch so the [:12] cap actually fires (cost_model.PRICES has exactly 12 keys today, which would otherwise make the cap a no-op).""" import cost_model as cm monkeypatch.setitem( cm.PRICES, "fake-model-13", cm.Price("Fake", 1.00, 2.00, 1.00, 0.10)) valid = list(cm.PRICES) # 13 unique keys (12 real + 1 test-only) raw = valid + ["bogus-extra"] # 13 valid + 1 invalid cfg = parse_repo_config(json.dumps({"compare_against": raw})) assert len(cfg["compare_against"]) == 12 assert cfg["compare_against"] == valid[:12] assert "fake-model-13" not in cfg["compare_against"] assert "bogus-extra" not in cfg["compare_against"] # --------------------------------------------------------------------------- # Task 6 — merge_confidence + REVIEW_HEADER confidence badge # --------------------------------------------------------------------------- def test_merge_confidence_clean_is_five(): assert merge_confidence([]) == 5 def test_merge_confidence_only_low_is_five(): f = {"severity": "low"} assert merge_confidence([f, f, f]) == 5 def test_merge_confidence_medium_drops_one(): f = {"severity": "medium"} assert merge_confidence([f]) == 4 def test_merge_confidence_high_drops_two(): f = {"severity": "high"} assert merge_confidence([f]) == 3 def test_merge_confidence_critical_drops_to_one(): f = {"severity": "critical"} assert merge_confidence([f]) == 1 def test_merge_confidence_multi_lens_drops_extra(): # The flag has moved to a kwarg; passing `_multi_lens` on the dict is no # longer enough — the kwarg is the only path that drops the score. f = {"severity": "low"} assert merge_confidence([f], multi_lens_observed=True) == 4 def test_merge_confidence_multi_lens_survives_normalization(): """Real flow: `_multi_lens` is set on the raw finding, but stripped by `_normalize_finding`. `merge_confidence(...)` with only the kwarg sees a normalized finding; the dedup must be triggered by `multi_lens_observed=` being true, not by reading `_multi_lens` off the dict.""" raw = {"_multi_lens": True, "severity": "low", "path": "x", "line": 1, "problem": "p", "fix": "", "suggestion": "", "reference": ""} normalized = _normalize_finding(raw) assert "_multi_lens" not in normalized # confirms the strip # Now call merge_confidence the way review_pr will: assert merge_confidence([normalized], multi_lens_observed=True) == 4 # And without the kwarg, the flag-on-dict path is gone: assert merge_confidence([normalized]) == 5 def test_merge_confidence_clamped(): # Three critical findings must NOT take the score below 1. f = {"severity": "critical"} assert merge_confidence([f, f, f]) == 1 def test_review_header_includes_confidence(): # REVIEW_HEADER gains a {confidence} placeholder; verify the format works. h = REVIEW_HEADER.format(model="glm-5.2:cloud", sha="abc1234567", confidence="3/5 🟡") assert "Merge confidence: 3/5 🟡" in h def test_confidence_badge_table_complete(): # Sanity-check the badge table the render layer reads from. assert _CONFIDENCE_BADGE == {5: "🟢", 4: "🟢", 3: "🟡", 2: "🟠", 1: "🔴"} def test_format_review_body_default_confidence_is_green(): # Default confidence kwarg should produce a green 5/5 badge in the header, # matching the pre-existing "clean PR" semantics. body = format_review_body("- [high] x:1 — bug", "glm-5.2:cloud", "abcdef1234567890") assert "Merge confidence: 5/5 🟢" in body def test_format_review_body_low_confidence_shows_red_badge(): body = format_review_body( "- [critical] x:1 — bug", "glm-5.2:cloud", "abcdef1234567890", confidence=1, ) assert "Merge confidence: 1/5 🔴" in body def test_format_review_body_confidence_clamps_out_of_range(): # Out-of-range confidence is clamped to [1, 5] in the badge string. body_hi = format_review_body("- x", "glm-5.2:cloud", "abcdef1234567890", confidence=99) assert "Merge confidence: 5/5 🟢" in body_hi body_lo = format_review_body("- x", "glm-5.2:cloud", "abcdef1234567890", confidence=0) assert "Merge confidence: 1/5 🔴" in body_lo