Open_Ultra_SAST/tests/test_unseen_source.py

"""Sanitized REST-shaped advisory fixtures; every external client is a fake."""

import copy
import importlib
import json
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
s = importlib.import_module("benchmarks.unseen.source")


def advisory():
    path = Path(__file__).resolve().parents[1] / "benchmarks/unseen/fixtures/advisory.json"
    return json.loads(path.read_bytes())


class API:
    def __init__(self):
        self.parents = ["b" * 40]
        self.spdx = "MIT"
        self.license_text = "SPDX-License-Identifier: MIT"
        self.calls = []
        self.repo = {"full_name": "fixture/project", "private": False, "size": 100, "clone_url": "https://github.com/fixture/project.git"}

    def get_repo(self, name):
        return copy.deepcopy(self.repo)

    def get_commit(self, name, sha):
        return {"sha": sha, "parents": [{"sha": p} for p in self.parents]}

    def get_license(self, name, ref):
        self.calls.append((name, ref))
        return {"text": self.license_text, "spdx_id": self.spdx, "path": "LICENSE"}


class OSV:
    def get(self, identifier):
        return {
            "affected": [
                {
                    "package": {"ecosystem": "PyPI", "name": "fixture"},
                    "ranges": [{"type": "ECOSYSTEM", "events": [{"introduced": "1.2"}, {"fixed": "1.3"}]}],
                }
            ]
        }


def test_candidate_cwe_cutoff_bracket_and_pinned_license():
    api = API()
    candidate = s.candidate(advisory(), api, OSV())
    assert candidate.family == "injection" and candidate.post_cutoff
    assert candidate.first_affected == "1.2"
    assert candidate.license_class == "permissive"
    assert api.calls == [("fixture/project", "b" * 40)]
    raw = advisory()
    raw["published_at"] = "2026-06-01T12:00:00Z"
    assert not s.candidate(raw, api, OSV()).post_cutoff
    raw["published_at"] = "2018-12-31T00:00:00Z"
    with pytest.raises(s.Rejected, match="date"):
        s.candidate(raw, api, OSV())


@pytest.mark.parametrize("spdx", ["GPL-3.0", "AGPL-3.0-only", "LGPL-2.1", "NOASSERTION", "", "BUSL-1.1", "MIT OR GPL-3.0"])
def test_private_licenses(spdx):
    api = API()
    api.spdx = spdx
    api.license_text = ""
    assert s.candidate(advisory(), api, OSV()).license_class == "private"


def test_license_file_overrides_api_and_missing_file_is_private():
    api = API()
    api.license_text = "SPDX-License-Identifier: GPL-3.0-only"
    assert s.candidate(advisory(), api, OSV()).license_class == "private"
    api.license_text = ""
    api.spdx = "Apache-2.0"
    assert s.candidate(advisory(), api, OSV()).license_spdx == "Apache-2.0"


@pytest.mark.parametrize(
    "mutation,reason",
    [
        (lambda a, api: api.parents.append("c" * 40), "parents"),
        (lambda a, api: a["references"].append("https://github.com/fixture/other/commit/" + "d" * 40), "fix_links"),
        (lambda a, api: a.update(cwes=[{"cwe_id": "CWE-99999"}]), "family"),
        (lambda a, api: api.repo.update(size=512001), "repository"),
        (lambda a, api: api.repo.update(private=True), "repository"),
        (lambda a, api: api.repo.update(clone_url="ssh://fixture/project"), "repository"),
    ],
)
def test_rejections(mutation, reason):
    api, raw = API(), advisory()
    mutation(raw, api)
    with pytest.raises(s.Rejected, match=reason):
        s.candidate(raw, api, OSV())


def test_post_cutoff_first_within_ecosystem_is_seeded():
    rows = [advisory() for _ in range(12)]
    for i, row in enumerate(rows):
        row["ghsa_id"] += str(i)
        if i % 2:
            row["published_at"] = "2025-01-01T00:00:00Z"
    ordered = s.ordered_advisories(rows, 17)
    assert ordered == s.ordered_advisories(list(reversed(rows)), 17)
    assert all(s.post_cutoff(row) for row in ordered[:6])


def test_http_adapters_with_fake_transport():
    import base64
    from urllib.parse import parse_qs, urlsplit

    class Transport:
        def __init__(self):
            self.calls = []

        def get(self, url, headers):
            self.calls.append((url, headers))
            parsed = urlsplit(url)
            if parsed.path == "/advisories":
                page = int(parse_qs(parsed.query)["page"][0])
                return [advisory()] * (100 if page == 1 else 1)
            if "/license" in parsed.path:
                assert parse_qs(parsed.query)["ref"] == ["b" * 40]
                return {
                    "license": {"spdx_id": "MIT"},
                    "path": "LICENSE",
                    "content": base64.b64encode(b"SPDX-License-Identifier: MIT").decode(),
                }
            if "/commits/" in parsed.path:
                return {"sha": "a" * 40, "parents": [{"sha": "b" * 40}]}
            if parsed.hostname == "api.osv.dev":
                return OSV().get("ignored")
            return API().repo

    transport = Transport()
    api = s.GitHub("fake-token", transport)
    assert len(list(api.advisories())) == 101
    assert s.candidate(advisory(), api, s.OSV(transport)).first_affected == "1.2"
    assert all(headers["Authorization"] == "Bearer fake-token" for url, headers in transport.calls if "api.github.com" in url)
    count = api.calls
    api.get_repo("fixture/project")
    assert api.calls == count


def test_http_failures_are_not_empty_results():
    class Transport:
        def get(self, url, headers):
            raise ValueError("http_failure")

    with pytest.raises(ValueError, match="http_failure"):
        list(s.GitHub("fake", Transport()).advisories())
    with pytest.raises(ValueError, match="http_failure"):
        s.OSV(Transport()).get("GHSA-test-test-test")


def test_transport_rate_limit_retry_and_sanitized_failure():
    import io
    from urllib.error import HTTPError

    waits, attempts = [], []

    def opener(request, timeout):
        attempts.append(request)
        if len(attempts) == 1:
            raise HTTPError(request.full_url, 403, "private message", {"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": "120"}, None)
        return io.BytesIO(b'{"ok": true}')

    transport = s.JSONTransport(opener=opener, sleeper=waits.append, clock=lambda: 100)
    assert transport.get("https://example.invalid", {}) == {"ok": True}
    assert waits == [60] and transport.attempts == 2

    def down(request, timeout):
        raise HTTPError(request.full_url, 503, "private name", {}, None)

    transport = s.JSONTransport(opener=down, sleeper=waits.append)
    with pytest.raises(ValueError, match="^http_failure$"):
        transport.get("https://example.invalid", {})
    assert transport.attempts == 5