from __future__ import annotations from typing import Any from unittest.mock import MagicMock import pytest from tests.helpers import make_alert from vuln_scan.github.advisory_client import GitHubAdvisoryClient from vuln_scan.github.base_client import AuthConfig # -------------------------------------------------------- # Fixtures & helpers # -------------------------------------------------------- @pytest.fixture def advisory_client(monkeypatch: pytest.MonkeyPatch) -> GitHubAdvisoryClient: monkeypatch.setattr(GitHubAdvisoryClient, "__post_init__", lambda self: None) return GitHubAdvisoryClient(auth_config=AuthConfig(token="t")) def _page( ghsa_ids: list[str], *, has_next: bool = False, end_cursor: str | None = None, ) -> dict[str, Any]: """Build a single GraphQL securityVulnerabilities page.""" return { "securityVulnerabilities": { "nodes": [_node(gid) for gid in ghsa_ids], "pageInfo": { "hasNextPage": has_next, "endCursor": end_cursor, }, } } def _node(ghsa_id: str) -> dict[str, Any]: return { "severity": "HIGH", "vulnerableVersionRange": "< 1.2.3", "package": {"name": "requests", "ecosystem": "PIP"}, "firstPatchedVersion": {"identifier": "1.2.3"}, "advisory": { "ghsaId": ghsa_id, "publishedAt": "2024-01-01T00:00:00Z", }, } @pytest.fixture(autouse=True) def _patch_from_advisory_node(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( "vuln_scan.github.advisory_client.SecurityVulnerability.from_advisory_node", lambda node: make_alert( ghsa_id=node["advisory"]["ghsaId"], package_name=node["package"]["name"], source="advisory", ), ) # -------------------------------------------------------- # get_advisories # -------------------------------------------------------- @pytest.mark.parametrize( "pages, max_pages, expected_ids, expected_calls", [ pytest.param( [_page(["GHSA-1"])], 10, ["GHSA-1"], 1, id="single-page", ), pytest.param( [ _page(["GHSA-1"], has_next=True, end_cursor="c1"), _page(["GHSA-2"]), ], 10, ["GHSA-1", "GHSA-2"], 2, id="multiple-pages", ), pytest.param( [_page(["GHSA-1"], has_next=True, end_cursor=None)], 10, ["GHSA-1"], 1, id="stops-on-missing-cursor", ), pytest.param( [ _page(["GHSA-1"], has_next=True, end_cursor="c1"), _page(["GHSA-2"], has_next=True, end_cursor="c2"), _page(["GHSA-3"], has_next=True, end_cursor="c3"), ], 2, ["GHSA-1", "GHSA-2"], 2, id="respects-max-pages", ), ], ) def test_get_advisories( monkeypatch: pytest.MonkeyPatch, advisory_client: GitHubAdvisoryClient, pages: list[dict[str, Any]], max_pages: int, expected_ids: list[str], expected_calls: int, ) -> None: calls: list[dict[str, Any]] = [] page_iter = iter(pages) def fake_graphql_with_retry(self, query, variables): calls.append(variables) return next(page_iter) monkeypatch.setattr( GitHubAdvisoryClient, "_graphql_with_retry", fake_graphql_with_retry, ) result = advisory_client.get_advisories("PIP", "requests", max_pages=max_pages) assert [item.ghsa_id for item in result] == expected_ids assert len(calls) == expected_calls assert calls[0]["ecosystem"] == "PIP" assert calls[0]["package"] == "requests" assert calls[0]["cursor"] is None assert calls[0]["page_size"] == 100 def test_get_advisories_passes_cursor_to_next_page( monkeypatch: pytest.MonkeyPatch, advisory_client: GitHubAdvisoryClient, ) -> None: calls: list[dict[str, Any]] = [] pages = iter( [ _page(["GHSA-1"], has_next=True, end_cursor="cursor-1"), _page(["GHSA-2"]), ] ) def fake_graphql_with_retry(self, query, variables): calls.append(dict(variables)) return next(pages) monkeypatch.setattr( GitHubAdvisoryClient, "_graphql_with_retry", fake_graphql_with_retry, ) advisory_client.get_advisories("PIP", "requests") assert len(calls) == 2 assert calls[0]["cursor"] is None assert calls[1]["cursor"] == "cursor-1" # -------------------------------------------------------- # get_advisories_batch # -------------------------------------------------------- def test_get_advisories_batch_returns_empty_for_no_packages( advisory_client: GitHubAdvisoryClient, ) -> None: assert advisory_client.get_advisories_batch([]) == {} def test_get_advisories_batch_single_chunk( monkeypatch: pytest.MonkeyPatch, advisory_client: GitHubAdvisoryClient, ) -> None: packages = [("PIP", "requests"), ("PIP", "flask")] monkeypatch.setattr( advisory_client, "_fetch_chunk", lambda chunk, page_size: ( { ("PIP", "requests"): [make_alert(ghsa_id="GHSA-1", source="advisory")], ("PIP", "flask"): [ make_alert(ghsa_id="GHSA-2", package_name="flask", source="advisory") ], }, {"cost": 2, "remaining": 4998, "limit": 5000, "resetAt": "2026-01-01T00:00:00Z"}, ), ) result = advisory_client.get_advisories_batch(packages, batch_size=10) assert set(result.keys()) == set(packages) assert [a.ghsa_id for a in result[("PIP", "requests")]] == ["GHSA-1"] assert [a.ghsa_id for a in result[("PIP", "flask")]] == ["GHSA-2"] def test_get_advisories_batch_multiple_chunks( monkeypatch: pytest.MonkeyPatch, advisory_client: GitHubAdvisoryClient, ) -> None: packages = [ ("PIP", "requests"), ("PIP", "flask"), ("PIP", "django"), ] def fake_fetch_chunk(chunk, page_size): return ( { pkg: [make_alert(ghsa_id=f"GHSA-{name}", package_name=name, source="advisory")] for _, name in chunk for pkg in [(chunk[[p[1] for p in chunk].index(name)][0], name)] }, {"cost": len(chunk)}, ) monkeypatch.setattr(advisory_client, "_fetch_chunk", fake_fetch_chunk) result = advisory_client.get_advisories_batch(packages, batch_size=2) assert set(result.keys()) == set(packages) assert result[("PIP", "requests")][0].ghsa_id == "GHSA-requests" assert result[("PIP", "flask")][0].ghsa_id == "GHSA-flask" assert result[("PIP", "django")][0].ghsa_id == "GHSA-django" def test_get_advisories_batch_keeps_empty_result_for_failed_chunk( monkeypatch: pytest.MonkeyPatch, advisory_client: GitHubAdvisoryClient, ) -> None: packages = [ ("PIP", "requests"), ("PIP", "flask"), ] def fake_fetch_chunk(chunk, page_size): if ("PIP", "flask") in chunk: raise RuntimeError("boom") return ( {("PIP", "requests"): [make_alert(ghsa_id="GHSA-1", source="advisory")]}, {"cost": 1}, ) monkeypatch.setattr(advisory_client, "_fetch_chunk", fake_fetch_chunk) result = advisory_client.get_advisories_batch(packages, batch_size=1) assert [a.ghsa_id for a in result[("PIP", "requests")]] == ["GHSA-1"] assert result[("PIP", "flask")] == [] def test_get_advisories_batch_uses_capped_worker_count( advisory_client: GitHubAdvisoryClient, ) -> None: advisory_client.max_workers = 99 packages = [("PIP", "a"), ("PIP", "b"), ("PIP", "c")] advisory_client._fetch_chunk = MagicMock( side_effect=[ ({("PIP", "a"): []}, None), ({("PIP", "b"): []}, None), ({("PIP", "c"): []}, None), ] ) result = advisory_client.get_advisories_batch(packages, batch_size=1) assert set(result.keys()) == set(packages) assert advisory_client._fetch_chunk.call_count == 3 # -------------------------------------------------------- # _fetch_chunk # -------------------------------------------------------- def test_fetch_chunk_parses_all_results( monkeypatch: pytest.MonkeyPatch, advisory_client: GitHubAdvisoryClient, ) -> None: monkeypatch.setattr( advisory_client, "_build_batch_query", lambda chunk, page_size: ("query", {"page_size": page_size}), ) monkeypatch.setattr( advisory_client, "_graphql_with_retry", lambda query, variables: { "result_0": { "nodes": [_node("GHSA-1")], "pageInfo": {"hasNextPage": False, "endCursor": None}, }, "result_1": { "nodes": [_node("GHSA-2")], "pageInfo": {"hasNextPage": False, "endCursor": None}, }, "rateLimit": {"cost": 2, "remaining": 4998, "limit": 5000, "resetAt": "x"}, }, ) chunk_results, rate_limit = advisory_client._fetch_chunk( [("PIP", "requests"), ("PIP", "flask")], page_size=100, ) assert [a.ghsa_id for a in chunk_results[("PIP", "requests")]] == ["GHSA-1"] assert [a.ghsa_id for a in chunk_results[("PIP", "flask")]] == ["GHSA-2"] assert rate_limit == {"cost": 2, "remaining": 4998, "limit": 5000, "resetAt": "x"} def test_fetch_chunk_falls_back_to_get_advisories_for_paginated_results( monkeypatch: pytest.MonkeyPatch, advisory_client: GitHubAdvisoryClient, ) -> None: monkeypatch.setattr( advisory_client, "_build_batch_query", lambda chunk, page_size: ("query", {"page_size": page_size}), ) monkeypatch.setattr( advisory_client, "_graphql_with_retry", lambda query, variables: { "result_0": { "nodes": [_node("GHSA-1")], "pageInfo": {"hasNextPage": True, "endCursor": "c1"}, }, "rateLimit": {"cost": 1}, }, ) fallback = [make_alert(ghsa_id="GHSA-full", source="advisory")] advisory_client.get_advisories = MagicMock(return_value=fallback) chunk_results, rate_limit = advisory_client._fetch_chunk( [("PIP", "requests")], page_size=50, ) assert chunk_results[("PIP", "requests")] == fallback assert rate_limit == {"cost": 1} advisory_client.get_advisories.assert_called_once_with( "PIP", "requests", page_size=50, ) def test_fetch_chunk_handles_missing_result_section( monkeypatch: pytest.MonkeyPatch, advisory_client: GitHubAdvisoryClient, ) -> None: monkeypatch.setattr( advisory_client, "_build_batch_query", lambda chunk, page_size: ("query", {"page_size": page_size}), ) monkeypatch.setattr( advisory_client, "_graphql_with_retry", lambda query, variables: {"rateLimit": {"cost": 0}}, ) chunk_results, rate_limit = advisory_client._fetch_chunk( [("PIP", "requests")], page_size=100, ) assert chunk_results == {("PIP", "requests"): []} assert rate_limit == {"cost": 0} # -------------------------------------------------------- # _build_batch_query # -------------------------------------------------------- def test_build_batch_query_contains_expected_variables_and_aliases( advisory_client: GitHubAdvisoryClient, ) -> None: query, variables = advisory_client._build_batch_query( [("PIP", "requests"), ("RUBYGEMS", "rails")], page_size=123, ) assert "$page_size: Int!" in query assert "$eco_0: SecurityAdvisoryEcosystem!" in query assert "$pkg_0: String!" in query assert "$eco_1: SecurityAdvisoryEcosystem!" in query assert "$pkg_1: String!" in query assert "result_0: securityVulnerabilities(" in query assert "result_1: securityVulnerabilities(" in query assert "rateLimit { cost remaining limit resetAt }" in query assert variables == { "page_size": 123, "eco_0": "PIP", "pkg_0": "requests", "eco_1": "RUBYGEMS", "pkg_1": "rails", } # -------------------------------------------------------- # _extract_security_vulnerabilities_page # -------------------------------------------------------- @pytest.mark.parametrize( "data, expected_nodes, expected_page_info", [ pytest.param( _page(["GHSA-1"], has_next=True, end_cursor="c1"), [_node("GHSA-1")], {"hasNextPage": True, "endCursor": "c1"}, id="normal-shape", ), pytest.param( {}, [], {}, id="missing-root", ), pytest.param( {"securityVulnerabilities": None}, [], {}, id="none-root", ), pytest.param( {"securityVulnerabilities": {"nodes": None, "pageInfo": None}}, [], {}, id="none-fields", ), ], ) def test_extract_security_vulnerabilities_page( advisory_client: GitHubAdvisoryClient, data: dict[str, Any], expected_nodes: list[dict[str, Any]], expected_page_info: dict[str, Any], ) -> None: nodes, page_info = advisory_client._extract_security_vulnerabilities_page(data) assert nodes == expected_nodes assert page_info == expected_page_info # -------------------------------------------------------- # _parse_advisory_nodes # -------------------------------------------------------- def test_parse_advisory_nodes_parses_valid_nodes( advisory_client: GitHubAdvisoryClient, ) -> None: result = advisory_client._parse_advisory_nodes( [_node("GHSA-1"), _node("GHSA-2")], "PIP", "requests", ) assert [item.ghsa_id for item in result] == ["GHSA-1", "GHSA-2"] def test_parse_advisory_nodes_skips_invalid_nodes( monkeypatch: pytest.MonkeyPatch, advisory_client: GitHubAdvisoryClient, ) -> None: def fake_from_advisory_node(node): ghsa_id = node["advisory"]["ghsaId"] if ghsa_id == "GHSA-bad": raise ValueError("bad node") return make_alert(ghsa_id=ghsa_id, source="advisory") monkeypatch.setattr( "vuln_scan.github.advisory_client.SecurityVulnerability.from_advisory_node", fake_from_advisory_node, ) result = advisory_client._parse_advisory_nodes( [_node("GHSA-good"), _node("GHSA-bad"), _node("GHSA-good-2")], "PIP", "requests", ) assert [item.ghsa_id for item in result] == ["GHSA-good", "GHSA-good-2"]