diff --git a/benchmarks/unseen/draft.py b/benchmarks/unseen/draft.py index cc8eaf4..46c28cd 100644 --- a/benchmarks/unseen/draft.py +++ b/benchmarks/unseen/draft.py @@ -219,7 +219,7 @@ def inputs(used, seed, known_empty, clones): "used_set_digest": digest({label: sorted(names) for label, names in used.sources.items()}), "known_empty": sorted(known_empty), "clone_limit_bytes": getattr(clones, "limit_bytes", 500 * 1024 * 1024), - "journal_version": 3, + "journal_version": 4, } # Hardcoded design thresholds and algorithms are covered as well as taxonomy data. for name in ("draft", "source", "extract", "eligibility", "journal"): @@ -232,15 +232,19 @@ def enumerate_advisories(github, journal): records = {} for ecosystem in source.ECOSYSTEMS: page = 1 + next_url = None while True: key = f"{ecosystem}:{page}" if key not in journal.rows["advisories"]: journal.boundary() - result = github.advisory_page(page, ecosystem=ecosystem) - journal.put("advisories", key, {"rows": [source.reduce_advisory(row) for row in result["rows"]], "done": result["done"]}) + result = github.advisory_page(next_url, ecosystem=ecosystem) + journal.put( + "advisories", key, {"rows": [source.reduce_advisory(row) for row in result["rows"]], "next_url": result["next_url"]} + ) result = journal.rows["advisories"][key] records.update((row["ghsa_id"], row) for row in result["rows"]) - if result["done"]: + next_url = result["next_url"] + if next_url is None: break page += 1 if not records: diff --git a/benchmarks/unseen/source.py b/benchmarks/unseen/source.py index 9c700a6..3b1f118 100644 --- a/benchmarks/unseen/source.py +++ b/benchmarks/unseen/source.py @@ -181,6 +181,7 @@ class JSONTransport: self.clock = clock or time.time self.attempts = 0 self.rate_limit = {} + self.response_headers = {} def get(self, url: str, headers: dict) -> object: from http.client import IncompleteRead @@ -194,7 +195,9 @@ class JSONTransport: with self.opener(Request(url, headers=headers), timeout=60) as response: retry_headers = getattr(response, "headers", {}) self.record_rate(retry_headers) - return json.load(response) + body = json.load(response) + self.response_headers = retry_headers + return body except HTTPError as exc: self.record_rate(exc.headers) if exc.code == 404: @@ -247,7 +250,8 @@ class GitHub: def get(self, path: str): self.calls += 1 - return self.transport.get("https://api.github.com" + path, self.headers) + url = path if path.startswith("https://api.github.com/") else "https://api.github.com" + path + return self.transport.get(url, self.headers) def get_repo(self, name: str) -> dict: name = normalize(name) @@ -269,14 +273,13 @@ class GitHub: "text": base64.b64decode(row.get("content", "")).decode("utf-8", errors="replace"), } - def advisory_page(self, page: int, *, ecosystem: str) -> dict: + def advisory_page(self, next_url: str | None = None, *, ecosystem: str) -> dict: if ecosystem not in ECOSYSTEMS: raise ValueError("invalid_ecosystem") # Oldest first: newly published advisories do not shift already persisted pages. parameters = { "type": "reviewed", "per_page": 100, - "page": page, "sort": "published", "direction": "asc", "ecosystem": ecosystem, @@ -290,25 +293,27 @@ class GitHub: # GitHub's advisories `cwes` filter expects bare numbers (89), not "CWE-89". if cwes and all(re.fullmatch(r"CWE-[1-9][0-9]*", cwe) for cwe in cwes): parameters["cwes"] = ",".join(sorted((cwe.removeprefix("CWE-") for cwe in cwes), key=int)) - rows = self.get("/advisories?" + urlencode(parameters)) + rows = self.get(next_url or "/advisories?" + urlencode(parameters)) if not isinstance(rows, list): raise ValueError("invalid_advisory_page") - return {"rows": rows, "done": len(rows) < 100} + link = self.transport.response_headers.get("Link", "") + match = re.search(r'<([^>]+)>;\s*rel="next"', link) + return {"rows": rows, "next_url": match[1] if match else None} def advisories(self): """Enumerate the candidate population, deduplicating multi-ecosystem advisories.""" seen = set() for ecosystem in ECOSYSTEMS: - page = 1 + next_url = None while True: - result = self.advisory_page(page, ecosystem=ecosystem) + result = self.advisory_page(next_url, ecosystem=ecosystem) for row in result["rows"]: if row["ghsa_id"] not in seen: seen.add(row["ghsa_id"]) yield row - if result["done"]: + next_url = result["next_url"] + if next_url is None: break - page += 1 class OSV: diff --git a/tests/test_unseen_draft.py b/tests/test_unseen_draft.py index 835b56e..f59db83 100644 --- a/tests/test_unseen_draft.py +++ b/tests/test_unseen_draft.py @@ -139,12 +139,14 @@ class FakeAPI: # P is MIT but actual introducing head is GPL: public output must use V. return {"text": "", "spdx_id": "MIT" if ref == "b" * 40 else "GPL-3.0", "path": "LICENSE"} - def advisory_page(self, page, *, ecosystem): + def advisory_page(self, next_url=None, *, ecosystem): self.calls += 1 - self.advisory_pages.append((ecosystem, page)) + self.advisory_pages.append((ecosystem, next_url)) population = list(self.advisories()) if ecosystem == "pip" else [] - rows = population[(page - 1) * 100 : page * 100] - return {"rows": rows, "done": len(rows) < 100} + offset = int(next_url.rsplit("=", 1)[1]) if next_url else 0 + rows = population[offset : offset + 100] + following = f"https://api.github.com/advisories?after={offset + 100}" if offset + 100 < len(population) else None + return {"rows": rows, "next_url": following} def advisories(self): for i in range(self.count): @@ -460,8 +462,8 @@ def test_max_hours_stops_after_completed_page(tmp_path): now = [0.0] class TimedAPI(FakeAPI): - def advisory_page(self, page, *, ecosystem): - result = super().advisory_page(page, ecosystem=ecosystem) + def advisory_page(self, next_url=None, *, ecosystem): + result = super().advisory_page(next_url, ecosystem=ecosystem) now[0] += 3601 return result @@ -544,9 +546,9 @@ def test_advisory_journal_reduces_nested_blobs_and_resumes_by_page(tmp_path): class Pages: calls = [] - def advisory_page(self, page, *, ecosystem): - self.calls.append((ecosystem, page)) - return {"rows": [raw], "done": True, "unused": "raw page blob"} + def advisory_page(self, next_url=None, *, ecosystem): + self.calls.append((ecosystem, next_url)) + return {"rows": [raw], "next_url": None, "unused": "raw page blob"} api, used = Pages(), e.UsedSet() parameters = d.inputs(used, 1, set(used.sources), FakeClones()) @@ -556,13 +558,13 @@ def test_advisory_journal_reduces_nested_blobs_and_resumes_by_page(tmp_path): assert rows == [expected] saved = [json.loads(line) for line in (path / "advisories.jsonl").read_bytes().splitlines()] assert [row["id"] for row in saved] == [f"{eco}:1" for eco in s.ECOSYSTEMS] - assert all(row["value"] == {"rows": [expected], "done": True} for row in saved) + assert all(row["value"] == {"rows": [expected], "next_url": None} for row in saved) assert b"unused" not in (path / "advisories.jsonl").read_bytes() assert b"raw advisory blob" not in (path / "advisories.jsonl").read_bytes() assert b"raw page blob" not in (path / "advisories.jsonl").read_bytes() with d.Journal(path).open(parameters) as journal: resumed = d.enumerate_advisories(api, journal) - assert api.calls == [(eco, 1) for eco in s.ECOSYSTEMS] and resumed == rows + assert api.calls == [(eco, None) for eco in s.ECOSYSTEMS] and resumed == rows class Bracket(FakeOSV): def get(self, identifier): @@ -603,12 +605,12 @@ def test_reduced_advisories_preserve_all_fix_exclusions_and_selection(): s.candidate(s.reduce_advisory(raw), FakeAPI(), FakeOSV()) -@pytest.mark.parametrize("old_version", [1, 2]) +@pytest.mark.parametrize("old_version", [1, 2, 3]) def test_old_advisory_journal_version_refused_before_calls(tmp_path, old_version): used, api, osv, clones = e.UsedSet(), FakeAPI(1), FakeOSV(), FakeClones() path = tmp_path / "journal" parameters = d.inputs(used, 7, set(used.sources), clones) - assert parameters["journal_version"] == 3 + assert parameters["journal_version"] == 4 with d.Journal(path).open({**parameters, "journal_version": old_version}) as journal: journal.put("advisories", "1", {"rows": list(api.advisories()), "done": True}) report = d.run( @@ -620,29 +622,30 @@ def test_old_advisory_journal_version_refused_before_calls(tmp_path, old_version def test_enumerate_ecosystem_pages_dedupes_and_resumes_partial_stream(tmp_path): rows = list(FakeAPI(3).advisories()) + cursor = "https://api.github.com/advisories?ecosystem=pip&after=opaque-cursor" pages = { - ("pip", 1): {"rows": [rows[0]], "done": False}, - ("pip", 2): {"rows": [rows[1]], "done": True}, - ("npm", 1): {"rows": [rows[0], rows[2]], "done": True}, - ("maven", 1): {"rows": [], "done": True}, - ("composer", 1): {"rows": [], "done": True}, + ("pip", None): {"rows": [rows[0]], "next_url": cursor}, + ("pip", cursor): {"rows": [rows[0], rows[1]], "next_url": None}, + ("npm", None): {"rows": [rows[0], rows[2]], "next_url": None}, + ("maven", None): {"rows": [], "next_url": None}, + ("composer", None): {"rows": [], "next_url": None}, } calls = [] class Pages: - def advisory_page(self, page, *, ecosystem): - calls.append((ecosystem, page)) - return pages[ecosystem, page] + def advisory_page(self, next_url=None, *, ecosystem): + calls.append((ecosystem, next_url)) + return pages[ecosystem, next_url] path = tmp_path / "journal" - with d.Journal(path).open({"journal_version": 3}) as journal: - journal.put("advisories", "pip:1", pages["pip", 1]) - with d.Journal(path).open({"journal_version": 3}) as journal: + with d.Journal(path).open({"journal_version": 4}) as journal: + journal.put("advisories", "pip:1", pages["pip", None]) + with d.Journal(path).open({"journal_version": 4}) as journal: result = d.enumerate_advisories(Pages(), journal) assert [row["ghsa_id"] for row in result] == [row["ghsa_id"] for row in rows] - assert list(journal.rows["advisories"]) == [f"{eco}:{page}" for eco, page in pages] + assert list(journal.rows["advisories"]) == ["pip:1", "pip:2", "npm:1", "maven:1", "composer:1"] assert calls == list(pages)[1:] - with d.Journal(path).open({"journal_version": 3}) as journal: + with d.Journal(path).open({"journal_version": 4}) as journal: assert d.enumerate_advisories(Pages(), journal) == result assert calls == list(pages)[1:] @@ -660,11 +663,15 @@ def test_filtered_http_population_reaches_draft_floor_with_bounded_pages(tmp_pat assert query["ecosystem"][0] in s.ECOSYSTEMS assert query["published"] == [">=2019-01-01"] assert "89" in query["cwes"][0].split(",") - eco, page = query["ecosystem"][0], int(query["page"][0]) - pages.append((eco, page)) - assert len(pages) <= 7 + eco, offset = query["ecosystem"][0], int(query.get("after", ["0"])[0]) + assert "page" not in query + pages.append((eco, offset)) + assert len(pages) <= 6 rows = population if eco == "pip" else [] - return rows[(page - 1) * 100 : page * 100] + self.response_headers = {} + if offset + 100 < len(rows): + self.response_headers = {"Link": f'<{url.split("&after=")[0]}&after={offset + 100}>; rel="next"'} + return rows[offset : offset + 100] api, used = FakeAPI(), e.UsedSet() http = s.GitHub("fake", FilteredTransport()) @@ -674,4 +681,4 @@ def test_filtered_http_population_reaches_draft_floor_with_bounded_pages(tmp_pat report = d.run(tmp_path, github=api, osv=FakeOSV(), clones=FakeClones(), used=used, seed=7, known_empty=set(used.sources)) assert report["status"] == "draft" and report["repositories"] == 300 assert report["advisories"] == 300 - assert pages == [("pip", n) for n in range(1, 5)] + [(eco, 1) for eco in s.ECOSYSTEMS[1:]] + assert pages == [("pip", n) for n in (0, 100, 200)] + [(eco, 0) for eco in s.ECOSYSTEMS[1:]] diff --git a/tests/test_unseen_source.py b/tests/test_unseen_source.py index 49b77bb..afa868f 100644 --- a/tests/test_unseen_source.py +++ b/tests/test_unseen_source.py @@ -114,6 +114,8 @@ def test_http_adapters_with_fake_transport(): from urllib.parse import parse_qs, urlsplit class Transport: + response_headers = {} + def __init__(self): self.calls = [] @@ -121,9 +123,10 @@ def test_http_adapters_with_fake_transport(): self.calls.append((url, headers)) parsed = urlsplit(url) if parsed.path == "/advisories": - page = int(parse_qs(parsed.query)["page"][0]) + after = parse_qs(parsed.query).get("after") + self.response_headers = {} if after else {"Link": f'<{url}&after=cursor>; rel="next"'} return [ - {**advisory(), "ghsa_id": f"GHSA-test-test-{i:04d}"} for i in range(0 if page == 1 else 100, 100 if page == 1 else 101) + {**advisory(), "ghsa_id": f"GHSA-test-test-{i:04d}"} for i in range(0 if not after else 100, 100 if not after else 101) ] if "/license" in parsed.path: assert parse_qs(parsed.query)["ref"] == ["b" * 40] @@ -150,6 +153,8 @@ def test_http_adapters_with_fake_transport(): def test_http_failures_are_not_empty_results(): class Transport: + response_headers = {} + def get(self, url, headers): raise ValueError("http_failure") @@ -283,12 +288,13 @@ def test_advisory_page_filters_match_candidate_population(ecosystem): expected = {c for f in taxonomy.families if f.id in s.FAMILIES for c in f.cwes} class Transport: + response_headers = {} + def get(self, url, headers): query = parse_qs(urlsplit(url).query) assert query == { "type": ["reviewed"], "per_page": ["100"], - "page": ["2"], "sort": ["published"], "direction": ["asc"], "ecosystem": [ecosystem], @@ -297,7 +303,7 @@ def test_advisory_page_filters_match_candidate_population(ecosystem): } return [] - assert s.GitHub("fake", Transport()).advisory_page(2, ecosystem=ecosystem) == {"rows": [], "done": True} + assert s.GitHub("fake", Transport()).advisory_page(ecosystem=ecosystem) == {"rows": [], "next_url": None} # Every enumerated CWE is selectable at the inclusive date boundary. for cwe in expected: row = advisory() @@ -317,6 +323,8 @@ def test_advisory_page_omits_entire_cwe_filter_for_non_exact_taxonomy(monkeypatc monkeypatch.setattr(s, "load_families", lambda: taxonomy) class Transport: + response_headers = {} + def get(self, url, headers): query = parse_qs(urlsplit(url).query) assert "cwes" not in query @@ -324,4 +332,61 @@ def test_advisory_page_omits_entire_cwe_filter_for_non_exact_taxonomy(monkeypatc assert query["ecosystem"] == ["pip"] return [] - s.GitHub("fake", Transport()).advisory_page(1, ecosystem="pip") + s.GitHub("fake", Transport()).advisory_page(ecosystem="pip") + + +@pytest.mark.parametrize("journaled", [False, True]) +def test_cursor_walk_ignores_page_numbers_and_stops_without_next(tmp_path, journaled): + import io + from email.message import Message + from urllib.parse import parse_qs, urlencode, urlsplit + + calls = [] + + def opener(request, timeout): + query = parse_qs(urlsplit(request.full_url).query) + eco = query["ecosystem"][0] + after = query.get("after", [None])[0] + calls.append((eco, after)) + assert len(calls) <= 12 # Fail loudly if a page-number loop repeats forever. + # Like /advisories, ignore any page parameter. Only the opaque cursor advances. + offset = 100 if after == "opaque+/=" else 0 + rows = [{**advisory(), "ghsa_id": f"GHSA-test-test-{i:04d}"} for i in range(offset, offset + 100)] + response = io.BytesIO(json.dumps(rows).encode()) + response.headers = Message() + if after is None: + following = "https://api.github.com/advisories?" + urlencode({**query, "after": ["opaque+/="]}, doseq=True) + response.headers["link"] = f'; rel="prev", <{following}>; rel="next"' + else: + # A full last page with a Link header, but no next relation, is exhausted. + response.headers["link"] = '; rel="prev"' + return response + + transport = s.JSONTransport(opener=opener) + first = "https://api.github.com/advisories?ecosystem=pip" + assert transport.get(first + "&page=1", {}) == transport.get(first + "&page=100", {}) + calls.clear() + api = s.GitHub("fake", transport) + if journaled: + draft = importlib.import_module("benchmarks.unseen.draft") + with draft.Journal(tmp_path / "journal").open({"journal_version": 4}) as journal: + rows = draft.enumerate_advisories(api, journal) + else: + rows = list(api.advisories()) + assert [row["ghsa_id"] for row in rows] == [f"GHSA-test-test-{i:04d}" for i in range(200)] + assert calls == [(eco, cursor) for eco in s.ECOSYSTEMS for cursor in (None, "opaque+/=")] + + +def test_transport_replaces_last_successful_response_headers(): + import io + + responses = [io.BytesIO(b"{}"), io.BytesIO(b'{"broken":'), io.BytesIO(b"[]")] + responses[0].headers = {"Link": '; rel="next"'} + responses[1].headers = {"Link": "failed response"} + waits = [] + transport = s.JSONTransport(opener=lambda *a, **kw: responses.pop(0), sleeper=waits.append) + assert transport.get("https://example.invalid", {}) == {} + assert "cursor" in transport.response_headers["Link"] + assert transport.get("https://example.invalid", {}) == [] + assert transport.response_headers == {} + assert waits == [1]