tests/test_cve_trigger.py

"""Tests for scanners.cve_trigger_runner."""
from __future__ import annotations

import pytest

from scanners import cve_trigger_runner as ctr


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _vuln(cve_id: str, fix: str = "99.0") -> dict:
    return {"rule": cve_id, "tool": "pip-audit", "message": f"fix in {fix}"}


# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------

def test_no_trigger_files_returns_empty(tmp_path):
    """No Python files → no findings."""
    out, msg = ctr.cve_trigger(str(tmp_path), pip_audit_findings=[])
    assert out == []
    assert isinstance(msg, str)


def test_torch_load_no_weights_only(tmp_path):
    """torch.load() in source + matching CVE should fire PYSEC-2026-139."""
    (tmp_path / "model.py").write_text("import torch\nm = torch.load('model.pt')\n")
    findings, _ = ctr.cve_trigger(
        str(tmp_path),
        pip_audit_findings=[_vuln("PYSEC-2026-139")],
    )
    rules = {f["rule"] for f in findings}
    assert "PYSEC-2026-139" in rules


def test_torch_load_with_weights_only_is_clean(tmp_path):
    """torch.load(weights_only=True) should not fire."""
    (tmp_path / "safe.py").write_text(
        "import torch\nm = torch.load('model.pt', weights_only=True)\n"
    )
    out, _ = ctr.cve_trigger(str(tmp_path), pip_audit_findings=[])
    # Should find zero or fewer torch-load findings
    torch_findings = [f for f in out if "PYSEC-2026-139" in f.get("rule", "")]
    assert len(torch_findings) == 0


def test_cve_forwarded_from_pip_audit(tmp_path):
    """CVE in pip_audit_findings + matching source trigger produces a finding."""
    # PYSEC-2026-139 triggers on torch.load(
    (tmp_path / "load.py").write_text("import torch\nmodel = torch.load('weights.pt')\n")
    findings, msg = ctr.cve_trigger(
        str(tmp_path),
        pip_audit_findings=[_vuln("PYSEC-2026-139")],
    )
    assert any(f["rule"] == "PYSEC-2026-139" for f in findings), f"Expected PYSEC-2026-139 in {findings}"


def test_severity_is_string(tmp_path):
    """All findings must have string severity."""
    (tmp_path / "x.py").write_text("import pickle; pickle.loads(data)\n")
    out, _ = ctr.cve_trigger(str(tmp_path), pip_audit_findings=[])
    for f in out:
        assert isinstance(f.get("severity"), str)


def test_owasp_present(tmp_path):
    """Findings from pip_audit forwarding must include OWASP tag."""
    (tmp_path / "load.py").write_text("import torch\nm = torch.load('w.pt')\n")
    findings, _ = ctr.cve_trigger(
        str(tmp_path),
        pip_audit_findings=[_vuln("PYSEC-2026-139")],
    )
    for f in findings:
        assert "owasp" in f or "A0" in str(f.get("owasp", "A06"))