diff --git a/scripts/confidence.py b/scripts/confidence.py new file mode 100644 index 00000000..813e4a37 --- /dev/null +++ b/scripts/confidence.py @@ -0,0 +1,320 @@ +# @WHO: scripts/confidence.py +# @WHAT: Confidence scoring for findings + schema versioning (issue #5) +# @PART: engine +# @ENTRY: score_finding(), SCHEMA_VERSION, stamp_schema_version() +"""Confidence scoring and schema versioning for CodeLens findings (issue #5). + +Two agent-UX features live here: + +1. **Confidence scores** — each finding gets a ``confidence`` field in [0.0, 1.0] + so agents can filter actionable findings vs. ones that need human review + without discarding everything. High-confidence findings (>= 0.9) are safe + to auto-triage; low-confidence ones (< 0.5) should be surfaced for review. + +2. **Schema versioning** — every JSON output gets a ``schema_version`` field + so consumers can handle breaking changes gracefully. The version follows + the CodeLens release version (e.g. ``"8.2"``). Bumped on any breaking + change to the AI-normalized output schema. + +Confidence model +---------------- +Confidence is *category-driven* with *modifier adjustments*. The base score +reflects how reliable a detection category is in general; modifiers nudge +the score up or down based on per-finding signals (e.g. test-file context +reduces confidence, dynamic-access patterns reduce it further). + +Examples:: + + score_finding("dead_code", "unreachable", {"after": "return"}) + # -> 0.95 (code after return is almost certainly dead) + + score_finding("dead_code", "unused_exports", {"source": "library_api"}) + # -> 0.40 (library exports are often public API; low confidence) + + score_finding("secrets", "pattern_match", {"in_test_file": True}) + # -> 0.55 (test-file match is often a fixture, not a real secret) + +The scores below are starting heuristics — they should be tuned over time +as we collect false-positive / false-negative data. Agents should treat +them as *ranking signals*, not ground truth. +""" + +from __future__ import annotations + +import os +from typing import Any, Dict, Optional + + +# ─── Schema version ──────────────────────────────────────────────────────── + +# Bumped on any breaking change to the AI-normalized output schema +# (formatters/_normalize_to_ai). Agents can read this to decide whether +# their parser still understands the output. +# +# History: +# 8.2 — initial schema_version field (issue #5) +SCHEMA_VERSION = "8.2" + + +def stamp_schema_version(payload: Dict[str, Any]) -> Dict[str, Any]: + """Add ``schema_version`` to a payload dict (in place + returned). + + Used by :func:`formatters._normalize_to_ai` so every MCP/JSON output + carries the version. Idempotent — if the field is already present it + is preserved (callers may override for testing). + """ + if isinstance(payload, dict) and "schema_version" not in payload: + payload["schema_version"] = SCHEMA_VERSION + return payload + + +# ─── Confidence base scores per category ─────────────────────────────────── +# +# These reflect *general reliability* of each detection category. Categories +# that rely on simple syntactic patterns score high; categories that depend +# on cross-file resolution or heuristics score lower. +# +# Tuning guidelines: +# 0.95+ — almost never wrong (e.g. code after return) +# 0.80 — usually right, occasional false positive (e.g. unreachable after throw) +# 0.65 — mixed; needs review (e.g. unused variables — could be intentional) +# 0.50 — often wrong; flag for review (e.g. zombie CSS — class may be dynamic) +# 0.35 — low confidence; almost certainly needs review (e.g. library exports) + +_DEAD_CODE_BASE: Dict[str, float] = { + "unreachable": 0.95, # code after return/throw/break — syntactic certainty + "unused_vars": 0.70, # could be intentional (debugging, future use) + "dead_listeners": 0.60, # element may be dynamically created + "zombie_css": 0.50, # classes often referenced dynamically + "unused_exports": 0.45, # public API — could be consumed externally + "registry_dead": 0.75, # ref_count==0 from scan — usually right but + # misses dynamic calls +} + +_SECRETS_BASE: Dict[str, float] = { + "pattern_match": 0.90, # matched a known secret pattern (e.g. AWS key format) + "env_exposed": 0.85, # secret reads from env in source — high signal + "entropy": 0.65, # high-entropy string — could be a hash, UUID, etc. +} + + +# ─── Modifier rules ──────────────────────────────────────────────────────── +# +# Each modifier is ``(predicate, delta)``. Predicates inspect the finding dict +# and return ``True`` when the modifier applies. Deltas are summed onto the +# base score, then clamped to [0.0, 1.0]. +# +# Negative deltas = *reduce* confidence (more likely false positive). +# Positive deltas = *increase* confidence (more likely true positive). + +def _is_test_file(finding: Dict[str, Any]) -> bool: + return bool(finding.get("in_test_file") or finding.get("source") == "test") + + +def _is_config_file(finding: Dict[str, Any]) -> bool: + return finding.get("source") == "config" + + +def _is_library_api(finding: Dict[str, Any]) -> bool: + return finding.get("source") == "library_api" + + +def _is_downgraded(finding: Dict[str, Any]) -> bool: + return bool(finding.get("downgraded")) + + +def _has_after_return(finding: Dict[str, Any]) -> bool: + return finding.get("after") == "return" + + +def _has_after_throw(finding: Dict[str, Any]) -> bool: + return finding.get("after") == "throw" + + +_DEAD_CODE_MODIFIERS = [ + # Test/config files: detections here are less reliable — they're often + # fixtures, examples, or intentionally dead for testing purposes. + (_is_test_file, -0.15), + (_is_config_file, -0.10), + # Library API exports: almost always false positives — public API. + (_is_library_api, -0.20), + # Already-downgraded findings: confidence already reflects doubt. + (_is_downgraded, -0.05), + # `after: return` is more certain than `after: throw` (throw can be + # caught by an outer try/catch in some languages). + (_has_after_return, +0.03), + (_has_after_throw, -0.03), +] + +_SECRETS_MODIFIERS = [ + # Test files: secrets here are almost always fixtures, not real. + (_is_test_file, -0.30), + # Already-downgraded severity: lower confidence. + (_is_downgraded, -0.10), +] + + +# ─── Public API ──────────────────────────────────────────────────────────── + + +def score_finding( + engine: str, + category: str, + finding: Dict[str, Any], +) -> float: + """Return a confidence score in [0.0, 1.0] for a finding. + + Args: + engine: ``"dead_code"`` or ``"secrets"`` (the engine that produced + the finding). + category: The finding's category within the engine (e.g. + ``"unreachable"``, ``"pattern_match"``). + finding: The finding dict. Modifiers inspect fields like + ``in_test_file``, ``source``, ``after``, ``downgraded``. + + Returns: + Float in [0.0, 1.0]. Higher = more confident the finding is real. + + Unknown engines/categories default to ``0.50`` (neutral) so a missing + score never breaks the pipeline — agents can still rank findings, + they just won't get the benefit of category-specific tuning. + """ + if engine == "dead_code": + base = _DEAD_CODE_BASE.get(category, 0.50) + modifiers = _DEAD_CODE_MODIFIERS + elif engine == "secrets": + base = _SECRETS_BASE.get(category, 0.50) + modifiers = _SECRETS_MODIFIERS + else: + return 0.50 + + score = base + for predicate, delta in modifiers: + try: + if predicate(finding): + score += delta + except Exception: + # Never let a modifier crash the scoring — fail safe. + continue + + return max(0.0, min(1.0, round(score, 2))) + + +def enrich_finding( + engine: str, + finding: Dict[str, Any], + *, + category_field: str = "category", + type_field: str = "type", +) -> Dict[str, Any]: + """Add a ``confidence`` field to a finding dict (in place + returned). + + Resolves the scoring category from the finding. For secrets, the + *detection method* lives in ``type`` (pattern_match / entropy / + env_exposed) and is the right key for confidence lookup; ``category`` + holds the specific secret kind (aws_key, etc.) which is too granular. + For dead-code, callers normally pass the category explicitly via + :func:`enrich_findings`; this single-finding helper falls back to + ``type`` then ``category``. + + Idempotent: if ``confidence`` is already present, it is preserved. + """ + if not isinstance(finding, dict): + return finding + if "confidence" in finding: + return finding + + # For secrets, prefer `type` (detection method) over `category` (secret kind). + # For dead-code, prefer `category` (set by callers) over `type` (subkind). + if engine == "secrets": + category = ( + finding.get(type_field) + or finding.get(category_field) + or "unknown" + ) + else: + category = ( + finding.get(category_field) + or finding.get(type_field) + or "unknown" + ) + + finding["confidence"] = score_finding(engine, category, finding) + return finding + + +def enrich_findings( + engine: str, + findings: Dict[str, Any], + *, + results_key: str = "results", + findings_key: str = "findings", +) -> Dict[str, Any]: + """Enrich all findings in a result payload with confidence scores. + + Works for both dead-code (category-keyed dict under ``results``) and + secrets (flat list under ``findings``). Mutates ``findings`` in place + and returns it. + + Args: + engine: ``"dead_code"`` or ``"secrets"``. + findings: The full result payload from the engine. + results_key: Key for category-keyed dict (dead-code style). + findings_key: Key for flat list (secrets style). + + Returns: + The same ``findings`` dict, with ``confidence`` added to each + finding. If the payload shape doesn't match either style, the + payload is returned unchanged (fail-safe). + """ + if not isinstance(findings, dict): + return findings + + # Dead-code style: results = {category: [finding, ...], ...} + if results_key in findings and isinstance(findings[results_key], dict): + for category, items in findings[results_key].items(): + if not isinstance(items, list): + continue + for item in items: + if isinstance(item, dict) and "confidence" not in item: + # For dead-code, the category is the dict key. + item["confidence"] = score_finding(engine, category, item) + + # Secrets style: findings = [finding, ...] + # For secrets, the *detection method* lives in ``type`` (pattern_match, + # entropy, env_exposed) while ``category`` holds the specific secret kind + # (aws_key, github_token, etc.). Confidence is keyed off the detection + # method, so we prefer ``type`` over ``category`` here. + if findings_key in findings and isinstance(findings[findings_key], list): + for item in findings[findings_key]: + if isinstance(item, dict) and "confidence" not in item: + category = item.get("type") or item.get("category") or "unknown" + item["confidence"] = score_finding(engine, category, item) + + return findings + + +# ─── Convenience: confidence distribution ────────────────────────────────── + + +def confidence_distribution(findings: list) -> Dict[str, int]: + """Bucket a list of (already-scored) findings by confidence range. + + Returns a dict with keys ``high`` (>= 0.85), ``medium`` (0.5..0.85), + ``low`` (< 0.5), and ``unscored`` (findings without a ``confidence`` + field). Useful for surfacing a quick reliability summary to agents. + """ + buckets = {"high": 0, "medium": 0, "low": 0, "unscored": 0} + for item in findings: + if not isinstance(item, dict): + continue + c = item.get("confidence") + if c is None: + buckets["unscored"] += 1 + elif c >= 0.85: + buckets["high"] += 1 + elif c >= 0.5: + buckets["medium"] += 1 + else: + buckets["low"] += 1 + return buckets diff --git a/scripts/deadcode_engine.py b/scripts/deadcode_engine.py index d133c107..c9323438 100755 --- a/scripts/deadcode_engine.py +++ b/scripts/deadcode_engine.py @@ -295,7 +295,17 @@ def _classify_source(rel: str) -> str: # Build categories dict (same as results, for API compatibility) categories_dict = {k: v for k, v in results.items() if v} - return { + # v8.2 (issue #5): enrich every finding with a confidence score so + # agents can rank actionable vs. needs-review findings. Scores are + # category-driven with modifier adjustments (test files, library API, + # downgraded severity, etc.). See scripts/confidence.py for the model. + try: + from confidence import enrich_findings + except ImportError: + # Defensive: never let confidence scoring break the engine. + enrich_findings = None + + payload = { "status": "ok", "workspace": workspace, "stats": { @@ -314,6 +324,11 @@ def _classify_source(rel: str) -> str: "duration_ms": int((time.time() - start_time) * 1000), } + if enrich_findings is not None: + payload = enrich_findings("dead_code", payload) + + return payload + def _detect_unreachable_code(content: str, ext: str, rel_path: str) -> List[Dict]: """Detect code that comes after return/throw/break/continue and is therefore unreachable. diff --git a/scripts/formatters/__init__.py b/scripts/formatters/__init__.py index c5b423cf..80d388e6 100644 --- a/scripts/formatters/__init__.py +++ b/scripts/formatters/__init__.py @@ -4,6 +4,21 @@ from typing import Any, Dict from formatters.markdown import to_markdown +# v8.2 (issue #5): every AI-normalized output carries a schema_version so +# downstream agents can detect breaking changes and migrate gracefully. +# The version follows the CodeLens release version. Bump on any breaking +# change to the normalized schema (added/removed/renamed top-level keys, +# changed item shape, etc.). +try: + from confidence import stamp_schema_version, SCHEMA_VERSION +except ImportError: # pragma: no cover - defensive + SCHEMA_VERSION = "8.2" + + def stamp_schema_version(payload): + if isinstance(payload, dict) and "schema_version" not in payload: + payload["schema_version"] = SCHEMA_VERSION + return payload + def _normalize_to_ai(data: Any, command: str = "") -> Dict[str, Any]: """Normalize command output to consistent AI-friendly schema. @@ -20,17 +35,17 @@ def _normalize_to_ai(data: Any, command: str = "") -> Dict[str, Any]: } """ if not isinstance(data, dict): - return {"status": "ok", "items": [data]} + return stamp_schema_version({"status": "ok", "items": [data]}) status = data.get("status", "ok") if status == "error": - return { + return stamp_schema_version({ "status": "error", "command": data.get("command", command), "error": data.get("error", ""), "error_type": data.get("error_type", ""), "suggestion": data.get("suggestion", ""), - } + }) result = { "status": status, @@ -165,6 +180,10 @@ def _normalize_to_ai(data: Any, command: str = "") -> Dict[str, Any]: if "_auto_setup" in data: result["auto_setup"] = data["_auto_setup"] + # v8.2 (issue #5): stamp every AI-normalized output with schema_version + # so downstream agents can detect breaking changes and migrate. + stamp_schema_version(result) + return result @@ -182,5 +201,12 @@ def format_output(data: Any, format_type: str = "json", command: str = "", if format_type == "compact": from formatters.compact import format_compact return format_compact(data, command, workspace) - # Default: JSON + # Default: JSON — stamp schema_version so raw JSON consumers also get the + # version signal (issue #5). Mutates a copy so the caller's dict is not + # modified in place (raw JSON should reflect what the engine returned, + # plus the version stamp). + if isinstance(data, dict): + stamped = dict(data) + stamp_schema_version(stamped) + return json.dumps(stamped, indent=2, ensure_ascii=False) return json.dumps(data, indent=2, ensure_ascii=False) diff --git a/scripts/secrets_engine.py b/scripts/secrets_engine.py index 15ba9677..7b4cdbaf 100755 --- a/scripts/secrets_engine.py +++ b/scripts/secrets_engine.py @@ -589,7 +589,17 @@ def detect_secrets( # ─── Generate recommendations ───────────────────────────── recommendations = _generate_recommendations(findings, env_exposed, stats) - return { + # v8.2 (issue #5): enrich every finding with a confidence score so + # agents can rank actionable vs. needs-review findings. Pattern matches + # score high; entropy-based detections score lower (could be hashes / + # UUIDs); test-file findings are heavily discounted. See + # scripts/confidence.py for the model. + try: + from confidence import enrich_findings as _enrich_findings + except ImportError: + _enrich_findings = None + + payload = { "status": "ok", "workspace": workspace, "severity_filter": severity, @@ -603,6 +613,11 @@ def detect_secrets( "files_skipped_regex_timeout": skipped_regex_timeout, } + if _enrich_findings is not None: + payload = _enrich_findings("secrets", payload) + + return payload + # ─── Pattern-based Scanner ───────────────────────────────────── def _scan_file_patterns(content: str, rel_path: str, ext: str, is_test: bool = False) -> List[Dict[str, Any]]: diff --git a/tests/test_confidence.py b/tests/test_confidence.py new file mode 100644 index 00000000..13aa22ba --- /dev/null +++ b/tests/test_confidence.py @@ -0,0 +1,595 @@ +"""Tests for confidence scoring + schema versioning (issue #5). + +Covers two agent-UX features from issue #5: + +1. **Confidence scores** — every finding gets a ``confidence`` field in + [0.0, 1.0] so agents can rank actionable vs. needs-review findings. + - :func:`score_finding` — base scores per category + modifier adjustments + - :func:`enrich_finding` — single-finding enrichment (idempotent) + - :func:`enrich_findings` — payload-level enrichment (dead-code + secrets shapes) + - :func:`confidence_distribution` — bucket findings by confidence range + +2. **Schema versioning** — every JSON output gets a ``schema_version`` field + so consumers can detect breaking changes. + - :data:`SCHEMA_VERSION` — current version string + - :func:`stamp_schema_version` — idempotent stamping + - Integration: ``formatters._normalize_to_ai`` stamps every normalized output + - Integration: ``formatters.format_output`` stamps raw JSON output + +Integration tests verify that ``deadcode_engine.detect_dead_code`` and +``secrets_engine.detect_secrets`` both return findings with ``confidence`` +fields, and that ``formatters.format_output`` adds ``schema_version`` to +both AI-normalized and raw JSON outputs. +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +import shutil + +import pytest + +# Make scripts/ importable. +SCRIPT_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts" +) +if SCRIPT_DIR not in sys.path: + sys.path.insert(0, SCRIPT_DIR) + +from confidence import ( # noqa: E402 + SCHEMA_VERSION, + score_finding, + enrich_finding, + enrich_findings, + stamp_schema_version, + confidence_distribution, +) + + +# ─── Fixtures ────────────────────────────────────────────────────────────── + + +@pytest.fixture +def workspace(): + """Yield a temporary workspace directory.""" + d = tempfile.mkdtemp(prefix="codelens_conf_test_") + yield d + shutil.rmtree(d, ignore_errors=True) + + +# ─── Schema version ──────────────────────────────────────────────────────── + + +class TestSchemaVersion: + """``schema_version`` must be present on every JSON output (issue #5).""" + + def test_schema_version_is_string(self): + assert isinstance(SCHEMA_VERSION, str) + assert "." in SCHEMA_VERSION # e.g. "8.2" + + def test_stamp_adds_schema_version(self): + payload = {"status": "ok", "items": []} + result = stamp_schema_version(payload) + assert result["schema_version"] == SCHEMA_VERSION + + def test_stamp_is_idempotent(self): + payload = {"status": "ok", "schema_version": "9.9"} + result = stamp_schema_version(payload) + # Existing version preserved — not overwritten. + assert result["schema_version"] == "9.9" + + def test_stamp_handles_non_dict(self): + # Should not crash on non-dict input. + assert stamp_schema_version(None) is None + assert stamp_schema_version([]) == [] + + def test_stamp_returns_same_object(self): + """stamp_schema_version mutates in place AND returns the payload.""" + payload = {"status": "ok"} + result = stamp_schema_version(payload) + assert result is payload + assert "schema_version" in payload + + +# ─── score_finding ───────────────────────────────────────────────────────── + + +class TestScoreFinding: + """``score_finding`` returns a confidence in [0.0, 1.0] per category.""" + + def test_dead_code_unreachable_high_confidence(self): + c = score_finding("dead_code", "unreachable", {"after": "return"}) + assert 0.85 <= c <= 1.0 + # After return is more certain than after throw + c_throw = score_finding("dead_code", "unreachable", {"after": "throw"}) + assert c > c_throw + + def test_dead_code_unused_exports_low_confidence(self): + c = score_finding("dead_code", "unused_exports", {}) + assert 0.0 <= c <= 0.6 # public API — often false positive + + def test_dead_code_library_api_reduces_confidence(self): + base = score_finding("dead_code", "unused_exports", {}) + with_lib = score_finding( + "dead_code", "unused_exports", {"source": "library_api"} + ) + assert with_lib < base + + def test_dead_code_test_file_reduces_confidence(self): + base = score_finding("dead_code", "unused_vars", {}) + with_test = score_finding( + "dead_code", "unused_vars", {"in_test_file": True} + ) + assert with_test < base + + def test_dead_code_config_file_reduces_confidence(self): + base = score_finding("dead_code", "unused_vars", {}) + with_config = score_finding( + "dead_code", "unused_vars", {"source": "config"} + ) + assert with_config < base + + def test_dead_code_unknown_category_defaults_neutral(self): + c = score_finding("dead_code", "bogus_category", {}) + assert c == 0.50 + + def test_secrets_pattern_match_high_confidence(self): + c = score_finding("secrets", "pattern_match", {}) + assert 0.80 <= c <= 1.0 + + def test_secrets_entropy_medium_confidence(self): + c = score_finding("secrets", "entropy", {}) + # Entropy could be a hash, UUID, etc. — not as certain as pattern match. + assert 0.40 <= c <= 0.80 + + def test_secrets_test_file_heavily_discounted(self): + base = score_finding("secrets", "pattern_match", {}) + with_test = score_finding( + "secrets", "pattern_match", {"in_test_file": True} + ) + # Test-file secrets are almost always fixtures — big discount. + assert with_test < base + assert base - with_test >= 0.25 + + def test_secrets_unknown_category_defaults_neutral(self): + assert score_finding("secrets", "bogus", {}) == 0.50 + + def test_unknown_engine_defaults_neutral(self): + assert score_finding("bogus_engine", "whatever", {}) == 0.50 + + def test_score_always_in_valid_range(self): + """Even with extreme modifier stacking, score stays in [0, 1].""" + # Stack all negative modifiers + finding = { + "in_test_file": True, + "source": "library_api", + "downgraded": True, + "after": "throw", + } + c = score_finding("dead_code", "unused_exports", finding) + assert 0.0 <= c <= 1.0 + + def test_score_rounded_to_2_decimals(self): + c = score_finding("dead_code", "unreachable", {"after": "return"}) + # Should be a clean 2-decimal number + assert round(c, 2) == c + + def test_modifier_predicate_crash_is_swallowed(self): + """A crashing modifier predicate must not propagate (fail-safe).""" + # Pass a finding that would make attribute access fail — the modifier + # predicates use .get() so they're safe, but this test documents the + # contract: scoring never raises. + c = score_finding("dead_code", "unreachable", None) + # None finding — predicates use .get() which returns None, treated falsy. + # Should not raise; should return a valid score. + assert 0.0 <= c <= 1.0 + + +# ─── enrich_finding ──────────────────────────────────────────────────────── + + +class TestEnrichFinding: + """``enrich_finding`` adds confidence to a single finding dict.""" + + def test_adds_confidence_field(self): + finding = {"type": "pattern_match", "file": "a.py", "line": 1} + result = enrich_finding("secrets", finding) + assert "confidence" in result + assert 0.0 <= result["confidence"] <= 1.0 + + def test_idempotent(self): + finding = {"type": "pattern_match", "confidence": 0.42} + result = enrich_finding("secrets", finding) + assert result["confidence"] == 0.42 # preserved, not overwritten + + def test_secrets_uses_type_for_category(self): + """For secrets, `type` (detection method) is the scoring key, not `category`.""" + finding = { + "type": "pattern_match", # detection method + "category": "aws_key", # specific secret kind — too granular for scoring + } + result = enrich_finding("secrets", finding) + # pattern_match base is 0.90 — if we'd used "aws_key" we'd get 0.50 (unknown). + assert result["confidence"] >= 0.80 + + def test_dead_code_uses_category_then_type(self): + finding = {"type": "unreachable", "after": "return"} + result = enrich_finding("dead_code", finding) + # `type` is "unreachable" which is a known category. + assert result["confidence"] >= 0.85 + + def test_handles_non_dict(self): + assert enrich_finding("secrets", None) is None + assert enrich_finding("secrets", "not a dict") == "not a dict" + + def test_unknown_engine_still_stamps_confidence(self): + finding = {"type": "whatever"} + result = enrich_finding("bogus", finding) + assert result["confidence"] == 0.50 + + +# ─── enrich_findings (payload-level) ─────────────────────────────────────── + + +class TestEnrichFindings: + """``enrich_findings`` enriches all findings in a result payload.""" + + def test_dead_code_payload_shape(self): + """Dead-code style: results = {category: [finding, ...], ...}.""" + payload = { + "status": "ok", + "results": { + "unreachable": [ + {"file": "a.py", "line": 10, "after": "return"}, + {"file": "b.py", "line": 20, "after": "throw"}, + ], + "unused_vars": [ + {"file": "c.py", "variable": "x", "in_test_file": True}, + ], + }, + } + result = enrich_findings("dead_code", payload) + for item in result["results"]["unreachable"]: + assert "confidence" in item + assert 0.0 <= item["confidence"] <= 1.0 + # After return should score higher than after throw + scores = [i["confidence"] for i in result["results"]["unreachable"]] + assert scores[0] > scores[1] + # Test-file unused var should be discounted + test_var = result["results"]["unused_vars"][0] + assert test_var["confidence"] < 0.70 + + def test_secrets_payload_shape(self): + """Secrets style: findings = [finding, ...] (flat list).""" + payload = { + "status": "ok", + "findings": [ + {"type": "pattern_match", "file": "a.py", "category": "aws_key"}, + {"type": "entropy", "file": "b.py", "category": "high_entropy"}, + {"type": "pattern_match", "file": "c.py", "in_test_file": True}, + ], + } + result = enrich_findings("secrets", payload) + for item in result["findings"]: + assert "confidence" in item + # pattern_match should score higher than entropy + assert result["findings"][0]["confidence"] > result["findings"][1]["confidence"] + # Test-file pattern match should be heavily discounted + assert result["findings"][2]["confidence"] < result["findings"][0]["confidence"] + + def test_idempotent(self): + payload = { + "status": "ok", + "results": { + "unreachable": [{"file": "a.py", "confidence": 0.99}], + }, + "findings": [{"type": "pattern_match", "confidence": 0.88}], + } + result = enrich_findings("dead_code", payload) + # dead_code enriches results; secrets-style findings key is also present + # but for dead_code engine it should not touch findings list (it's empty + # of confidence-stamped items already). + assert result["results"]["unreachable"][0]["confidence"] == 0.99 + + def test_handles_non_dict_payload(self): + assert enrich_findings("secrets", None) is None + assert enrich_findings("secrets", []) == [] + + def test_handles_empty_payload(self): + result = enrich_findings("secrets", {"status": "ok"}) + assert result == {"status": "ok"} + + def test_preserves_other_fields(self): + payload = { + "status": "ok", + "stats": {"total": 1}, + "findings": [{"type": "pattern_match"}], + "risk": "high", + } + result = enrich_findings("secrets", payload) + assert result["stats"] == {"total": 1} + assert result["risk"] == "high" + assert "confidence" in result["findings"][0] + + +# ─── confidence_distribution ─────────────────────────────────────────────── + + +class TestConfidenceDistribution: + """``confidence_distribution`` buckets findings by confidence range.""" + + def test_basic_bucketing(self): + findings = [ + {"confidence": 0.95}, # high + {"confidence": 0.90}, # high + {"confidence": 0.70}, # medium + {"confidence": 0.50}, # medium + {"confidence": 0.30}, # low + {"confidence": 0.10}, # low + ] + buckets = confidence_distribution(findings) + assert buckets["high"] == 2 + assert buckets["medium"] == 2 + assert buckets["low"] == 2 + assert buckets["unscored"] == 0 + + def test_unscored_findings_counted(self): + findings = [ + {"confidence": 0.95}, + {"no_confidence_here": True}, + {"also_unscored": True}, + ] + buckets = confidence_distribution(findings) + assert buckets["high"] == 1 + assert buckets["unscored"] == 2 + + def test_boundary_high(self): + # 0.85 is the high/medium boundary — inclusive on high side. + assert confidence_distribution([{"confidence": 0.85}])["high"] == 1 + assert confidence_distribution([{"confidence": 0.84}])["medium"] == 1 + + def test_boundary_medium_low(self): + # 0.5 is the medium/low boundary — inclusive on medium side. + assert confidence_distribution([{"confidence": 0.50}])["medium"] == 1 + assert confidence_distribution([{"confidence": 0.49}])["low"] == 1 + + def test_empty_list(self): + buckets = confidence_distribution([]) + assert buckets == {"high": 0, "medium": 0, "low": 0, "unscored": 0} + + def test_non_dict_items_skipped(self): + findings = [{"confidence": 0.95}, "not a dict", 42, None] + buckets = confidence_distribution(findings) + assert buckets["high"] == 1 + assert buckets["unscored"] == 0 # non-dicts skipped, not counted as unscored + + +# ─── Integration: deadcode_engine ────────────────────────────────────────── + + +class TestDeadCodeEngineIntegration: + """``detect_dead_code`` must return findings with ``confidence`` fields.""" + + def _create_workspace(self, code, filename="app.js"): + ws = tempfile.mkdtemp() + with open(os.path.join(ws, filename), "w") as f: + f.write(code) + return ws + + def test_unreachable_findings_have_confidence(self): + code = """ +function process(data) { + return data; + console.log("unreachable"); +} +""" + ws = self._create_workspace(code) + try: + from deadcode_engine import detect_dead_code + result = detect_dead_code(ws) + assert result["status"] == "ok" + if "unreachable" in result["results"]: + for item in result["results"]["unreachable"]: + assert "confidence" in item + assert 0.0 <= item["confidence"] <= 1.0 + finally: + shutil.rmtree(ws, ignore_errors=True) + + def test_confidence_reflects_after_return_vs_throw(self): + code = """ +function a() { + return 1; + console.log("unreachable after return"); +} +function b() { + throw new Error("x"); + console.log("unreachable after throw"); +} +""" + ws = self._create_workspace(code) + try: + from deadcode_engine import detect_dead_code + result = detect_dead_code(ws) + items = result["results"].get("unreachable", []) + after_return = [i for i in items if i.get("after") == "return"] + after_throw = [i for i in items if i.get("after") == "throw"] + if after_return and after_throw: + # After return should score higher (more certain) + assert after_return[0]["confidence"] >= after_throw[0]["confidence"] + finally: + shutil.rmtree(ws, ignore_errors=True) + + def test_confidence_does_not_break_engine(self): + """Even if confidence module has issues, engine must still return.""" + code = "function test() { return true; }" + ws = self._create_workspace(code) + try: + from deadcode_engine import detect_dead_code + result = detect_dead_code(ws) + assert result["status"] == "ok" + # Must have results key (even if empty) + assert "results" in result + finally: + shutil.rmtree(ws, ignore_errors=True) + + +# ─── Integration: secrets_engine ─────────────────────────────────────────── + + +class TestSecretsEngineIntegration: + """``detect_secrets`` must return findings with ``confidence`` fields.""" + + def _create_workspace(self, code, filename="config.js"): + ws = tempfile.mkdtemp() + with open(os.path.join(ws, filename), "w") as f: + f.write(code) + return ws + + def test_pattern_match_finding_has_confidence(self): + ws = self._create_workspace( + 'const API_KEY = "sk-1234567890abcdef1234567890abcdef";' + ) + try: + from secrets_engine import detect_secrets + result = detect_secrets(ws) + assert result["status"] == "ok" + for finding in result["findings"]: + assert "confidence" in finding + assert 0.0 <= finding["confidence"] <= 1.0 + finally: + shutil.rmtree(ws, ignore_errors=True) + + def test_pattern_match_scores_high(self): + ws = self._create_workspace( + 'const API_KEY = "sk-1234567890abcdef1234567890abcdef";' + ) + try: + from secrets_engine import detect_secrets + result = detect_secrets(ws) + for finding in result["findings"]: + if finding.get("type") == "pattern_match": + # Pattern matches are high confidence + assert finding["confidence"] >= 0.80 + finally: + shutil.rmtree(ws, ignore_errors=True) + + def test_confidence_does_not_break_engine(self): + """Even if confidence module has issues, engine must still return.""" + ws = self._create_workspace('function hello() { return "hi"; }') + try: + from secrets_engine import detect_secrets + result = detect_secrets(ws) + assert result["status"] == "ok" + assert "findings" in result + finally: + shutil.rmtree(ws, ignore_errors=True) + + +# ─── Integration: formatters ─────────────────────────────────────────────── + + +class TestFormatterIntegration: + """``format_output`` must stamp ``schema_version`` on every output.""" + + def test_ai_format_has_schema_version(self): + from formatters import format_output + data = {"status": "ok", "stats": {}, "findings": []} + out = format_output(data, format_type="ai", command="test") + parsed = json.loads(out) + assert parsed["schema_version"] == SCHEMA_VERSION + + def test_json_format_has_schema_version(self): + from formatters import format_output + data = {"status": "ok", "stats": {}, "findings": []} + out = format_output(data, format_type="json", command="test") + parsed = json.loads(out) + assert parsed["schema_version"] == SCHEMA_VERSION + + def test_json_format_does_not_mutate_input(self): + """Raw JSON formatter stamps a copy, not the caller's dict.""" + from formatters import format_output + data = {"status": "ok", "findings": []} + format_output(data, format_type="json", command="test") + # Original dict must NOT have schema_version (stamping is on a copy). + assert "schema_version" not in data + + def test_ai_normalize_error_path_has_schema_version(self): + from formatters import _normalize_to_ai + result = _normalize_to_ai( + {"status": "error", "error": "boom"}, command="test" + ) + assert result["schema_version"] == SCHEMA_VERSION + assert result["status"] == "error" + + def test_ai_normalize_non_dict_has_schema_version(self): + from formatters import _normalize_to_ai + result = _normalize_to_ai("just a string", command="test") + assert result["schema_version"] == SCHEMA_VERSION + assert result["status"] == "ok" + + def test_ai_normalize_ok_path_has_schema_version(self): + from formatters import _normalize_to_ai + result = _normalize_to_ai( + {"status": "ok", "stats": {"x": 1}, "findings": []}, + command="test", + ) + assert result["schema_version"] == SCHEMA_VERSION + + def test_markdown_format_does_not_crash(self): + """Markdown format doesn't go through _normalize_to_ai; just verify + it doesn't crash. schema_version is a JSON-output concern.""" + from formatters import format_output + data = {"status": "ok", "stats": {"total_findings": 0}} + out = format_output(data, format_type="markdown", command="test") + assert isinstance(out, str) + assert len(out) > 0 + + +# ─── Integration: end-to-end CLI-style ───────────────────────────────────── + + +class TestEndToEndSchemaVersion: + """Verify schema_version propagates through the full formatter pipeline.""" + + def test_dead_code_ai_output_has_schema_version(self, workspace): + from formatters import format_output + from deadcode_engine import detect_dead_code + + # Empty workspace — engine should still return a valid payload. + result = detect_dead_code(workspace) + out = format_output(result, format_type="ai", command="dead-code") + parsed = json.loads(out) + assert parsed["schema_version"] == SCHEMA_VERSION + assert parsed["status"] == "ok" + assert parsed["command"] == "dead-code" + + def test_secrets_ai_output_has_schema_version(self, workspace): + from formatters import format_output + from secrets_engine import detect_secrets + + result = detect_secrets(workspace) + out = format_output(result, format_type="ai", command="secrets") + parsed = json.loads(out) + assert parsed["schema_version"] == SCHEMA_VERSION + assert parsed["status"] == "ok" + + def test_dead_code_json_output_has_schema_version(self, workspace): + from formatters import format_output + from deadcode_engine import detect_dead_code + + result = detect_dead_code(workspace) + out = format_output(result, format_type="json", command="dead-code") + parsed = json.loads(out) + assert parsed["schema_version"] == SCHEMA_VERSION + + def test_secrets_json_output_has_schema_version(self, workspace): + from formatters import format_output + from secrets_engine import detect_secrets + + result = detect_secrets(workspace) + out = format_output(result, format_type="json", command="secrets") + parsed = json.loads(out) + assert parsed["schema_version"] == SCHEMA_VERSION