compounding-intelligence/tests/test_vulnerability_scanner.py

#!/usr/bin/env python3
"""Tests for scripts/vulnerability_scanner.py — 10 tests."""

import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch, MagicMock

sys.path.insert(0, os.path.join(os.path.dirname(__file__) or ".", ".."))
import importlib.util
spec = importlib.util.spec_from_file_location(
    "vulnerability_scanner",
    os.path.join(os.path.dirname(__file__) or ".", "..", "scripts", "vulnerability_scanner.py"))
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)

parse_requirements_file = mod.parse_requirements_file
query_osv = mod.query_osv
parse_osv_vuln = mod.parse_osv_vuln
filter_by_severity = mod.filter_by_severity
Vulnerability = mod.Vulnerability


# --- Test Data ---

SAMPLE_OSV_RESPONSE = [
    {
        "id": "GHSA-xxxx-xxxx-xxxx",
        "summary": " Arbitrary code execution in django",
        "severity": [{"type": "CVSS_V3", "score": {"baseScore": 9.8, "baseSeverity": "CRITICAL"}}],
        "affected": [{
            "ranges": [{
                "events": [
                    {"introduced": "0"},
                    {"fixed": "3.2.14"}
                ]
            }]
        }]
    },
    {
        "id": "PYSEC-2024-1234",
        "summary": " Denial of service in cryptography",
        "severity": [{"type": "CVSS_V3", "score": {"baseScore": 5.3, "baseSeverity": "MEDIUM"}}],
        "affected": [{
            "ranges": [{
                "events": [
                    {"introduced": "0"},
                    {"fixed": "42.0.0"}
                ]
            }]
        }]
    }
]


# --- Tests ---


def test_parse_requirements_simple():
    """Should parse a simple requirements file."""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
        f.write("django==4.2.0\n")
        f.write("requests>=2.28.0\n")
        f.write("click~=8.0\n")
        f.flush()
        pkgs = parse_requirements_file(f.name)
        os.unlink(f.name)

    assert "django" in pkgs
    assert pkgs["django"] == "==4.2.0"
    assert "requests" in pkgs
    assert pkgs["requests"] == ">=2.28.0"
    assert "click" in pkgs
    print("PASS: test_parse_requirements_simple")


def test_parse_requirements_extras_and_comments():
    """Should skip comments, blank lines, and handle package extras."""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
        f.write("# This is a comment\n")
        f.write("django[argon2]==4.2.0\n")
        f.write("\n")
        f.write("   requests   >=2.28.0   # inline comment\n")
        f.flush()
        pkgs = parse_requirements_file(f.name)
        os.unlink(f.name)

    assert "django" in pkgs
    assert pkgs["django"] == "==4.2.0"
    assert "requests" in pkgs
    # Version should capture the comparison
    assert ">=" in pkgs["requests"]
    print("PASS: test_parse_requirements_extras_and_comments")


def test_parse_requirements_include_recursive():
    """Should follow -r includes up to depth 3."""
    with tempfile.TemporaryDirectory() as tmpdir:
        # Main requirements.txt
        main = os.path.join(tmpdir, "requirements.txt")
        with open(main, 'w') as f:
            f.write("django==4.2.0\n")
            f.write("-r base.txt\n")

        # base.txt
        base = os.path.join(tmpdir, "base.txt")
        with open(base, 'w') as f:
            f.write("requests>=2.28.0\n")
            f.write("-r deep.txt\n")

        # deep.txt
        deep = os.path.join(tmpdir, "deep.txt")
        with open(deep, 'w') as f:
            f.write("click~=8.0\n")

        pkgs = parse_requirements_file(main)

        assert "django" in pkgs
        assert "requests" in pkgs
        assert "click" in pkgs
        print("PASS: test_parse_requirements_include_recursive")


