pragent/tests/pilot/test_ai_review.py

"""Unit tests for pragent pilot pure helpers. No network."""
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"))

from ai_review import (  # noqa: E402
    build_user_prompt,
    compute_attribution,
    format_review_body,
    format_usage_section,
    inline_comment_body,
    parse_diff_anchors,
    parse_findings,
    parse_repo_config,
    parse_review_output,
    parse_text_blocks,
    prior_review_bodies,
    reviewed_shas,
    split_findings,
    summary_bullets,
    truncate_diff,
)

# ---------------------------------------------------------------------------
# truncate_diff
# ---------------------------------------------------------------------------


def test_truncate_diff_short():
    text, truncated, n = truncate_diff("abc", 100)
    assert text == "abc"
    assert truncated is False
    assert n == 3


def test_truncate_diff_exact_boundary():
    text, truncated, n = truncate_diff("x" * 100, 100)
    assert truncated is False
    assert n == 100
    assert text == "x" * 100


def test_truncate_diff_over_cap():
    text, truncated, n = truncate_diff("x" * 250, 100)
    assert truncated is True
    assert n == 250
    assert text.startswith("x" * 100)
    assert "[diff truncated at 100 characters]" in text


def test_truncate_diff_none():
    text, truncated, n = truncate_diff(None, 100)  # type: ignore[arg-type]
    assert text == ""
    assert truncated is False
    assert n == 0


# ---------------------------------------------------------------------------
# parse_text_blocks
# ---------------------------------------------------------------------------


def test_parse_text_blocks_text_only():
    content = [{"type": "text", "text": "hello"}, {"type": "text", "text": "world"}]
    assert parse_text_blocks(content) == "hello\nworld"


def test_parse_text_blocks_drops_thinking():
    content = [
        {"type": "thinking", "thinking": "reasoning here"},
        {"type": "text", "text": "- [high] a.go:3 — bug. fix."},
    ]
    assert parse_text_blocks(content) == "- [high] a.go:3 — bug. fix."


def test_parse_text_blocks_empty_and_malformed():
    assert parse_text_blocks([]) == ""
    assert parse_text_blocks(None) == ""  # type: ignore[arg-type]
    assert parse_text_blocks([{"type": "text"}, "garbage", 5]) == ""


def test_parse_text_blocks_real_glm_shape():
    # Captured from glm-5.2:cloud via headroom 8789.
    content = [
        {"type": "thinking", "thinking": "Analyze the request..."},
        {"type": "text", "text": "- [critical] auth.py:12 — token compared with `==`. Use hmac.compare_digest."},
    ]
    assert "compare_digest" in parse_text_blocks(content)


# ---------------------------------------------------------------------------
# format_review_body
# ---------------------------------------------------------------------------


def test_format_review_body_findings():
    body = format_review_body("- [high] x:1 — bug. fix.", "glm-5.2:cloud", "abcdef1234567890")
    assert "pragent pilot" in body
    assert "glm-5.2:cloud" in body
    assert "`abcdef12`" in body  # 8-char sha
    assert "- [high] x:1" in body


def test_format_review_body_empty_findings():
    body = format_review_body("", "glm-5.2:cloud", "abcdef1234567890")
    assert "No issues found." in body


def test_format_review_body_whitespace_findings():
    body = format_review_body("   \n  ", "glm-5.2:cloud", "abcdef1234567890")
    assert "No issues found." in body


def test_format_review_body_no_sha():
    body = format_review_body("- [low] y:2 — nit", "glm-5.2:cloud", "")
    assert "`unknown`" in body


# ---------------------------------------------------------------------------
# build_user_prompt
# ---------------------------------------------------------------------------


def test_build_user_prompt_includes_title_and_diff():
    p = build_user_prompt("Fix login", "Closes #1", "diff --git a/x b/x")
    assert "Fix login" in p
    assert "Closes #1" in p
    assert "diff --git a/x b/x" in p


def test_build_user_prompt_truncates_long_body():
    long_body = "B" * 6000
    p = build_user_prompt("t", long_body, "d")
    assert "[PR body truncated]" in p
    assert p.count("B") < 6000


