from __future__ import annotations from datetime import UTC, datetime from pathlib import Path from unittest.mock import MagicMock import pytest from tests.helpers import make_alert, make_dependency, make_graph from vuln_scan.core.dependency_graph import DependencyGraph from vuln_scan.core.models import ( Dependency, Ecosystem, Finding, FindingStatus, PolicyDecision, ScanConfig, ScanResult, Severity, ) from vuln_scan.core.policy import BlockingPolicy from vuln_scan.core.registry import EcosystemRegistry from vuln_scan.core.scanner import VulnerabilityScanner from vuln_scan.ecosystems.base import EcosystemHandler from vuln_scan.github.publisher import StatusPublisher from vuln_scan.reporting.base import Reporter # -------------------------------------------------------- # Helpers # -------------------------------------------------------- def _cfg( skip_phrase: str = "SKIP-SECURITY-CHECK", pr_comment_text: str = "", dependabot_url: str = "https://example.com", ) -> ScanConfig: return ScanConfig( owner="test-owner", repo="test-repo", base_path=Path("/tmp/test"), skip_phrase=skip_phrase, pr_comment_text=pr_comment_text, dependabot_url=dependabot_url, ) _SENTINEL = object() def _mock_finding( *, dep: Dependency | None | object = _SENTINEL, alert=None, status: FindingStatus = FindingStatus.VULNERABLE, is_blocking: bool = False, ) -> Finding: return Finding( dependency=make_dependency(version="2.25.0") if dep is _SENTINEL else dep, alert=alert or make_alert(), decision=MagicMock(status=status, is_blocking=is_blocking), ) def _mock_handler(*, is_vulnerable: bool = True) -> MagicMock: handler = MagicMock(spec=EcosystemHandler) handler.is_vulnerable.return_value = (is_vulnerable, None) return handler # -------------------------------------------------------- # Fixtures # -------------------------------------------------------- @pytest.fixture def scanner() -> VulnerabilityScanner: s = VulnerabilityScanner( workspace_scanner=MagicMock(), default_branch_scanner=MagicMock(), alerts_client=MagicMock(), advisory_client=MagicMock(), registry=MagicMock(spec=EcosystemRegistry), policy=MagicMock(spec=BlockingPolicy), reporter=MagicMock(spec=Reporter), publisher=MagicMock(spec=StatusPublisher), ) s._workspace_scanner.scan.return_value = DependencyGraph() s._default_branch_scanner.scan.return_value = DependencyGraph() s._alerts_client.get_open_alerts.return_value = [] s._advisory_client.get_advisories_batch.return_value = {} s._reporter.render.return_value = "" return s # -------------------------------------------------------- # Skip logic # -------------------------------------------------------- class TestSkip: @pytest.mark.parametrize( "skip_phrase, pr_comment_text, expect_skip", [ pytest.param( "SKIP-SECURITY-CHECK", "This PR has SKIP-SECURITY-CHECK in it", True, id="skip-phrase-found", ), pytest.param( "SKIP-SECURITY-CHECK", "Normal comment", False, id="skip-phrase-not-found", ), pytest.param( "", "Some comment", False, id="no-skip-phrase-configured", ), ], ) def test_scan_skip_logic( self, scanner: VulnerabilityScanner, skip_phrase: str, pr_comment_text: str, expect_skip: bool, ) -> None: cfg = _cfg(skip_phrase=skip_phrase, pr_comment_text=pr_comment_text) result = scanner.scan(cfg) assert isinstance(result, ScanResult) if expect_skip: assert result.findings == [] scanner._publisher.set_status.assert_called_with("success") scanner._workspace_scanner.scan.assert_not_called() else: scanner._workspace_scanner.scan.assert_called_once() # -------------------------------------------------------- # Graph scanning & deltas # -------------------------------------------------------- class TestGraphs: def test_scan_graphs_calls_both_scanners(self, scanner: VulnerabilityScanner) -> None: scanner._workspace_scanner.scan.return_value = make_graph( ("requirements.txt", make_dependency(version="2.25.0")) ) scanner._default_branch_scanner.scan.return_value = make_graph( ("requirements.txt", make_dependency(version="2.20.0")) ) w, d = scanner._scan_graphs() assert w.dependency_count() == 1 assert d.dependency_count() == 1 scanner._workspace_scanner.scan.assert_called_once() scanner._default_branch_scanner.scan.assert_called_once() @pytest.mark.parametrize( "workspace_deps, default_deps, check_field, expected_pkg", [ pytest.param( [ ("requirements.txt", make_dependency(name="requests", version="2.25.0")), ("requirements.txt", make_dependency(name="flask", version="1.0.0")), ], [("requirements.txt", make_dependency(name="requests", version="2.20.0"))], "changed_packages", (Ecosystem.PIP, "flask"), id="introduced-package-in-changed-packages", ), pytest.param( [("requirements.txt", make_dependency(name="requests", version="2.25.0"))], [ ("requirements.txt", make_dependency(name="requests", version="2.20.0")), ("requirements.txt", make_dependency(name="flask", version="1.0.0")), ], "removed", (Ecosystem.PIP, "flask"), id="removed-package", ), pytest.param( [ ("requirements.txt", make_dependency(name="requests", version="2.25.0")), ("requirements.txt", make_dependency(name="flask", version="1.0.0")), ], [("requirements.txt", make_dependency(name="requests", version="2.20.0"))], "introduced_versioned", (Ecosystem.PIP, "flask", "1.0.0"), id="introduced-versioned-contains-new-package", ), ], ) def test_compute_deltas( self, scanner: VulnerabilityScanner, workspace_deps: list[tuple[str, Dependency]], default_deps: list[tuple[str, Dependency]], check_field: str, expected_pkg: tuple, ) -> None: workspace = make_graph(*workspace_deps) default = make_graph(*default_deps) removed, changed_packages, introduced_versioned = scanner._compute_deltas( workspace, default ) target = { "changed_packages": changed_packages, "removed": removed, "introduced_versioned": introduced_versioned, }[check_field] assert expected_pkg in target def test_compute_deltas_introduced_versioned_excludes_preexisting_version( self, scanner: VulnerabilityScanner, ) -> None: """ lodash@4.17.21 exists on default branch. lodash@4.17.15 is added in the PR as a new transitive dep. Only 4.17.15 should appear in introduced_versioned — not 4.17.21. """ workspace = make_graph( ( "yarn.lock", make_dependency( name="lodash", version="4.17.21", ecosystem=Ecosystem.NPM, manifest_path="yarn.lock", is_lockfile=True, ), ), ( "yarn.lock", make_dependency( name="lodash", version="4.17.15", ecosystem=Ecosystem.NPM, manifest_path="yarn.lock", is_lockfile=True, ), ), ) default = make_graph( ( "yarn.lock", make_dependency( name="lodash", version="4.17.21", ecosystem=Ecosystem.NPM, manifest_path="yarn.lock", is_lockfile=True, ), ), ) _, _, introduced_versioned = scanner._compute_deltas(workspace, default) assert (Ecosystem.NPM, "lodash", "4.17.15") in introduced_versioned assert (Ecosystem.NPM, "lodash", "4.17.21") not in introduced_versioned # -------------------------------------------------------- # Alert evaluation # -------------------------------------------------------- class TestAlerts: def test_fetch_alerts(self, scanner: VulnerabilityScanner) -> None: workspace = make_graph(("requirements.txt", make_dependency(version="2.25.0"))) alerts = [make_alert(severity=Severity.CRITICAL)] scanner._alerts_client.get_open_alerts.return_value = alerts result = scanner._fetch_alerts(workspace) assert len(result) == 1 assert result[0].ghsa_id == "GHSA-1234-5678-90ab" scanner._alerts_client.get_open_alerts.assert_called_once() @pytest.mark.parametrize( "has_deps, has_handler, expected_count, dep_is_none", [ pytest.param(True, True, 1, False, id="matching-dependency"), pytest.param(False, True, 1, True, id="no-matching-dependency"), pytest.param(True, False, 0, None, id="no-handler-for-ecosystem"), ], ) def test_evaluate_alerts( self, scanner: VulnerabilityScanner, has_deps: bool, has_handler: bool, expected_count: int, dep_is_none: bool | None, ) -> None: workspace = ( make_graph(("requirements.txt", make_dependency(version="2.25.0"))) if has_deps else DependencyGraph() ) alert = make_alert( package_name="requests" if has_deps else "nonexistent", ) scanner._registry.for_ecosystem.return_value = _mock_handler() if has_handler else None scanner._policy.evaluate.return_value = MagicMock( status=FindingStatus.VULNERABLE if has_deps else FindingStatus.REMOVED, ) findings = scanner._evaluate_alerts(workspace, [alert], set(), set()) assert len(findings) == expected_count if dep_is_none is not None and expected_count > 0: assert (findings[0].dependency is None) == dep_is_none def test_evaluate_alerts_preexisting_version_gets_grace_period( self, scanner: VulnerabilityScanner, ) -> None: """ lodash@4.17.21 exists on both branches (not introduced). lodash@4.17.15 is newly introduced. An alert matching both should apply grace period to 4.17.21 and treat 4.17.15 as introduced (no grace period). """ workspace = make_graph( ( "yarn.lock", make_dependency( name="lodash", version="4.17.21", ecosystem=Ecosystem.NPM, manifest_path="yarn.lock", is_lockfile=True, ), ), ( "yarn.lock", make_dependency( name="lodash", version="4.17.15", ecosystem=Ecosystem.NPM, manifest_path="yarn.lock", is_lockfile=True, ), ), ) alert = make_alert( package_name="lodash", ecosystem=Ecosystem.NPM, manifest_path="yarn.lock", ) introduced_versioned = {(Ecosystem.NPM, "lodash", "4.17.15")} scanner._registry.for_ecosystem.return_value = _mock_handler(is_vulnerable=True) scanner._policy.evaluate.return_value = MagicMock( status=FindingStatus.VULNERABLE, is_blocking=True ) scanner._evaluate_alerts(workspace, [alert], set(), introduced_versioned) calls = scanner._policy.evaluate.call_args_list assert len(calls) == 2 introduced_flags = {call.kwargs["is_introduced"] for call in calls} # One dep is introduced (4.17.15), one is not (4.17.21) assert introduced_flags == {True, False} def test_handle_no_match_with_removed_package(self, scanner: VulnerabilityScanner) -> None: scanner._policy.evaluate.return_value = MagicMock(status=FindingStatus.REMOVED) finding = scanner._handle_no_match(make_alert(), {(Ecosystem.PIP, "requests")}) assert finding.dependency is None scanner._policy.evaluate.assert_called_once() # -------------------------------------------------------- # Advisory evaluation # -------------------------------------------------------- class TestAdvisories: @pytest.mark.parametrize( "handler_exists, batch_result, has_workspace_deps, expected_count", [ pytest.param( True, {("PIP", "requests"): [make_alert()]}, True, 1, id="batch-fetches-and-evaluates", ), pytest.param( True, {("PIP", "requests"): []}, True, 0, id="no-advisories-skips", ), pytest.param( False, {}, True, 0, id="no-handler-skips", ), pytest.param( True, {("PIP", "requests"): [make_alert()]}, False, 0, id="no-workspace-deps-skips", ), ], ) def test_evaluate_advisories( self, scanner: VulnerabilityScanner, handler_exists: bool, batch_result: dict[tuple[str, str], list], has_workspace_deps: bool, expected_count: int, ) -> None: workspace = ( make_graph(("requirements.txt", make_dependency(version="2.25.0"))) if has_workspace_deps else DependencyGraph() ) changed = {(Ecosystem.PIP, "requests")} if handler_exists: scanner._registry.for_ecosystem.return_value = _mock_handler() scanner._advisory_client.get_advisories_batch.return_value = batch_result else: scanner._registry.for_ecosystem.return_value = None scanner._policy.evaluate.return_value = MagicMock( status=FindingStatus.VULNERABLE, is_blocking=True, ) findings = scanner._evaluate_advisories(workspace, changed, set()) assert len(findings) == expected_count if not handler_exists or not has_workspace_deps: scanner._advisory_client.get_advisories_batch.assert_not_called() else: scanner._advisory_client.get_advisories_batch.assert_called_once_with( [("PIP", "requests")] ) def test_fetch_advisories_for_work_items_rekeys_batch_results( self, scanner: VulnerabilityScanner, ) -> None: handler = _mock_handler() deps = [make_dependency(name="requests", version="2.25.0")] work_items = [(Ecosystem.PIP, "requests", handler, deps)] advisory = make_alert(package_name="requests") scanner._advisory_client.get_advisories_batch.return_value = { ("PIP", "requests"): [advisory] } result = scanner._fetch_advisories_for_work_items(work_items) assert result == {(Ecosystem.PIP, "requests"): [advisory]} scanner._advisory_client.get_advisories_batch.assert_called_once_with([("PIP", "requests")]) def test_fetch_advisories_for_work_items_returns_empty_when_batch_missing_key( self, scanner: VulnerabilityScanner, ) -> None: handler = _mock_handler() deps = [make_dependency(name="requests", version="2.25.0")] work_items = [(Ecosystem.PIP, "requests", handler, deps)] scanner._advisory_client.get_advisories_batch.return_value = {} result = scanner._fetch_advisories_for_work_items(work_items) assert result == {(Ecosystem.PIP, "requests"): []} def test_build_advisory_work_items_skips_missing_deps_and_missing_handler( self, scanner: VulnerabilityScanner, ) -> None: deps_by_package = { (Ecosystem.PIP, "requests"): [make_dependency(name="requests", version="2.25.0")] } changed_packages = { (Ecosystem.PIP, "requests"), (Ecosystem.PIP, "flask"), } def _registry_lookup(ecosystem: Ecosystem): return _mock_handler() if ecosystem == Ecosystem.PIP else None scanner._registry.for_ecosystem.side_effect = _registry_lookup work_items = scanner._build_advisory_work_items(changed_packages, deps_by_package) assert len(work_items) == 1 assert work_items[0][0] == Ecosystem.PIP assert work_items[0][1] == "requests" # -------------------------------------------------------- # Filtering & sorting # -------------------------------------------------------- class TestFiltering: def test_deduplicates_by_ghsa_id(self, scanner: VulnerabilityScanner) -> None: alert = make_alert(ghsa_id="GHSA-1") findings = [_mock_finding(alert=alert), _mock_finding(alert=alert)] assert len(scanner._filter_and_sort_findings(findings)) == 1 def test_keeps_only_vulnerable(self, scanner: VulnerabilityScanner) -> None: findings = [ _mock_finding(alert=make_alert(ghsa_id="GHSA-1")), _mock_finding( alert=make_alert(package_name="flask", ghsa_id="GHSA-2"), status=FindingStatus.FIXED, ), ] result = scanner._filter_and_sort_findings(findings) assert len(result) == 1 assert result[0].alert.ghsa_id == "GHSA-1" def test_selects_highest_patch_version(self, scanner: VulnerabilityScanner) -> None: """Two findings for the same installed version should collapse, keeping the highest patch.""" findings = [ _mock_finding( dep=make_dependency(version="2.25.0"), alert=make_alert(ghsa_id="GHSA-1", first_patched="2.26.0"), ), _mock_finding( dep=make_dependency(version="2.25.0"), alert=make_alert(ghsa_id="GHSA-2", first_patched="2.27.0"), ), ] result = scanner._filter_and_sort_findings(findings) assert len(result) == 1 assert result[0].alert.first_patched_version == "2.27.0" def test_elevates_severity_and_blocking(self, scanner: VulnerabilityScanner) -> None: """Collapsed group should pick worst severity and most restrictive blocking. Covers the `return Finding(...)` branch in _pick_representative (line 342). """ findings = [ Finding( dependency=make_dependency(version="1.0.0"), alert=make_alert( ghsa_id="GHSA-1", first_patched="1.2.0", severity=Severity.LOW, ), decision=PolicyDecision( blocking_date=None, remaining_days=None, is_blocking=False, status=FindingStatus.VULNERABLE, message="low severity", reason="low", ), ), Finding( dependency=make_dependency(version="1.0.0"), alert=make_alert( ghsa_id="GHSA-2", first_patched="1.1.0", severity=Severity.CRITICAL, ), decision=PolicyDecision( blocking_date=None, remaining_days=None, is_blocking=True, status=FindingStatus.VULNERABLE, message="critical severity", reason="critical", ), ), ] result = scanner._filter_and_sort_findings(findings) assert len(result) == 1 # Patch from best (highest patch = 1.2.0) assert result[0].alert.first_patched_version == "1.2.0" # Severity from worst (CRITICAL) assert result[0].alert.severity == Severity.CRITICAL # Blocking from any (True) assert result[0].decision.is_blocking is True # Reason from worst assert result[0].decision.reason == "critical" def test_elevated_decision_fields_come_from_the_blocking_finding( self, scanner: VulnerabilityScanner ) -> None: """Regression test: duplicate alerts for the same package/version (e.g. GitHub reporting both "pyjwt" and "PyJWT" alerts) must not produce a collapsed row that mixes a not-yet-blocking finding's blocking_date/remaining_days with another finding's is_blocking/reason. """ not_yet_blocking = Finding( dependency=make_dependency(version="2.12.1"), alert=make_alert( ghsa_id="GHSA-1", first_patched="2.13.0", severity=Severity.HIGH, ), decision=PolicyDecision( blocking_date=datetime(2026, 8, 15, tzinfo=UTC), remaining_days=44, is_blocking=False, status=FindingStatus.VULNERABLE, message="Dependency is within vulnerable range", reason="44 days remaining", ), ) already_blocking = Finding( dependency=make_dependency(version="2.12.1"), alert=make_alert( ghsa_id="GHSA-2", first_patched="2.13.0", severity=Severity.HIGH, ), decision=PolicyDecision( blocking_date=datetime(2026, 1, 1, tzinfo=UTC), remaining_days=0, is_blocking=True, status=FindingStatus.VULNERABLE, message="Dependency is within vulnerable range", reason="Grace period exceeded", ), ) result = scanner._filter_and_sort_findings([not_yet_blocking, already_blocking]) assert len(result) == 1 decision = result[0].decision # Blocking status, date, remaining days and reason must all agree with each other. assert decision.is_blocking is True assert decision.blocking_date == datetime(2026, 1, 1, tzinfo=UTC) assert decision.remaining_days == 0 assert decision.reason == "Grace period exceeded" def test_different_versions_not_collapsed(self, scanner: VulnerabilityScanner) -> None: """Two findings with different installed versions should remain separate.""" findings = [ _mock_finding( dep=make_dependency(version="2.24.0"), alert=make_alert(ghsa_id="GHSA-1", first_patched="2.26.0"), ), _mock_finding( dep=make_dependency(version="2.25.0"), alert=make_alert(ghsa_id="GHSA-2", first_patched="2.27.0"), ), ] result = scanner._filter_and_sort_findings(findings) assert len(result) == 2 def test_finding_without_dependency(self, scanner: VulnerabilityScanner) -> None: """Covers _installed_version, _manifest_path, and _sort_findings when dependency is None.""" findings = [ _mock_finding( dep=None, alert=make_alert(ghsa_id="GHSA-1"), ), _mock_finding( dep=make_dependency(version="1.0.0"), alert=make_alert(ghsa_id="GHSA-2", package_name="flask"), ), ] result = scanner._filter_and_sort_findings(findings) assert len(result) == 2 # Both should be present — no crash on None dependency dep_values = [f.dependency for f in result] assert None in dep_values # -------------------------------------------------------- # Version parsing # -------------------------------------------------------- class TestVersionParsing: @pytest.mark.parametrize( "version, expected", [ pytest.param("1.2.3", (1, 2, 3), id="valid-semver"), pytest.param("2.0.0", (2, 0, 0), id="valid-zeros"), pytest.param("10.20.30", (10, 20, 30), id="valid-large"), pytest.param(None, (0,), id="none"), pytest.param("1.2.x", (0,), id="invalid-suffix"), pytest.param("invalid", (0,), id="invalid-string"), ], ) def test_parse_patch_version( self, scanner: VulnerabilityScanner, version: str | None, expected: tuple, ) -> None: assert scanner._parse_patch_version(version) == expected # -------------------------------------------------------- # Publishing # -------------------------------------------------------- class TestPublishing: @pytest.mark.parametrize( "findings, expected_status, expect_comment, expected_has_blocking", [ pytest.param( [_mock_finding(is_blocking=True)], "failure", True, True, id="blocking-sets-failure-and-comments", ), pytest.param( [_mock_finding(is_blocking=False)], "success", True, False, id="non-blocking-sets-success-and-comments", ), pytest.param( [], "success", False, None, id="empty-sets-success-no-comment", ), ], ) def test_publish( self, scanner: VulnerabilityScanner, findings: list[Finding], expected_status: str, expect_comment: bool, expected_has_blocking: bool | None, ) -> None: scanner._publish(findings, "report content", "https://example.com") scanner._publisher.set_status.assert_called_with(expected_status) if expect_comment: scanner._publisher.comment.assert_called_once_with( "report content", "https://example.com", expected_has_blocking, ) else: scanner._publisher.comment.assert_not_called() # -------------------------------------------------------- # Full flow # -------------------------------------------------------- class TestFullFlow: def test_full_scan_workflow(self, scanner: VulnerabilityScanner) -> None: scanner._workspace_scanner.scan.return_value = make_graph( ("requirements.txt", make_dependency(name="requests", version="2.25.0")) ) scanner._default_branch_scanner.scan.return_value = make_graph( ("requirements.txt", make_dependency(name="requests", version="2.20.0")) ) alert = make_alert(package_name="requests", first_patched="2.26.0") scanner._alerts_client.get_open_alerts.return_value = [alert] scanner._registry.for_ecosystem.return_value = _mock_handler() decision = MagicMock(status=FindingStatus.VULNERABLE, is_blocking=True) scanner._policy.evaluate.return_value = decision scanner._reporter.render.return_value = "report output" result = scanner.scan(_cfg()) assert isinstance(result, ScanResult) assert len(result.findings) >= 1 scanner._publisher.set_status.assert_called()