"""Tests for the unified-review scoring module (score_review metrics).""" from __future__ import annotations import sys from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).parent.parent)) from code_review_graph.scoring import ( # noqa: E402 THRESHOLDS, _grade, compute_exception_coverage, compute_redundancy_rate, compute_sql_risk, compute_vulnerability_heuristic, dedupe_findings, ) @pytest.fixture def tmp_repo(tmp_path: Path) -> Path: """A small repository root with one source file.""" Path(tmp_path, "app.py").write_text( 'import sqlite3\n' 'def query(conn, user_id):\n' ' conn.execute("SELECT * FROM users WHERE id = " + str(user_id))\n' 'def safe(conn, user_id):\n' ' conn.execute("SELECT * FROM users WHERE id = ?", (user_id,))\n' 'def risky(a):\n' ' try:\n' ' return compute(a)\n' ' except Exception as e:\n' ' return None\n', encoding="utf-8", ) return tmp_path # --------------------------------------------------------------------------- # _grade threshold boundaries # --------------------------------------------------------------------------- class TestGradeThresholds: def test_sql_risk_counts(self): # 0 = good, warn_at count = warn, fail_at count = fail. assert _grade("sql_risk", 0) == "good" assert _grade("sql_risk", 1) == "warn" assert _grade("sql_risk", 2) == "warn" assert _grade("sql_risk", 3) == "fail" def test_vulnerability_risk_counts(self): assert _grade("vulnerability_risk", 0) == "good" assert _grade("vulnerability_risk", 1) == "warn" assert _grade("vulnerability_risk", 2) == "fail" def test_exception_coverage_pct(self): assert _grade("exception_coverage", 60.0) == "good" assert _grade("exception_coverage", 40.0) == "warn" assert _grade("exception_coverage", 20.0) == "fail" def test_redundancy_rate_pct(self): assert _grade("redundancy_rate", 5.0) == "good" assert _grade("redundancy_rate", 15.0) == "warn" assert _grade("redundancy_rate", 25.0) == "fail" # --------------------------------------------------------------------------- # Metric functions # --------------------------------------------------------------------------- class TestSQLRisk: def test_detects_interpolated_sql(self, tmp_repo): result = compute_sql_risk(["app.py"], tmp_repo) assert result["value"] >= 1 assert result["grade"] in ("warn", "fail") assert any("line" in loc for loc in result["evidence"]) def test_clean_file_no_risk(self, tmp_path): Path(tmp_path, "clean.py").write_text( "def f(a):\n return a * 2\n", encoding="utf-8", ) result = compute_sql_risk(["clean.py"], tmp_path) assert result["value"] == 0 assert result["grade"] == "good" class TestExceptionCoverage: def test_counts_exception_paths(self, tmp_repo): result = compute_exception_coverage(["app.py"], tmp_repo) # app.py has a try/except path, so exception coverage > 0. assert result["value"] > 0 assert "exception_path_lines" in result["evidence"] def test_empty_file(self, tmp_path): Path(tmp_path, "e.py").write_text("# comment only\n", encoding="utf-8") result = compute_exception_coverage(["e.py"], tmp_path) assert result["value"] == 0.0 class TestRedundancy: def test_repeated_blocks_detected(self, tmp_path): body = ( "def transform_item(item):\n" " return item.strip().lower().replace(' ', '_')\n" ) Path(tmp_path, "r.py").write_text( body + body + body + "def other(x):\n return x\n", encoding="utf-8", ) result = compute_redundancy_rate(["r.py"], tmp_path) # The shared normalised transform line appears >=3 times. assert result["value"] > 0 assert len(result["evidence"]) >= 1 def test_no_redundancy(self, tmp_path): Path(tmp_path, "n.py").write_text( "def a(x):\n return x\n" "def b(y):\n return y * 2\n" "def c(z):\n return z - 1\n", encoding="utf-8", ) result = compute_redundancy_rate(["n.py"], tmp_path) assert result["value"] == 0.0 class TestVulnerabilityHeuristic: def test_detects_secret_like_pattern(self, tmp_path): Path(tmp_path, "s.py").write_text( 'password = "hunter2"\ndef f():\n pass\n', encoding="utf-8", ) result = compute_vulnerability_heuristic(["s.py"], tmp_path) assert result["value"] >= 1 assert result["grade"] in ("warn", "fail") # --------------------------------------------------------------------------- # dedupe_findings # --------------------------------------------------------------------------- class TestDedupeFindings: def test_same_fingerprint_merges(self): findings = [ {"path": "a.py", "line": 1, "category": "sql", "severity": "CRITICAL", "confidence": 8, "source": "main"}, {"path": "a.py", "line": 1, "category": "sql", "severity": "CRITICAL", "confidence": 6, "source": "security"}, ] result = dedupe_findings(findings) assert len(result["merged"]) == 1 # Multi-source confirmed: confidence 8 -> 9 (cap 10). assert result["merged"][0]["confidence"] == 9.0 assert result["merged"][0]["multi_source_confirmed"] is True def test_confidence_cap_at_10(self): findings = [ {"path": "a.py", "line": 1, "category": "sql", "severity": "CRITICAL", "confidence": 10, "source": "main"}, {"path": "a.py", "line": 1, "category": "sql", "severity": "CRITICAL", "confidence": 10, "source": "security"}, ] result = dedupe_findings(findings) assert result["merged"][0]["confidence"] == 10.0 def test_low_confidence_suppressed(self): findings = [ {"path": "a.py", "line": 1, "category": "style", "severity": "INFORMATIONAL", "confidence": 1, "source": "main"}, ] result = dedupe_findings(findings) assert result["merged"] == [] assert result["suppressed_by_confidence"] == 1 def test_appendix_routing(self): findings = [ {"path": "a.py", "line": 1, "category": "style", "severity": "INFORMATIONAL", "confidence": 3, "source": "main"}, ] result = dedupe_findings(findings) assert result["merged"][0]["display"] == "appendix" def test_quality_score_formula(self): findings = [ {"path": "a.py", "line": 1, "category": "sql", "severity": "CRITICAL", "confidence": 9, "source": "main"}, {"path": "b.py", "line": 2, "category": "style", "severity": "INFORMATIONAL", "confidence": 7, "source": "main"}, {"path": "c.py", "line": 3, "category": "style", "severity": "INFORMATIONAL", "confidence": 7, "source": "main"}, ] result = dedupe_findings(findings) # 10 - (1*2 + 2*0.5) = 10 - 3 = 7.0 assert result["quality_score"] == 7.0 assert result["counts"] == {"critical": 1, "informational": 2} def test_prior_suppression(self): findings = [ {"path": "a.py", "line": 1, "category": "style", "severity": "INFORMATIONAL", "confidence": 8, "source": "main"}, ] prior = [ {"path": "a.py", "line": 1, "category": "style"}, ] result = dedupe_findings(findings, suppress_prior=prior) assert result["merged"] == [] assert result["suppressed_by_prior"] == 1 def test_distinct_fingerprints_not_merged(self): findings = [ {"path": "a.py", "line": 1, "category": "sql", "severity": "CRITICAL", "confidence": 8, "source": "main"}, {"path": "a.py", "line": 2, "category": "sql", "severity": "CRITICAL", "confidence": 8, "source": "main"}, ] result = dedupe_findings(findings) assert len(result["merged"]) == 2