def test_build_user_prompt_no_body():
    p = build_user_prompt("t", "", "d")
    assert "Description:" not in p


def test_build_user_prompt_with_config_and_prior():
    cfg = {"focus": ["security"], "instructions": "Use Result<T,E>."}
    prior = ["🤖 **AI Review** …\n- [high] x:1 — old."]
    p = build_user_prompt("t", "b", "diff --git a/x b/x", config=cfg, prior_reviews=prior)
    assert "## Repo review config" in p
    assert "security" in p
    assert "Result<T,E>" in p
    assert "## PREVIOUS REVIEWS" in p
    assert "old." in p


# ---------------------------------------------------------------------------
# parse_diff_anchors
# ---------------------------------------------------------------------------


_DIFF = """\
diff --git a/src/a.py b/src/a.py
index 1..2 100644
--- a/src/a.py
+++ b/src/a.py
@@ -1,4 +1,5 @@
 context
-removed
+added
 context2
@@ -10,3 +10,4 @@
 keep
+new
 last
diff --git a/binary.bin b/binary.bin
new file mode 100644
index 0..1
Binary files differ
"""


def test_parse_diff_anchors_context_and_added():
    a = parse_diff_anchors(_DIFF)
    # context(1), +added(2), context2(3) | keep(10), +new(11), last(12)
    assert a["src/a.py"] == {1, 2, 3, 10, 11, 12}
    # removed line (-removed, old line 2) has no new-line anchor
    assert 2 in a["src/a.py"]  # 2 here is the +added line, not the removed one


def test_parse_diff_anchors_binary_file_present_no_lines():
    a = parse_diff_anchors(_DIFF)
    assert "binary.bin" in a
    assert a["binary.bin"] == set()


def test_parse_diff_anchors_empty():
    assert parse_diff_anchors("") == {}
    assert parse_diff_anchors(None) == {}  # type: ignore[arg-type]


def test_parse_diff_anchors_new_file():
    diff = "diff --git a/new.ts b/new.ts\nnew file mode 100644\n--- /dev/null\n+++ b/new.ts\n@@ -0,0 +1,3 @@\n+a\n+b\n+c\n"
    a = parse_diff_anchors(diff)
    assert a["new.ts"] == {1, 2, 3}


# ---------------------------------------------------------------------------
# parse_findings
# ---------------------------------------------------------------------------


def test_parse_findings_clean_json():
    txt = '{"findings":[{"severity":"high","path":"a.py","line":3,"problem":"x","fix":"y","suggestion":"z"}]}'
    fs = parse_findings(txt)
    assert len(fs) == 1
    assert fs[0]["severity"] == "high"
    assert fs[0]["path"] == "a.py"
    assert fs[0]["line"] == 3


def test_parse_findings_fenced_json():
    txt = '```json\n{"findings":[{"severity":"low","path":"b.go","line":1,"problem":"p","fix":"","suggestion":""}]}\n```'
    fs = parse_findings(txt)
    assert len(fs) == 1
    assert fs[0]["path"] == "b.go"


def test_parse_findings_json_in_prose():
    txt = 'Here is my review: {"findings":[{"severity":"critical","path":"c","line":9,"problem":"q"}]} thanks!'
    fs = parse_findings(txt)
    assert len(fs) == 1
    assert fs[0]["severity"] == "critical"


def test_parse_findings_empty():
    assert parse_findings('{"findings":[]}') == []
    assert parse_findings("") == []
    assert parse_findings("not json at all") == []


def test_parse_findings_drops_bad_entries():
    # missing path, bad line, unknown severity (normalised)
    txt = '{"findings":[{"line":1},{"path":"x","line":-1},{"path":"x","line":2,"severity":"bogus","problem":"p"}]}'
    fs = parse_findings(txt)
    assert len(fs) == 1
    assert fs[0]["severity"] == "medium"


# ---------------------------------------------------------------------------
# split_findings + inline_comment_body + summary_bullets
# ---------------------------------------------------------------------------


