mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Merge unseen-resolution-cache: cache enumeration + resolution to ousast-corpus
Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user