def test_parse_requirements_skip_editable():
    """Should skip -e editable installs and other flags."""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
        f.write("-e git+https://github.com/user/repo.git@branch#egg=package\n")
        f.write("--index-url https://pypi.org/simple\n")
        f.write("django==4.2.0\n")
        f.flush()
        pkgs = parse_requirements_file(f.name)
        os.unlink(f.name)

    assert "django" in pkgs
    assert "package" not in pkgs  # should not pick up editable name
    print("PASS: test_parse_requirements_skip_editable")


def test_parse_requirements_nonexistent():
    """Should exit with error on missing file."""
    with patch('sys.exit') as mock_exit:
        pkgs = parse_requirements_file("/nonexistent/requirements.txt")
        mock_exit.assert_called_once_with(1)
    print("PASS: test_parse_requirements_nonexistent")


def test_filter_by_severity():
    """Should filter vulnerabilities by severity threshold."""
    vulns = [
        Vulnerability("pkg1", "==1.0", "V1", "critical", 9.8, "summary", "url", []),
        Vulnerability("pkg2", "==2.0", "V2", "high", 7.5, "summary", "url", []),
        Vulnerability("pkg3", "==3.0", "V3", "medium", 5.0, "summary", "url", []),
        Vulnerability("pkg4", "==4.0", "V4", "low", 2.0, "summary", "url", []),
    ]

    # min_severity: low includes all
    filtered = filter_by_severity(vulns, "low")
    assert len(filtered) == 4

    # min_severity: medium excludes low
    filtered = filter_by_severity(vulns, "medium")
    assert len(filtered) == 3
    assert all(v.severity in ("critical", "high", "medium") for v in filtered)

    # min_severity: high excludes medium + low
    filtered = filter_by_severity(vulns, "high")
    assert len(filtered) == 2

    # min_severity: critical only
    filtered = filter_by_severity(vulns, "critical")
    assert len(filtered) == 1

    print("PASS: test_filter_by_severity")


def test_parse_osv_vuln():
    """Should parse OSV API response correctly."""
    parsed = parse_osv_vuln(SAMPLE_OSV_RESPONSE, "django", "==4.2.0")

    assert len(parsed) == 2
    assert parsed[0].package == "django"
    assert parsed[0].vuln_id == "GHSA-xxxx-xxxx-xxxx"
    assert parsed[0].severity == "critical"
    assert parsed[0].cvss_score == 9.8
    assert parsed[1].severity == "medium"
    assert parsed[1].cvss_score == 5.3
    print("PASS: test_parse_osv_vuln")


def test_parse_osv_vuln_empty():
    """Should handle empty OSV response."""
    parsed = parse_osv_vuln([], "django", "==4.2.0")
    assert parsed == []
    print("PASS: test_parse_osv_vuln_empty")


def test_query_osv_network_success():
    """Should successfully query OSV API for a real known vulnerable package."""
    # Query for an old django version that likely has known CVEs
    # This test actually hits the network — tagged as integration
    vulns = query_osv("django", "==3.2.0")
    # We don't assert specific results since vulns change over time
    # But we assert the function returns a list and doesn't error
    assert isinstance(vulns, list)
    print("PASS: test_query_osv_network_success")


def test_query_osv_404_no_vulns():
    """OSV returns empty list for packages with no vulns (404-like)."""
    # Mock a 404 response from OSV API
    with patch('urllib.request.urlopen') as mock_urlopen:
        mock_response = MagicMock()
        mock_response.read.return_value = b'{"vulns": []}'
        mock_response.__enter__ = lambda self: self
        mock_response.__exit__ = lambda self, *args: None
        mock_urlopen.return_value = mock_response

        result = query_osv("nonexistent-package-xyz123", "==1.0.0")
        assert result == []
    print("PASS: test_query_osv_404_no_vulns")


if __name__ == '__main__':
    # Run all tests
    test_parse_requirements_simple()
    test_parse_requirements_extras_and_comments()
    test_parse_requirements_include_recursive()
    test_parse_requirements_skip_editable()
    test_parse_requirements_nonexistent()
    test_filter_by_severity()
    test_parse_osv_vuln()
    test_parse_osv_vuln_empty()
    test_query_osv_network_success()
    test_query_osv_404_no_vulns()
    print("\nAll tests passed.")