def test_split_findings_by_anchor():
    anchors = {"a.py": {1, 3, 4}}
    fs = [
        {"severity": "high", "path": "a.py", "line": 3, "problem": "p", "fix": "f", "suggestion": ""},
        {"severity": "low", "path": "a.py", "line": 99, "problem": "off", "fix": "", "suggestion": ""},
        {"severity": "medium", "path": "other.go", "line": 1, "problem": "x", "fix": "", "suggestion": ""},
    ]
    anchored, unanchored = split_findings(fs, anchors)
    assert [f["line"] for f in anchored] == [3]
    assert len(unanchored) == 2


def test_inline_comment_body_with_suggestion():
    f = {"severity": "high", "path": "a", "line": 1, "problem": "bad", "fix": "swap", "suggestion": "good()"}
    body = inline_comment_body(f)
    assert "**[HIGH]**" in body
    assert "bad" in body
    # no extension → bare fence (Gitea 1.26.x has no apply-suggestion; we tag
    # with the file language for highlighting instead of ```suggestion)
    assert "```\ngood()\n```" in body
    assert "good()" in body


def test_inline_comment_body_suggestion_lang_tagged():
    f = {"severity": "high", "path": "src/Foo.java", "line": 1,
         "problem": "bad", "fix": "swap", "suggestion": "good();"}
    body = inline_comment_body(f)
    assert "```java\ngood();\n```" in body


def test_inline_comment_body_no_suggestion():
    f = {"severity": "low", "path": "a", "line": 1, "problem": "p", "fix": "f", "suggestion": ""}
    body = inline_comment_body(f)
    assert "```" not in body
    assert "Fix: f" in body


def test_summary_bullets_format():
    fs = [{"severity": "high", "path": "a.py", "line": 7, "problem": "p", "fix": "f", "suggestion": ""}]
    b = summary_bullets(fs)
    assert "- **[HIGH]**" in b
    assert "`a.py:7`" in b


# ---------------------------------------------------------------------------
# repo config parsing
# ---------------------------------------------------------------------------


def test_parse_repo_config_full():
    raw = '{"focus":["security","perf"],"exclude_paths":["vendor/**"],"languages":["go"],"instructions":"be strict"}'
    c = parse_repo_config(raw)
    assert c["focus"] == ["security", "perf"]
    assert c["exclude_paths"] == ["vendor/**"]
    assert c["instructions"] == "be strict"


def test_parse_repo_config_partial_and_bad():
    assert parse_repo_config('{"focus":"not-a-list"}') == {}
    assert parse_repo_config('{"focus":["ok"]}') == {"focus": ["ok"]}
    assert parse_repo_config("") == {}
    assert parse_repo_config("not json") == {}
    assert parse_repo_config('{"instructions":"   "}') == {}


# ---------------------------------------------------------------------------
# dedupe / prior-context parsing
# ---------------------------------------------------------------------------


def test_reviewed_shas_extracts_marker():
    reviews = [
        {"body": "🤖 AI Review · glm · `abcdef12`\n\n<!-- pragent:sha=abcdef1234567890 -->"},
        {"body": "human comment, no marker"},
        {"body": "<!-- pragent:sha=0987654 -->"},
    ]
    shas = reviewed_shas(reviews)
    assert "abcdef1234567890" in shas
    assert "0987654" in shas


def test_reviewed_shas_empty():
    assert reviewed_shas([]) == set()
    assert reviewed_shas([{"body": "no marker"}]) == set()


def test_prior_review_bodies_skips_current_sha():
    reviews = [
        {"body": "r1\n<!-- pragent:sha=1111111 -->"},
        {"body": "r2\n<!-- pragent:sha=2222222 -->"},
        {"body": "no marker here"},
    ]
    prior = prior_review_bodies(reviews, current_sha="2222222")
    assert len(prior) == 1
    assert "r1" in prior[0]


def test_format_review_body_has_sha_marker():
    body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890")
    assert "<!-- pragent:sha=abcdef1234567890 -->" in body


# ---------------------------------------------------------------------------
# parse_review_output (opencode engine: {summary, findings} + reference)
# ---------------------------------------------------------------------------


