Merge unseen-resolution-cache: cache enumeration + resolution to ousast-corpus

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
Marius Ciepluch
2026-10-10 16:33:29 +02:00
co-authored by Claude Opus 4.8
5 changed files with 477 additions and 15 deletions
+86
View File
@@ -105,6 +105,92 @@ class CorpusBank:
self.client = client self.client = client
self.seed = seed self.seed = seed
self.decisions_prefix = f"{prefix}/{seed}/decisions/" self.decisions_prefix = f"{prefix}/{seed}/decisions/"
self.resolved_prefix = f"{prefix}/{seed}/resolved/"
self.advisories_key = f"{prefix}/{seed}/advisories/list.json"
self.repositories_prefix = f"{prefix}/{seed}/repositories/"
@staticmethod
def advisory_key(advisory_id: str) -> str:
return hashlib.sha256(advisory_id.encode()).hexdigest()
def _put(self, key: str, body: dict, *, labels: dict) -> None:
existing = _retry(self.client.get, key)
if existing is not None:
previous = json.loads(existing[0])
if {field: value for field, value in previous.items() if field != "ts"} == body:
return
body = {**body, "ts": datetime.now(UTC).isoformat()}
data = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
_retry(self.client.put, key, data, labels=labels)
def bank_resolution(self, advisory_id: str, result: dict, *, contract: str) -> None:
decision = "candidate" if "candidate" in result else "reject"
self._put(
self.resolved_prefix + self.advisory_key(advisory_id) + ".json",
{
"advisory": advisory_id,
"decision": decision,
"result": result,
"reason": result.get("rejected"),
"resolution_contract": contract,
"seed": self.seed,
},
labels={"decision": decision, "contract": contract},
)
def _listed(self, prefix: str, identity: str) -> dict[str, dict]:
results = {}
for key in sorted(self.client.keys(prefix)):
stored = _retry(self.client.get, key)
if stored is None:
raise ValueError("corpus_resolution_missing")
body = json.loads(stored[0])
name = body[identity]
if key != prefix + self.advisory_key(name) + ".json" or body["seed"] != self.seed:
raise ValueError("corpus_resolution_identity")
results[name] = body
return results
def resolved(self) -> dict[str, dict]:
results = self._listed(self.resolved_prefix, "advisory")
for body in results.values():
expected = "candidate" if "candidate" in body["result"] else "reject"
if body["decision"] != expected:
raise ValueError("corpus_resolution_invalid")
return results
def cache_advisories(self, rows: list, *, contract: str) -> None:
if self.cached_advisories(contract=contract) is not None:
return
self._put(
self.advisories_key,
{"advisories": rows, "advisories_contract": contract, "seed": self.seed},
labels={"contract": contract},
)
def cached_advisories(self, *, contract: str) -> list | None:
stored = _retry(self.client.get, self.advisories_key)
if stored is None:
return None
body = json.loads(stored[0])
if body["seed"] != self.seed:
raise ValueError("corpus_advisories_identity")
return body["advisories"] if body["advisories_contract"] == contract else None
def bank_repository(self, name: str, result: dict | None, *, contract: str) -> None:
"""Retain redirects, fork identity and 404s needed by the all-fix index."""
self._put(
self.repositories_prefix + self.advisory_key(name) + ".json",
{"repository": name, "result": result, "resolution_contract": contract, "seed": self.seed},
labels={"contract": contract},
)
def repositories(self, *, contract: str) -> dict[str, dict | None]:
return {
name: body["result"]
for name, body in self._listed(self.repositories_prefix, "repository").items()
if body["resolution_contract"] == contract
}
@staticmethod @staticmethod
def key_for(repository: str, fix: str) -> str: def key_for(repository: str, fix: str) -> str:
+70 -14
View File
@@ -26,7 +26,7 @@ from pathlib import Path
from . import eligibility, extract, source from . import eligibility, extract, source
from .bank import CorpusBank, CorpusObjects, corpus_settings, split_decided from .bank import CorpusBank, CorpusObjects, corpus_settings, split_decided
from .extract import EXTRACTION_CONTRACT, Git, InstrumentFailure from .extract import EXTRACTION_CONTRACT, RESOLUTION_CONTRACT, Git, InstrumentFailure
from .journal import Journal, ResolvedGitHub, ResumeMismatch, Stopped, atomic, digest, external from .journal import Journal, ResolvedGitHub, ResumeMismatch, Stopped, atomic, digest, external
@@ -209,6 +209,43 @@ def prove_source(git, revision: str) -> dict:
raise InstrumentFailure("zero_source_bytes") raise InstrumentFailure("zero_source_bytes")
class BankedGitHub(ResolvedGitHub):
"""Stream the fix index's repository metadata before continuing the draw."""
def __init__(self, github, journal, bank):
super().__init__(github, journal)
self.bank = bank
self.resolutions = bank.repositories(contract=RESOLUTION_CONTRACT)
journal.rows["resolution"].update({name: body["metadata"] if body is not None else None for name, body in self.resolutions.items()})
def resolve(self, name):
key = eligibility.normalize(name)
if key not in self.resolutions:
try:
resolved = eligibility.resolve(name, self)
except eligibility.RepositoryNotFound:
body = None
else:
body = {
"metadata": self.journal.rows["resolution"].get(key),
"full_name": resolved.full_name,
"names": sorted(resolved.names),
"network": resolved.network,
}
self.bank.bank_repository(key, body, contract=RESOLUTION_CONTRACT)
self.resolutions[key] = body
body = self.resolutions[key]
if body is None:
raise eligibility.RepositoryNotFound("not_found")
return eligibility.Resolution(body["full_name"], frozenset(body["names"]), body["network"])
def resolve_repository(name, github):
if isinstance(github, BankedGitHub):
return github.resolve(name)
return eligibility.resolve(name, github)
def advisory_index(rows: list[dict], github) -> dict[str, set[str]]: def advisory_index(rows: list[dict], github) -> dict[str, set[str]]:
"""Include ALL fixes in the enumerated population, even from rejected candidates. """Include ALL fixes in the enumerated population, even from rejected candidates.
@@ -219,7 +256,7 @@ def advisory_index(rows: list[dict], github) -> dict[str, set[str]]:
for row in rows: for row in rows:
for name, sha in source.fix_links(row): for name, sha in source.fix_links(row):
try: try:
resolved = eligibility.resolve(name, github) resolved = resolve_repository(name, github)
except eligibility.RepositoryNotFound: except eligibility.RepositoryNotFound:
index[name].add(sha) index[name].add(sha)
continue continue
@@ -305,7 +342,7 @@ def decide(row, github, osv, used, selected_networks=()):
if len(links) != 1: if len(links) != 1:
raise source.Rejected("fix_links") raise source.Rejected("fix_links")
name, _ = next(iter(links)) name, _ = next(iter(links))
resolved = eligibility.resolve(name, github) resolved = resolve_repository(name, github)
rejected_by = eligibility.rejection(resolved, used, set()) rejected_by = eligibility.rejection(resolved, used, set())
if rejected_by: if rejected_by:
raise source.Rejected("used_" + rejected_by) raise source.Rejected("used_" + rejected_by)
@@ -382,6 +419,7 @@ def ordered_extractions(
pool_size, pool_size,
bank: CorpusBank | None = None, bank: CorpusBank | None = None,
decided=None, decided=None,
stale_resolutions=(),
): ):
"""Bounded in-flight work, completion journaling, and seed-order acceptance. """Bounded in-flight work, completion journaling, and seed-order acceptance.
@@ -400,6 +438,15 @@ def ordered_extractions(
context = multiprocessing.get_context("fork") context = multiprocessing.get_context("fork")
pool = ProcessPoolExecutor(max_workers=workers, mp_context=context) pool = ProcessPoolExecutor(max_workers=workers, mp_context=context)
def resolution(row):
key = row["ghsa_id"]
if key not in checkpoint.rows["eligibility"]:
result = decide(row, api, osv, used)
if bank is not None:
bank.bank_resolution(key, result, contract=RESOLUTION_CONTRACT)
checkpoint.put("eligibility", key, result)
return checkpoint.rows["eligibility"][key]
def remember(repository, fix, result): def remember(repository, fix, result):
bank.bank(repository, fix, result, contract=EXTRACTION_CONTRACT) bank.bank(repository, fix, result, contract=EXTRACTION_CONTRACT)
# Reuse decisions made during this invocation too: another advisory for # Reuse decisions made during this invocation too: another advisory for
@@ -461,13 +508,15 @@ def ordered_extractions(
exhausted = True exhausted = True
break break
key = row["ghsa_id"] key = row["ghsa_id"]
if key in stale_resolutions:
resolution(row)
identity = None identity = None
if bank is not None: if bank is not None:
links = source.fix_links(row) links = source.fix_links(row)
if len(links) == 1: if len(links) == 1:
repository, fix = next(iter(links)) repository, fix = next(iter(links))
try: try:
resolved = eligibility.resolve(repository, api) resolved = resolve_repository(repository, api)
repository = resolved.full_name repository = resolved.full_name
except eligibility.RepositoryNotFound: except eligibility.RepositoryNotFound:
resolved = None resolved = None
@@ -508,9 +557,7 @@ def ordered_extractions(
checkpoint.put("extraction", extraction_key, result) checkpoint.put("extraction", extraction_key, result)
pending.append((row, decision, extraction_key)) pending.append((row, decision, extraction_key))
break break
if key not in checkpoint.rows["eligibility"]: decision = resolution(row)
checkpoint.put("eligibility", key, decide(row, api, osv, used))
decision = checkpoint.rows["eligibility"][key]
extraction_key, future = None, None extraction_key, future = None, None
if "candidate" in decision: if "candidate" in decision:
candidate = source.Candidate(**decision["candidate"]) candidate = source.Candidate(**decision["candidate"])
@@ -619,6 +666,7 @@ def run(
scratch = ExitStack() scratch = ExitStack()
try: try:
decided = bank.decided() if bank is not None else {} decided = bank.decided() if bank is not None else {}
banked = bank.resolved() if bank is not None else {}
accepted, reusable = split_decided(decided, contract=EXTRACTION_CONTRACT) accepted, reusable = split_decided(decided, contract=EXTRACTION_CONTRACT)
decided = {key: value for key, value in decided.items() if key in reusable} decided = {key: value for key, value in decided.items() if key in reusable}
known_empty = set(known_empty or ()) known_empty = set(known_empty or ())
@@ -638,16 +686,23 @@ def run(
) )
with checkpoint.open(inputs(used, seed, known_empty, clones, pool_size=pool_size, slices=slices)): with checkpoint.open(inputs(used, seed, known_empty, clones, pool_size=pool_size, slices=slices)):
checkpoint.boundary() checkpoint.boundary()
api = ResolvedGitHub(github, checkpoint) api = BankedGitHub(github, checkpoint, bank) if bank is not None else ResolvedGitHub(github, checkpoint)
checkpoint.rows["eligibility"].update(
{key: body["result"] for key, body in banked.items() if body["resolution_contract"] == RESOLUTION_CONTRACT}
)
used = copy.deepcopy(used) # canonicalization must not mutate the resume input used = copy.deepcopy(used) # canonicalization must not mutate the resume input
eligibility.canonicalize_used(used, api) eligibility.canonicalize_used(used, api)
report["used_set"] = used.report(known_empty) report["used_set"] = used.report(known_empty)
try: rows = bank.cached_advisories(contract=RESOLUTION_CONTRACT) if bank is not None else None
rows = enumerate_advisories(api, checkpoint) if rows is None:
except InstrumentFailure as exc: try:
if str(exc) != "empty_advisories" or not accepted: rows = enumerate_advisories(api, checkpoint)
raise except InstrumentFailure as exc:
rows = [] if str(exc) != "empty_advisories" or not accepted:
raise
rows = []
if bank is not None:
bank.cache_advisories(rows, contract=RESOLUTION_CONTRACT)
if bank is not None: if bank is not None:
# Retain accepted entries even if their advisory disappeared upstream. # Retain accepted entries even if their advisory disappeared upstream.
present = {row["ghsa_id"] for row in rows} present = {row["ghsa_id"] for row in rows}
@@ -680,6 +735,7 @@ def run(
pool_size, pool_size,
bank=bank, bank=bank,
decided=decided, decided=decided,
stale_resolutions={key for key, body in banked.items() if body["resolution_contract"] != RESOLUTION_CONTRACT},
) as outcomes: ) as outcomes:
for row, decision, result in outcomes: for row, decision, result in outcomes:
counts[source.ecosystem(row)] += 1 counts[source.ecosystem(row)] += 1
+2
View File
@@ -21,6 +21,8 @@ from .source import Candidate, Rejected
# Bump only when accept/reject outcomes or accepted entry bytes change. # Bump only when accept/reject outcomes or accepted entry bytes change.
EXTRACTION_CONTRACT = "v1" EXTRACTION_CONTRACT = "v1"
# Bump only when enumeration, eligibility or source.candidate outputs change.
RESOLUTION_CONTRACT = "v1"
LANGUAGES = {".py", ".js", ".jsx", ".mjs", ".cjs", ".ts", ".tsx", ".php", ".java"} LANGUAGES = {".py", ".js", ".jsx", ".mjs", ".cjs", ".ts", ".tsx", ".php", ".java"}
HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@", re.M) HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@", re.M)
+101
View File
@@ -214,3 +214,104 @@ def test_import_does_not_require_boto3(monkeypatch):
module = importlib.util.module_from_spec(spec) module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module) spec.loader.exec_module(module)
assert module.CorpusBank.key_for("a", "b") == CorpusBank.key_for("a", "b") assert module.CorpusBank.key_for("a", "b") == CorpusBank.key_for("a", "b")
@pytest.mark.parametrize(
"result",
[
{"candidate": {"repository": "org/é"}, "network": "org/root"},
{"rejected": "used_pairs"},
{"rejected": "family", "network": "org/root"},
],
)
def test_resolution_roundtrip_idempotent_and_contract_replacement(result):
objects = FakeObjects()
bank = CorpusBank(objects, seed=42, prefix="custom")
advisory = "GHSA-test/advisory"
key = f"custom/42/resolved/{hashlib.sha256(advisory.encode()).hexdigest()}.json"
assert bank.advisory_key(advisory) == hashlib.sha256(advisory.encode()).hexdigest()
bank.bank_resolution(advisory, result, contract="old")
first = objects.objects.copy()
bank.bank_resolution(advisory, result, contract="old")
assert objects.objects == first and objects.calls.count("put") == 1
row = bank.resolved()[advisory]
assert row == {
"advisory": advisory,
"result": result,
"decision": "candidate" if "candidate" in result else "reject",
"reason": result.get("rejected"),
"resolution_contract": "old",
"seed": 42,
"ts": row["ts"],
}
assert list(objects.objects) == [key]
assert datetime.fromisoformat(row["ts"]).utcoffset().total_seconds() == 0
bank.bank_resolution(advisory, result, contract="v1")
assert bank.resolved()[advisory]["resolution_contract"] == "v1"
assert objects.calls.count("put") == 2
def test_advisory_snapshot_written_once_per_contract():
objects = FakeObjects()
bank = CorpusBank(objects, seed=7)
rows = [{"ghsa_id": "GHSA-é", "references": ["a", "b"]}]
assert bank.cached_advisories(contract="v1") is None
bank.cache_advisories(rows, contract="v1")
original = objects.objects.copy()
assert bank.cached_advisories(contract="v1") == rows
assert bank.cached_advisories(contract="v2") is None
bank.cache_advisories([], contract="v1")
assert objects.objects == original and objects.calls.count("put") == 1
body = json.loads(objects.objects["unseen/7/advisories/list.json"])
assert body == {"advisories": rows, "advisories_contract": "v1", "seed": 7, "ts": body["ts"]}
bank.cache_advisories([], contract="v2")
assert bank.cached_advisories(contract="v2") == []
assert bank.cached_advisories(contract="v1") is None
@pytest.mark.parametrize("method", ["resolution", "advisories", "repository"])
@pytest.mark.parametrize("operation", ["get", "put"])
@pytest.mark.parametrize("failures", [1, 2])
def test_new_bank_writes_retry_once(method, operation, failures):
objects = FakeObjects()
bank = CorpusBank(objects, seed=7)
objects.failures[operation] = failures
def write():
if method == "resolution":
bank.bank_resolution("GHSA-a", {"rejected": "family"}, contract="v1")
elif method == "advisories":
bank.cache_advisories([], contract="v1")
else:
bank.bank_repository("org/repo", None, contract="v1")
if failures == 2:
with pytest.raises(OSError):
write()
assert objects.calls.count(operation) == 2
else:
write()
assert len(objects.objects) == 1
assert objects.failures[operation] == 0
@pytest.mark.parametrize("method", ["resolved", "cached_advisories", "repositories"])
@pytest.mark.parametrize("failures", [1, 2])
def test_new_bank_reads_retry_once(method, failures):
objects = FakeObjects()
bank = CorpusBank(objects, seed=7)
bank.bank_resolution("GHSA-a", {"rejected": "family"}, contract="v1")
bank.cache_advisories([], contract="v1")
bank.bank_repository("org/old", {"full_name": "org/new"}, contract="v1")
bank.bank_repository("org/missing", None, contract="old")
objects.failures["get"] = failures
read = getattr(bank, method)
kwargs = {} if method == "resolved" else {"contract": "v1"}
if failures == 2:
with pytest.raises(OSError):
read(**kwargs)
else:
result = read(**kwargs)
assert result is not None
if method == "repositories":
assert result == {"org/old": {"full_name": "org/new"}}
+218 -1
View File
@@ -1651,6 +1651,8 @@ def test_bank_crash_resume_and_durability_failure(tmp_path, monkeypatch, failure
puts = 0 puts = 0
def put(self, key, data, labels): def put(self, key, data, labels):
if "/decisions/" not in key:
return super().put(key, data, labels)
self.puts += 1 self.puts += 1
if failure == "persistent_put" and not self.interrupted: if failure == "persistent_put" and not self.interrupted:
raise OSError("offline simulated bank outage") raise OSError("offline simulated bank outage")
@@ -1687,7 +1689,7 @@ def test_bank_crash_resume_and_durability_failure(tmp_path, monkeypatch, failure
report = run(tmp_path / "resume", bank) report = run(tmp_path / "resume", bank)
assert report["status"] == "instrument_failure" assert report["status"] == "instrument_failure"
assert objects.puts == 2 assert objects.puts == 2
assert not objects.objects assert not bank.decided()
assert not (tmp_path / "resume" / "draft-p1.toml").exists() assert not (tmp_path / "resume" / "draft-p1.toml").exists()
objects.interrupted = True objects.interrupted = True
resumed = run(tmp_path / "resume", bank) resumed = run(tmp_path / "resume", bank)
@@ -1817,3 +1819,218 @@ def test_bank_duplicate_advisory_cannot_overwrite_pair_decision(tmp_path, monkey
assert all(report["status"] == "draft" for report in reports) assert all(report["status"] == "draft" for report in reports)
assert reports[0]["manifest_sha256"] == reports[1]["manifest_sha256"] assert reports[0]["manifest_sha256"] == reports[1]["manifest_sha256"]
assert bank.decided()[bank.key_for(repository, fix)]["decision"] == "accept" assert bank.decided()[bank.key_for(repository, fix)]["decision"] == "accept"
@pytest.mark.parametrize("workers", [1, 4])
def test_bank_enumeration_resolution_resume_and_contracts(tmp_path, monkeypatch, workers):
from benchmarks.unseen.bank import CorpusBank
monkeypatch.setattr(importlib.import_module("benchmarks.unseen.journal").os, "fsync", lambda fd: None)
rows = s.ordered_advisories(list(FakeAPI(8).advisories()), 7)
rows[0]["cwes"] = [] # Bank a rejected advisory as well as candidates.
names = [next(iter(s.fix_links(row)))[0] for row in rows]
changes = entries()[0]["changes"]
monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (copy.deepcopy(changes), {}, {"bytes": 12, "files": 1}))
calls = Counter()
original_enumerate, original_decide, original_candidate = d.enumerate_advisories, d.decide, s.candidate
original_resolve = e.resolve
def resolve(*args):
calls["resolve"] += 1
return original_resolve(*args)
monkeypatch.setattr(e, "resolve", resolve)
class API(FakeAPI):
def advisories(self):
return iter(copy.deepcopy(rows))
def get_repo(self, name):
calls["get_repo"] += 1
result = super().get_repo(name)
if name == names[2]:
result["source"] = {"full_name": names[1]}
return result
def enumerate_rows(*args):
calls["enumerate"] += 1
return original_enumerate(*args)
def decide(row, *args):
calls[row["ghsa_id"]] += 1
return original_decide(row, *args)
def candidate(*args):
calls["candidate"] += 1
return original_candidate(*args)
monkeypatch.setattr(d, "enumerate_advisories", enumerate_rows)
monkeypatch.setattr(d, "decide", decide)
monkeypatch.setattr(s, "candidate", candidate)
class Objects(CorpusFakeObjects):
def __init__(self):
super().__init__()
self.puts = Counter()
def put(self, key, data, labels):
self.puts[key] += 1
super().put(key, data, labels)
objects = Objects()
bank = CorpusBank(objects, seed=7)
def run(name, bank):
used = e.UsedSet()
return d.run(
tmp_path / "repo",
github=API(),
osv=FakeOSV(),
clones=FakeClones(),
used=used,
seed=7,
known_empty=set(used.sources),
journal_path=tmp_path / name,
extract_workers=workers,
pool_size=6,
slices=1,
bank=bank,
)
full = run("full", None)
assert full["status"] == "draft", full
calls.clear()
first = run("first", bank)
assert first["manifest_sha256"] == full["manifest_sha256"]
assert calls["enumerate"] == 1 and calls["candidate"] == 8
resolved = bank.resolved()
assert len(resolved) == 8
assert {body["decision"] for body in resolved.values()} == {"candidate", "reject"}
assert all(objects.puts[key] == 1 for key in objects.objects if "/resolved/" in key or "/advisories/" in key)
assert all(body["result"] == original_decide(row, API(), FakeOSV(), e.UsedSet()) for row in rows for body in [resolved[row["ghsa_id"]]])
calls.clear()
second = run("second", bank)
assert not calls # No enumeration, decide, candidate, or upstream repository lookup.
assert second["manifest_sha256"] == first["manifest_sha256"]
for filename in ("draft-p1.toml", "private-draft-p1.toml"):
assert (tmp_path / "second" / filename).read_bytes() == (tmp_path / "full" / filename).read_bytes()
# A restart with resolution complete but extraction unfinished also uses
# the exact cached candidates, without re-running any prefix operation.
for key in list(objects.objects):
if "/decisions/" in key:
del objects.objects[key]
remaining = run("remaining-extraction", bank)
assert remaining["manifest_sha256"] == full["manifest_sha256"]
assert not calls
stale_id = rows[1]["ghsa_id"]
bank.bank_resolution(stale_id, {"rejected": "old"}, contract="old")
stale = run("stale-resolution", bank)
assert stale["manifest_sha256"] == full["manifest_sha256"]
assert calls == {stale_id: 1, "candidate": 1}
assert bank.resolved()[stale_id]["resolution_contract"] == d.RESOLUTION_CONTRACT
bank.cache_advisories([{"ghsa_id": "obsolete"}], contract="old")
calls.clear()
renewed = run("stale-advisories", bank)
assert renewed["manifest_sha256"] == full["manifest_sha256"]
assert calls == {"enumerate": 1}
assert s.ordered_advisories(bank.cached_advisories(contract=d.RESOLUTION_CONTRACT), 7) == s.ordered_advisories(
[s.reduce_advisory(row) for row in rows], 7
)
@pytest.mark.parametrize("stage", ["advisories", "repositories", "resolved"])
@pytest.mark.parametrize("failure", ["interrupt", "put_failure"])
def test_prefix_streaming_failure_and_restart(tmp_path, monkeypatch, stage, failure):
from benchmarks.unseen.bank import CorpusBank
monkeypatch.setattr(importlib.import_module("benchmarks.unseen.journal").os, "fsync", lambda fd: None)
changes = entries()[0]["changes"]
monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (copy.deepcopy(changes), {}, {"bytes": 1, "files": 1}))
calls = Counter()
original = d.decide
def decide(row, *args):
calls[row["ghsa_id"]] += 1
return original(row, *args)
monkeypatch.setattr(d, "decide", decide)
class Objects(CorpusFakeObjects):
armed = True
attempts = 0
def put(self, key, data, labels):
if self.armed and f"/{stage}/" in key:
self.attempts += 1
if failure == "put_failure":
raise OSError("offline outage")
super().put(key, data, labels)
raise KeyboardInterrupt
super().put(key, data, labels)
objects = Objects()
bank = CorpusBank(objects, seed=7)
def run(name, bank):
used = e.UsedSet()
return d.run(
tmp_path / "repo",
github=FakeAPI(4),
osv=FakeOSV(),
clones=FakeClones(),
used=used,
seed=7,
known_empty=set(used.sources),
journal_path=tmp_path / name,
pool_size=4,
slices=1,
extract_workers=1,
bank=bank,
)
if failure == "interrupt":
with pytest.raises(KeyboardInterrupt):
run("interrupted", bank)
assert objects.attempts == 1
else:
assert run("interrupted", bank)["status"] == "instrument_failure"
assert objects.attempts == 2
saved = bank.resolved()
objects.armed = False
calls.clear()
resumed = run("resumed", bank)
assert resumed["status"] == "draft", resumed
assert all(calls[key] == 0 for key in saved)
assert len(bank.resolved()) == 4
baseline = run("baseline", None)
assert resumed["manifest_sha256"] == baseline["manifest_sha256"]
def test_banked_fix_index_preserves_renames_multiple_links_and_404s(tmp_path, monkeypatch):
from benchmarks.unseen.bank import CorpusBank
rows = list(FakeAPI(3).advisories())
rows[0]["references"] = ["https://github.com/fixture/old/commit/" + "a" * 40]
rows[1]["references"] = ["https://github.com/fixture/missing/commit/" + "b" * 40]
rows[2]["references"] += ["https://github.com/fixture/old/commit/" + "c" * 40]
class API(FakeAPI):
def get_repo(self, name):
if name == "fixture/missing":
raise e.RepositoryNotFound("not_found")
return super().get_repo(name)
bank = CorpusBank(CorpusFakeObjects(), seed=7)
with d.Journal(tmp_path / "fresh").open({}) as checkpoint:
fresh = d.advisory_index(rows, d.BankedGitHub(API(), checkpoint, bank))
assert fresh == {"fixture/project000": {"a" * 40, "c" * 40}, "fixture/missing": {"b" * 40}, "fixture/project002": {"a" * 40}}
def unexpected(*args):
pytest.fail("banked fix index must not repeat resolution")
monkeypatch.setattr(e, "resolve", unexpected)
monkeypatch.setattr(API, "get_repo", unexpected)
with d.Journal(tmp_path / "resumed").open({}) as checkpoint:
assert d.advisory_index(rows, d.BankedGitHub(API(), checkpoint, bank)) == fresh