From a6f917561a159e429c773f19c2beda8c47e9696f Mon Sep 17 00:00:00 2001 From: karthickmk2001 Date: Thu, 12 Nov 2026 04:24:32 +0000 Subject: [PATCH] feat(runtime): import coverage files as runtime evidence (#54) secfoo runtime import-coverage turns a coverage.py JSON report (>= 7.5, per-function data) or a V8 coverage folder into #53 events: watched symbols whose function ran become hits, the session carries the app's own function coverage, and the session id is content-derived so re-imports never double-count. Explains how to include dependencies when a coverage.py report omits site-packages. Committed fixtures for both formats are tested end to end (import -> ingest -> evidence rows). Closes #54 Co-Authored-By: Claude Opus 5.5 --- docs/REFERENCE.md | 25 ++ src/secfoo/cli.py | 37 +++ src/secfoo/runtime/coverage_import.py | 213 +++++++++++++ tests/fixtures/coverage/python-coverage.json | 280 ++++++++++++++++++ .../v8/coverage-4242-1794450000-0.json | 117 ++++++++ tests/test_runtime/test_coverage_import.py | 136 +++++++++ 6 files changed, 808 insertions(+) create mode 100644 src/secfoo/runtime/coverage_import.py create mode 100644 tests/fixtures/coverage/python-coverage.json create mode 100644 tests/fixtures/coverage/v8/coverage-4242-1794450000-0.json create mode 100644 tests/test_runtime/test_coverage_import.py diff --git a/docs/REFERENCE.md b/docs/REFERENCE.md index 10114a4..f126f50 100644 --- a/docs/REFERENCE.md +++ b/docs/REFERENCE.md @@ -473,6 +473,31 @@ runtime-confirmed findings, but only once there is runtime data. ![SCA vulnerabilities list with the Runtime column and a highlighted conflict](assets/sca-runtime.jpg) +### Coverage files as runtime evidence + +The easiest evidence needs no sensor: coverage from your test suite or a +staging run. + +```bash +coverage run --source=app,yaml,jinja2 -m pytest && coverage json # include the dependencies you watch +secfoo runtime import-coverage coverage.json --watchlist watchlist.json -o events.ndjson +NODE_V8_COVERAGE=./v8-cov node app.js +secfoo runtime import-coverage ./v8-cov --watchlist watchlist.json -o events.ndjson +secfoo runtime ingest events.ndjson --project payments +``` + +- **Python:** a coverage.py JSON report from coverage 7.5 or later, which has + per-function data. coverage.py **omits site-packages by default**, so + measure the watched packages with `--source` or `[run] include`; the command + says so when no dependency files are in the report. +- **Node:** a V8 coverage folder, or one of its files. Its function ranges + already include `node_modules`. + +A watched symbol counts as **Executed** when its function ran. Coverage has no +routes or stack traces. The session line gets the app's own function +coverage. The session id comes from the file contents, so re-importing the +same coverage never double-counts. + ### Python probe The probe runs inside a Python app and records which watched vulnerable diff --git a/src/secfoo/cli.py b/src/secfoo/cli.py index e6cfff4..c75187a 100644 --- a/src/secfoo/cli.py +++ b/src/secfoo/cli.py @@ -1821,6 +1821,43 @@ def runtime_ingest( raise typer.Exit(code=1) +@runtime_app.command(name="import-coverage") +def runtime_import_coverage( + coverage: Annotated[Path, typer.Argument( + exists=True, help="coverage.py JSON report, or a V8 coverage folder/file (NODE_V8_COVERAGE).", + )], + watchlist: Annotated[Path, typer.Option( + "--watchlist", exists=True, dir_okay=False, help="From `secfoo runtime watchlist`.", + )], + output: Annotated[Path | None, typer.Option("--output", "-o", help="Write events here instead of stdout.")] = None, +) -> None: + """Turn coverage from tests or staging into runtime evidence (no sensor needed). + + Then: secfoo runtime ingest events.ndjson --project + """ + from secfoo.runtime.coverage_import import ( + CoverageImportError, + load_coverage, + to_events, + ) + + try: + cov = load_coverage(coverage) + events, notes = to_events(cov, json.loads(watchlist.read_text(encoding="utf-8"))) + except (CoverageImportError, OSError, json.JSONDecodeError) as exc: + err_console.print(f"[red]Error:[/] {exc}") + raise typer.Exit(code=1) from None + text = "".join(json.dumps(e) + "\n" for e in events) + if output is None: + sys.stdout.write(text) + else: + output.write_text(text, encoding="utf-8") + for note in notes: + err_console.print(f"[yellow]Note:[/] {note}") + hits = sum(1 for e in events if e["type"] == "hit") + err_console.print(f"{cov.language} coverage: {hits} watched symbol(s) executed" + (f". Wrote {output}." if output else ".")) + + @runtime_app.command(name="watchlist") def runtime_watchlist( project: Annotated[str, typer.Option("--project", help="Project id, name, or path/URL it was scanned as.")], diff --git a/src/secfoo/runtime/coverage_import.py b/src/secfoo/runtime/coverage_import.py new file mode 100644 index 0000000..82b3b54 --- /dev/null +++ b/src/secfoo/runtime/coverage_import.py @@ -0,0 +1,213 @@ +"""Import coverage files as runtime evidence (#54). + +Many projects already produce coverage from their test suites or staging +runs. This turns it into #53 events, with no sensor: + + secfoo runtime import-coverage coverage.json --watchlist watchlist.json -o events.ndjson + secfoo runtime import-coverage ./v8-coverage/ --watchlist watchlist.json -o events.ndjson + +- **Python:** a coverage.py JSON report (`coverage json`, coverage >= 7.5, + which records per-function executed lines). Dependencies must have been + measured (`--source` / `[run] include`), since site-packages are omitted + by default. +- **Node:** a V8 coverage directory (`NODE_V8_COVERAGE=dir node app.js`) or + one of its files. Function-level ranges already include node_modules. + +A watched symbol is a hit when its function ran. Coverage has no stack +traces or routes, so `entry_frame`/`stack_sample` stay empty and `route` is +"unknown" (shown as Executed). The session id is derived from the file +contents, so importing the same coverage twice never double-counts. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath +from urllib.parse import unquote, urlsplit + +_DEP_DIRS = ("site-packages", "dist-packages") + + +class CoverageImportError(ValueError): + pass + + +@dataclass +class _Coverage: + language: str + started_at: str + digest: str + # symbol-candidate -> call count (0 = present but never ran) + functions: dict[str, int] = field(default_factory=dict) + app_total: int = 0 + app_executed: int = 0 + dependency_files: int = 0 + + +def _iso(ts) -> str: + if isinstance(ts, (int, float)): + return datetime.fromtimestamp(ts, timezone.utc).isoformat() + if isinstance(ts, str) and ts: + try: + parsed = datetime.fromisoformat(ts) + except ValueError: + parsed = None + if parsed is not None: + return (parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)).isoformat() + return datetime.now(timezone.utc).isoformat() + + +# ---------- Python: coverage.py JSON ---------- + + +def _python_module(path: str) -> str | None: + """'.../site-packages/yaml/constructor.py' -> 'yaml.constructor'.""" + parts = PurePosixPath(path.replace("\\", "/")).parts + for marker in _DEP_DIRS: + if marker in parts: + rel = list(parts[parts.index(marker) + 1:]) + break + else: + return None + if not rel or not rel[-1].endswith(".py"): + return None + rel[-1] = rel[-1][:-3] + if rel[-1] == "__init__": + rel = rel[:-1] + return ".".join(rel) or None + + +def _load_python(report: dict, digest: str) -> _Coverage: + cov = _Coverage("python", _iso(report.get("meta", {}).get("timestamp")), digest) + saw_functions = False + for path, entry in (report.get("files") or {}).items(): + functions = entry.get("functions") + if functions is None: + continue + saw_functions = True + module = _python_module(path) + for qualname, info in functions.items(): + if not qualname: # module-level code + continue + ran = 1 if info.get("executed_lines") else 0 + if module is None: + cov.app_total += 1 + cov.app_executed += ran + else: + cov.functions[f"{module}.{qualname}"] = ran + if module is not None: + cov.dependency_files += 1 + if not saw_functions: + raise CoverageImportError( + "this coverage.py JSON has no per-function data; regenerate it with coverage >= 7.5 (`coverage json`)" + ) + if cov.dependency_files == 0: + raise CoverageImportError( + "the report contains no dependency files, so nothing can be matched to a vulnerable package. " + "coverage.py omits site-packages by default: collect with the packages included, e.g. " + "`coverage run --source=app,yaml,jinja2 ...` or `[run] include = */site-packages/yaml/*` " + "in .coveragerc, then `coverage json` again" + ) + return cov + + +# ---------- Node: V8 coverage ---------- + +_NODE_MODULES_RE = re.compile(r"node_modules/((?:@[^/]+/)?[^/]+)/") + + +def _node_package(url: str) -> str | None: + path = unquote(urlsplit(url).path) if url.startswith("file:") else url + matches = _NODE_MODULES_RE.findall(path.replace("\\", "/")) + return matches[-1] if matches else None # innermost node_modules: nested copies are their own package + + +def _load_v8(documents: list[dict], digest: str) -> _Coverage: + timestamps = [d.get("timestamp") for d in documents if d.get("timestamp")] + cov = _Coverage("node", _iso(min(timestamps) if timestamps else None), digest) + for document in documents: + for script in document.get("result") or []: + url = script.get("url") or "" + if not url.startswith(("file:", "/")) and ":\\" not in url: + continue # node:internal, eval, ... + package = _node_package(url) + for function in script.get("functions") or []: + name = function.get("functionName") or "" + ranges = function.get("ranges") or [] + count = int(ranges[0].get("count", 0)) if ranges else 0 + if not name: + continue + if package is None: + cov.app_total += 1 + cov.app_executed += 1 if count else 0 + else: + key = f"{package}#{name}" + cov.functions[key] = max(cov.functions.get(key, 0), count) + if package is not None: + cov.dependency_files += 1 + return cov + + +def _npm_candidates(symbol: str) -> list[str]: + """'lodash#template' -> ['lodash#template']; 'pkg#a.b' also matches 'pkg#b' + (V8 records the function's own name, not its export path).""" + package, _, export = symbol.partition("#") + return [symbol] + ([f"{package}#{export.rsplit('.', 1)[-1]}"] if "." in export else []) + + +# ---------- shared ---------- + + +def load_coverage(path: Path) -> _Coverage: + if path.is_dir(): + files = sorted(path.glob("coverage-*.json")) or sorted(path.glob("*.json")) + if not files: + raise CoverageImportError(f"{path}: no V8 coverage files (coverage-*.json) in this folder") + raw = b"".join(f.read_bytes() for f in files) + return _load_v8([json.loads(f.read_text(encoding="utf-8")) for f in files], hashlib.sha256(raw).hexdigest()) + raw = path.read_bytes() + try: + document = json.loads(raw) + except json.JSONDecodeError as exc: + raise CoverageImportError(f"{path}: not JSON ({exc.msg})") from None + digest = hashlib.sha256(raw).hexdigest() + if isinstance(document, dict) and "files" in document and "meta" in document: + return _load_python(document, digest) + if isinstance(document, dict) and "result" in document: + return _load_v8([document], digest) + raise CoverageImportError(f"{path}: not a coverage.py JSON report or a V8 coverage file") + + +def to_events(cov: _Coverage, watchlist: dict) -> tuple[list[dict], list[str]]: + """#53 lines for the watched symbols this coverage covers, plus notes.""" + session_id = "coverage-" + cov.digest[:32] + session = {"type": "session", "id": session_id, "language": cov.language, "source": "coverage", + "started_at": cov.started_at} + if cov.app_total: + session["app_functions_total"] = cov.app_total + session["app_functions_executed"] = cov.app_executed + events: list[dict] = [session] + notes: list[str] = [] + ecosystem = {"python": "PyPI", "node": "npm"}[cov.language] + entries = [e for e in watchlist.get("entries", []) if e.get("ecosystem") == ecosystem] + if not entries: + notes.append(f"the watchlist has no {ecosystem} entries, so nothing in this coverage can match") + unmatched = 0 + for entry in entries: + for symbol in entry.get("symbols") or []: + candidates = _npm_candidates(symbol) if cov.language == "node" else [symbol] + count = max((cov.functions.get(c, -1) for c in candidates), default=-1) + if count < 0: + unmatched += 1 # the function isn't in the report: its package wasn't measured + elif count > 0: + events.append({"type": "hit", "session": session_id, + "finding_fingerprint": entry["finding_fingerprint"], "symbol": symbol, + "route": "unknown", "count": count, "first_seen": cov.started_at, + "last_seen": cov.started_at}) + if unmatched: + notes.append(f"{unmatched} watched symbol(s) weren't in the coverage report (their package wasn't measured)") + return events, notes diff --git a/tests/fixtures/coverage/python-coverage.json b/tests/fixtures/coverage/python-coverage.json new file mode 100644 index 0000000..9e2ef3b --- /dev/null +++ b/tests/fixtures/coverage/python-coverage.json @@ -0,0 +1,280 @@ +{ + "files": { + "/srv/app/.venv/lib/python3.12/site-packages/vulnpkg/__init__.py": { + "classes": { + "": { + "excluded_lines": [], + "executed_lines": [ + 1, + 2, + 5, + 9, + 10 + ], + "missing_lines": [ + 6 + ], + "start_line": 1, + "summary": { + "covered_lines": 5, + "excluded_lines": 0, + "missing_lines": 1, + "num_statements": 6, + "percent_covered": 83.33333333333333, + "percent_covered_display": "83", + "percent_statements_covered": 83.33333333333333, + "percent_statements_covered_display": "83" + } + }, + "Loader": { + "excluded_lines": [], + "executed_lines": [ + 11 + ], + "missing_lines": [], + "start_line": 9, + "summary": { + "covered_lines": 1, + "excluded_lines": 0, + "missing_lines": 0, + "num_statements": 1, + "percent_covered": 100.0, + "percent_covered_display": "100", + "percent_statements_covered": 100.0, + "percent_statements_covered_display": "100" + } + } + }, + "excluded_lines": [], + "executed_lines": [ + 1, + 2, + 5, + 9, + 10, + 11 + ], + "functions": { + "": { + "excluded_lines": [], + "executed_lines": [ + 1, + 5, + 9, + 10 + ], + "missing_lines": [], + "start_line": 1, + "summary": { + "covered_lines": 4, + "excluded_lines": 0, + "missing_lines": 0, + "num_statements": 4, + "percent_covered": 100.0, + "percent_covered_display": "100", + "percent_statements_covered": 100.0, + "percent_statements_covered_display": "100" + } + }, + "Loader.construct": { + "excluded_lines": [], + "executed_lines": [ + 11 + ], + "missing_lines": [], + "start_line": 10, + "summary": { + "covered_lines": 1, + "excluded_lines": 0, + "missing_lines": 0, + "num_statements": 1, + "percent_covered": 100.0, + "percent_covered_display": "100", + "percent_statements_covered": 100.0, + "percent_statements_covered_display": "100" + } + }, + "load": { + "excluded_lines": [], + "executed_lines": [ + 2 + ], + "missing_lines": [], + "start_line": 1, + "summary": { + "covered_lines": 1, + "excluded_lines": 0, + "missing_lines": 0, + "num_statements": 1, + "percent_covered": 100.0, + "percent_covered_display": "100", + "percent_statements_covered": 100.0, + "percent_statements_covered_display": "100" + } + }, + "safe_load": { + "excluded_lines": [], + "executed_lines": [], + "missing_lines": [ + 6 + ], + "start_line": 5, + "summary": { + "covered_lines": 0, + "excluded_lines": 0, + "missing_lines": 1, + "num_statements": 1, + "percent_covered": 0.0, + "percent_covered_display": "0", + "percent_statements_covered": 0.0, + "percent_statements_covered_display": "0" + } + } + }, + "missing_lines": [ + 6 + ], + "summary": { + "covered_lines": 6, + "excluded_lines": 0, + "missing_lines": 1, + "num_statements": 7, + "percent_covered": 85.71428571428571, + "percent_covered_display": "86", + "percent_statements_covered": 85.71428571428571, + "percent_statements_covered_display": "86" + } + }, + "/srv/app/app/main.py": { + "classes": { + "": { + "excluded_lines": [], + "executed_lines": [ + 1, + 4, + 5, + 8, + 12, + 13, + 14 + ], + "missing_lines": [ + 9 + ], + "start_line": 1, + "summary": { + "covered_lines": 7, + "excluded_lines": 0, + "missing_lines": 1, + "num_statements": 8, + "percent_covered": 87.5, + "percent_covered_display": "88", + "percent_statements_covered": 87.5, + "percent_statements_covered_display": "88" + } + } + }, + "excluded_lines": [], + "executed_lines": [ + 1, + 4, + 5, + 8, + 12, + 13, + 14 + ], + "functions": { + "": { + "excluded_lines": [], + "executed_lines": [ + 1, + 4, + 8, + 12, + 13, + 14 + ], + "missing_lines": [], + "start_line": 1, + "summary": { + "covered_lines": 6, + "excluded_lines": 0, + "missing_lines": 0, + "num_statements": 6, + "percent_covered": 100.0, + "percent_covered_display": "100", + "percent_statements_covered": 100.0, + "percent_statements_covered_display": "100" + } + }, + "handle": { + "excluded_lines": [], + "executed_lines": [ + 5 + ], + "missing_lines": [], + "start_line": 4, + "summary": { + "covered_lines": 1, + "excluded_lines": 0, + "missing_lines": 0, + "num_statements": 1, + "percent_covered": 100.0, + "percent_covered_display": "100", + "percent_statements_covered": 100.0, + "percent_statements_covered_display": "100" + } + }, + "unused": { + "excluded_lines": [], + "executed_lines": [], + "missing_lines": [ + 9 + ], + "start_line": 8, + "summary": { + "covered_lines": 0, + "excluded_lines": 0, + "missing_lines": 1, + "num_statements": 1, + "percent_covered": 0.0, + "percent_covered_display": "0", + "percent_statements_covered": 0.0, + "percent_statements_covered_display": "0" + } + } + }, + "missing_lines": [ + 9 + ], + "summary": { + "covered_lines": 7, + "excluded_lines": 0, + "missing_lines": 1, + "num_statements": 8, + "percent_covered": 87.5, + "percent_covered_display": "88", + "percent_statements_covered": 87.5, + "percent_statements_covered_display": "88" + } + } + }, + "meta": { + "branch_coverage": false, + "format": 3, + "show_contexts": false, + "timestamp": "2026-11-12T04:21:57.154719", + "version": "7.17.2" + }, + "totals": { + "covered_lines": 13, + "excluded_lines": 0, + "missing_lines": 2, + "num_statements": 15, + "percent_covered": 86.66666666666667, + "percent_covered_display": "87", + "percent_statements_covered": 86.66666666666667, + "percent_statements_covered_display": "87" + } +} \ No newline at end of file diff --git a/tests/fixtures/coverage/v8/coverage-4242-1794450000-0.json b/tests/fixtures/coverage/v8/coverage-4242-1794450000-0.json new file mode 100644 index 0000000..1b95b90 --- /dev/null +++ b/tests/fixtures/coverage/v8/coverage-4242-1794450000-0.json @@ -0,0 +1,117 @@ +{ + "result": [ + { + "scriptId": "1", + "url": "file:///srv/app/node_modules/lodash/lodash.js", + "functions": [ + { + "functionName": "", + "ranges": [ + { + "startOffset": 0, + "endOffset": 900, + "count": 1 + } + ], + "isBlockCoverage": true + }, + { + "functionName": "template", + "ranges": [ + { + "startOffset": 10, + "endOffset": 200, + "count": 4 + } + ], + "isBlockCoverage": true + }, + { + "functionName": "escape", + "ranges": [ + { + "startOffset": 210, + "endOffset": 300, + "count": 0 + } + ], + "isBlockCoverage": false + } + ] + }, + { + "scriptId": "2", + "url": "file:///srv/app/node_modules/@babel/core/lib/index.js", + "functions": [ + { + "functionName": "transform", + "ranges": [ + { + "startOffset": 0, + "endOffset": 80, + "count": 0 + } + ], + "isBlockCoverage": false + } + ] + }, + { + "scriptId": "3", + "url": "file:///srv/app/src/server.js", + "functions": [ + { + "functionName": "", + "ranges": [ + { + "startOffset": 0, + "endOffset": 500, + "count": 1 + } + ], + "isBlockCoverage": true + }, + { + "functionName": "handle", + "ranges": [ + { + "startOffset": 20, + "endOffset": 120, + "count": 4 + } + ], + "isBlockCoverage": true + }, + { + "functionName": "unused", + "ranges": [ + { + "startOffset": 130, + "endOffset": 160, + "count": 0 + } + ], + "isBlockCoverage": false + } + ] + }, + { + "scriptId": "4", + "url": "node:internal/main/run_main_module", + "functions": [ + { + "functionName": "", + "ranges": [ + { + "startOffset": 0, + "endOffset": 10, + "count": 1 + } + ], + "isBlockCoverage": false + } + ] + } + ], + "timestamp": 1794450000.123 +} \ No newline at end of file diff --git a/tests/test_runtime/test_coverage_import.py b/tests/test_runtime/test_coverage_import.py new file mode 100644 index 0000000..d13ff13 --- /dev/null +++ b/tests/test_runtime/test_coverage_import.py @@ -0,0 +1,136 @@ +"""#54: coverage files as runtime evidence -- end to end with committed fixtures.""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from secfoo.cli import app +from secfoo.runtime.coverage_import import CoverageImportError, load_coverage, to_events +from secfoo.storage.repository import RunRepository + +FIXTURES = Path(__file__).resolve().parents[1] / "fixtures" / "coverage" +PY_COVERAGE = FIXTURES / "python-coverage.json" +V8_DIR = FIXTURES / "v8" +runner = CliRunner(env={"COLUMNS": "250"}) + + +def _watchlist(path: Path) -> Path: + path.write_text(json.dumps({"version": 1, "generated_at": "x", "entries": [ + {"finding_fingerprint": "fp-py", "advisory": None, "ecosystem": "PyPI", "package": "vulnpkg", "version": "1", + "granularity": "symbol", "symbols": ["vulnpkg.load", "vulnpkg.safe_load", "vulnpkg.Loader.construct", + "vulnpkg.missing"], "source": "osv"}, + {"finding_fingerprint": "fp-npm", "advisory": None, "ecosystem": "npm", "package": "lodash", "version": "4", + "granularity": "symbol", "symbols": ["lodash#template", "lodash#escape"], "source": "osv"}, + {"finding_fingerprint": "fp-babel", "advisory": None, "ecosystem": "npm", "package": "@babel/core", + "version": "7", "granularity": "symbol", "symbols": ["@babel/core#transform"], "source": "osv"}, + ]}), encoding="utf-8") + return path + + +def _hits(events): + return {e["symbol"]: e for e in events if e["type"] == "hit"} + + +# ---------- Python ---------- + + +def test_python_coverage_hits_and_app_coverage(tmp_path): + events, notes = to_events(load_coverage(PY_COVERAGE), json.loads(_watchlist(tmp_path / "w.json").read_text())) + session = events[0] + assert session["source"] == "coverage" and session["language"] == "python" + assert (session["app_functions_total"], session["app_functions_executed"]) == (2, 1) # handle ran, unused didn't + hits = _hits(events) + assert set(hits) == {"vulnpkg.load", "vulnpkg.Loader.construct"} # safe_load never ran + assert all("stack_sample" not in h and "entry_frame" not in h for h in hits.values()) + assert any("weren't in the coverage report" in n for n in notes) # vulnpkg.missing + + +def test_python_coverage_without_dependencies_explains_how_to_collect_them(tmp_path): + report = json.loads(PY_COVERAGE.read_text()) + report["files"] = {k: v for k, v in report["files"].items() if "site-packages" not in k} + path = tmp_path / "app-only.json" + path.write_text(json.dumps(report)) + with pytest.raises(CoverageImportError, match=r"omits site-packages by default.*--source"): + load_coverage(path) + + +def test_old_coverage_without_function_data_is_explained(tmp_path): + report = json.loads(PY_COVERAGE.read_text()) + for entry in report["files"].values(): + entry.pop("functions") + path = tmp_path / "old.json" + path.write_text(json.dumps(report)) + with pytest.raises(CoverageImportError, match="coverage >= 7.5"): + load_coverage(path) + + +# ---------- Node / V8 ---------- + + +def test_v8_coverage_directory(tmp_path): + events, _ = to_events(load_coverage(V8_DIR), json.loads(_watchlist(tmp_path / "w.json").read_text())) + session = events[0] + assert session["language"] == "node" + assert (session["app_functions_total"], session["app_functions_executed"]) == (2, 1) + hits = _hits(events) + assert set(hits) == {"lodash#template"} # escape and @babel/core#transform had count 0 + assert hits["lodash#template"]["count"] == 4 + + +def test_a_single_v8_file_works_too(tmp_path): + [file] = V8_DIR.glob("*.json") + events, _ = to_events(load_coverage(file), json.loads(_watchlist(tmp_path / "w.json").read_text())) + assert set(_hits(events)) == {"lodash#template"} + + +def test_unrecognised_files_are_rejected(tmp_path): + bad = tmp_path / "x.json" + bad.write_text('{"hello": 1}') + with pytest.raises(CoverageImportError, match="not a coverage.py JSON report or a V8"): + load_coverage(bad) + + +# ---------- end to end: import -> ingest -> evidence rows ---------- + + +@pytest.mark.parametrize("source, fingerprint, symbol, count", [ + (PY_COVERAGE, "fp-py", "vulnpkg.load", 1), + (V8_DIR, "fp-npm", "lodash#template", 4), +]) +def test_end_to_end_import_ingest_evidence(tmp_path, monkeypatch, source, fingerprint, symbol, count): + db = tmp_path / "db.sqlite" + monkeypatch.setattr("secfoo.cli.RunRepository", lambda: RunRepository(db_path=db)) + with RunRepository(db_path=db) as repo: + pid = repo.upsert_project("https://github.com/acme/app", "app", "github") + run_id = repo.get_run(repo.create_run(project_id=pid, skill_id="sca-reachability", skill_name="SCA", + agent_id="claude", confluence_urls=[])).id + for fp in ("fp-py", "fp-npm", "fp-babel"): + repo.upsert_sca_finding(project_id=pid, fingerprint=fp, run_id=run_id, current_ref="D1", package=fp, + severity="High") + out = tmp_path / "events.ndjson" + imported = runner.invoke(app, ["runtime", "import-coverage", str(source), "--watchlist", + str(_watchlist(tmp_path / "w.json")), "-o", str(out)]) + assert imported.exit_code == 0, imported.output + ingested = runner.invoke(app, ["runtime", "ingest", str(out), "--project", "app"]) + assert ingested.exit_code == 0, ingested.output + again = runner.invoke(app, ["runtime", "ingest", str(out), "--project", "app"]) # idempotent + assert again.exit_code == 0 + with RunRepository(db_path=db) as repo: + rows = [tuple(r) for r in repo.connection.execute( + "SELECT e.finding_fingerprint, e.symbol, e.count, s.source FROM runtime_evidence e " + "JOIN runtime_sessions s ON s.id = e.session_id WHERE e.finding_fingerprint = ?", (fingerprint,))] + assert (fingerprint, symbol, count, "coverage") in rows + assert len([r for r in rows if r[1] == symbol]) == 1 + + +def test_cli_reports_missing_dependencies_clearly(tmp_path): + report = json.loads(PY_COVERAGE.read_text()) + report["files"] = {k: v for k, v in report["files"].items() if "site-packages" not in k} + path = tmp_path / "app-only.json" + path.write_text(json.dumps(report)) + result = runner.invoke(app, ["runtime", "import-coverage", str(path), "--watchlist", + str(_watchlist(tmp_path / "w.json"))]) + assert result.exit_code == 1 and "site-packages" in " ".join(result.output.split())