def test_parse_review_output_summary_and_findings():
    txt = (
        "This PR adds an eval helper — risky. See findings.\n\n"
        "```json\n"
        '{"summary":"Adds eval() — security risk.","findings":['
        '{"severity":"critical","path":"src/x.ts","line":4,"problem":"eval on user input",'
        '"fix":"parse explicitly","suggestion":"const n = Number(s)","reference":"https://owasp.org/x"}'
        "]}",
        "\n```",
    )
    summary, fs = parse_review_output("".join(txt))
    assert "eval()" in summary
    assert len(fs) == 1
    assert fs[0]["severity"] == "critical"
    assert fs[0]["reference"] == "https://owasp.org/x"
    assert fs[0]["suggestion"] == "const n = Number(s)"


def test_parse_review_output_bare_findings_no_summary():
    txt = '```json\n{"findings":[{"severity":"low","path":"a","line":1,"problem":"p"}]}\n```'
    summary, fs = parse_review_output(txt)
    assert summary == ""
    assert len(fs) == 1
    assert fs[0]["reference"] == ""  # default


def test_parse_review_output_empty_and_bogus():
    assert parse_review_output("") == ("", [])
    assert parse_review_output("no json here") == ("", [])
    assert parse_review_output('{"findings":[]}') == ("", [])


def test_parse_review_output_uses_last_json_block():
    # Agent emits a stray json-ish block first, then the real one last.
    txt = (
        "```json\n{\"findings\":[{\"path\":\"x\",\"line\":1,\"severity\":\"low\"}]}\n```\n"
        "more prose\n"
        "```json\n{\"summary\":\"real\",\"findings\":[{\"path\":\"y\",\"line\":2,\"severity\":\"high\"}]}\n```"
    )
    summary, fs = parse_review_output(txt)
    assert summary == "real"
    assert len(fs) == 1
    assert fs[0]["path"] == "y"


# ---------------------------------------------------------------------------
# reference rendering in inline_comment_body + summary_bullets + summary section
# ---------------------------------------------------------------------------


def test_inline_comment_body_renders_reference():
    f = {"severity": "high", "path": "a", "line": 1, "problem": "p", "fix": "f",
         "suggestion": "", "reference": "https://cve.example/X"}
    body = inline_comment_body(f)
    assert "📎 ref: https://cve.example/X" in body


def test_inline_comment_body_no_reference_no_ref_line():
    f = {"severity": "low", "path": "a", "line": 1, "problem": "p", "fix": "",
         "suggestion": "", "reference": ""}
    assert "📎 ref" not in inline_comment_body(f)


def test_summary_bullets_renders_reference():
    fs = [{"severity": "high", "path": "a.py", "line": 7, "problem": "p", "fix": "f",
           "suggestion": "", "reference": "https://r.example"}]
    b = summary_bullets(fs)
    assert "https://r.example" in b
    assert "`a.py:7`" in b


def test_format_review_body_with_summary_section():
    body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890",
                              summary="This PR adds a risky helper.")
    assert "This PR adds a risky helper." in body
    assert "- [high] x:1" in body
    assert "<!-- pragent:sha=abcdef1234567890 -->" in body
    # summary appears before the findings bullets
    assert body.index("risky helper") < body.index("[high]")


# ---------------------------------------------------------------------------
# AI-USAGE: compute_attribution + format_usage_section + inline 🪙 line
# ---------------------------------------------------------------------------


def test_compute_attribution_weighted_split():
    # weights 1 (problem="a") and 3 (problem="aaa"), output 100 → 25 / 75
    fs = [
        {"severity": "high", "path": "x", "line": 1, "problem": "a", "fix": "", "suggestion": ""},
        {"severity": "low", "path": "x", "line": 2, "problem": "aaa", "fix": "", "suggestion": ""},
    ]
    compute_attribution(fs, 100)
    assert fs[0]["_tok_attrib"] == 25
    assert fs[1]["_tok_attrib"] == 75
    assert abs(fs[0]["_tok_pct"] - 0.25) < 1e-9
    assert abs(fs[1]["_tok_pct"] - 0.75) < 1e-9


def test_compute_attribution_zero_weights_splits_evenly():
    fs = [
        {"severity": "high", "path": "x", "line": 1, "problem": "", "fix": "", "suggestion": ""},
        {"severity": "low", "path": "x", "line": 2, "problem": "", "fix": "", "suggestion": ""},
    ]
    compute_attribution(fs, 80)
    assert fs[0]["_tok_attrib"] == 40
    assert fs[1]["_tok_attrib"] == 40
    assert abs(fs[0]["_tok_pct"] - 0.5) < 1e-9


def test_compute_attribution_noop_on_empty_or_zero_budget():
    fs = [{"severity": "high", "path": "x", "line": 1, "problem": "a", "fix": "", "suggestion": ""}]
    compute_attribution([], 100)
    compute_attribution(fs, 0)
    assert "_tok_attrib" not in fs[0]


def test_inline_comment_body_with_attribution_line():
    f = {"severity": "high", "path": "a", "line": 1, "problem": "bad", "fix": "swap",
         "suggestion": "", "_tok_attrib": 180, "_tok_pct": 0.29}
    body = inline_comment_body(f)
    assert "🪙 ~180 tok" in body
    assert "29%" in body
    assert "attributed output" in body


def test_inline_comment_body_no_attribution_no_coin_line():
    f = {"severity": "high", "path": "a", "line": 1, "problem": "bad", "fix": "swap",
         "suggestion": ""}
    assert "🪙" not in inline_comment_body(f)


def test_format_usage_section_renders_totals_and_table():
    fs = [
        {"severity": "critical", "path": "src/Foo.java", "line": 98,
         "problem": "p"*10, "fix": "f", "suggestion": "", "_tok_attrib": 180, "_tok_pct": 0.29},
    ]
    usage = {"input": 18420, "output": 612, "reasoning": 0, "cache_read": 15210,
             "cache_write": 0, "total": 19032, "cost": 0.0, "steps": 7, "duration_s": 142.0}
    sec = format_usage_section(usage, fs, "glm-5.2:cloud")
    assert "## 🔋 AI usage" in sec
    assert "`glm-5.2:cloud`" in sec
    assert "agent steps: 7" in sec
    assert "duration: 142.0s" in sec
    assert "18420 in" in sec and "612 out" in sec and "19032 total" in sec
    assert "$0.00" in sec
    assert "whole-repo checkout" in sec
    assert "attributed" in sec
    # table
    assert "| severity | location | ≈out tok | % |" in sec
    assert "CRITICAL" in sec
    assert "`src/Foo.java:98`" in sec
    assert "180" in sec and "29%" in sec


def test_format_usage_section_omits_table_when_no_attributed_rows():
    usage = {"input": 10, "output": 0, "reasoning": 0, "cache_read": 0,
             "cache_write": 0, "total": 10, "cost": 0.0, "steps": 1, "duration_s": 1.0}
    sec = format_usage_section(usage, [], "glm-5.2:cloud")
    assert "## 🔋 AI usage" in sec
    assert "severity | location" not in sec  # no rows → no table


def test_format_usage_section_none_returns_empty():
    assert format_usage_section(None, [], "m") == ""


def test_format_usage_section_cost_nonzero():
    usage = {"input": 10, "output": 0, "reasoning": 0, "cache_read": 0,
             "cache_write": 0, "total": 10, "cost": 0.0123, "steps": 1, "duration_s": 1.0}
    sec = format_usage_section(usage, [], "m")
    assert "$0.0123" in sec
    assert "billed by provider" in sec


def test_format_review_body_usage_section_between_summary_and_findings():
    usage_sec = "## 🔋 AI usage\n\n- model: `m`"
    body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890",
                              summary="This PR is risky.", usage_section=usage_sec)
    # order: header < summary < usage < findings < marker
    assert body.index("risky.") < body.index("AI usage")
    assert body.index("AI usage") < body.index("[high]")
    assert body.index("[high]") < body.index("<!-- pragent:sha=")
    assert "## 🔋 AI usage" in body


def test_format_review_body_no_usage_section_omitted():
    body = format_review_body("- [high] x:1 — b", "glm-5.2:cloud", "abcdef1234567890")
    assert "AI usage" not in body