mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
unseen-eval group 2 (2.1-2.4): advisory sourcing, vulnerability-introducing ranges, ordinary-commit draw, draft manifest and slices; journal with resume
Implemented with Codex. Sourcing run (2.5) next, by the coordinator. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
8213d3baad
commit
48ae918dec
@@ -0,0 +1,527 @@
|
||||
"""Draft and coordinator tests contain synthetic identities only and use injected clients."""
|
||||
|
||||
import copy
|
||||
import importlib
|
||||
import sys
|
||||
import tomllib
|
||||
from collections import Counter
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
d = importlib.import_module("benchmarks.unseen.draft")
|
||||
s = importlib.import_module("benchmarks.unseen.source")
|
||||
e = importlib.import_module("benchmarks.unseen.eligibility")
|
||||
|
||||
|
||||
def entries():
|
||||
rows = []
|
||||
for i in range(300):
|
||||
changes = [{"kind": "vulnerability", "method": "introducing", "base": "a" * 40, "head": "b" * 40}]
|
||||
changes += [{"kind": "ordinary", "method": "first_parent", "base": f"{j:040x}", "head": f"{j + 20:040x}"} for j in range(12)]
|
||||
rows.append(
|
||||
{
|
||||
"repository": f"fixture/project{i:03d}",
|
||||
"url": f"https://github.com/fixture/project{i:03d}",
|
||||
"ecosystem": s.ECOSYSTEMS[i % 4],
|
||||
"post_cutoff": i % 7 == 0,
|
||||
"license_spdx": "MIT" if i % 2 else "GPL-3.0",
|
||||
"license_class": "permissive" if i % 2 else "private",
|
||||
"license_ref": "b" * 40,
|
||||
"changes": changes,
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
def test_draft_split_digest_and_balanced_slices(tmp_path):
|
||||
rows = entries()
|
||||
assigned = d.assign_slices(rows, seed=19)
|
||||
assert Counter(r["slice"] for r in assigned) == {1: 100, 2: 100, 3: 100}
|
||||
assert assigned == d.assign_slices(list(reversed(rows)), seed=19)
|
||||
strata = Counter((r["ecosystem"], r["post_cutoff"]) for r in assigned)
|
||||
for slice_id in (1, 2, 3):
|
||||
counts = Counter((r["ecosystem"], r["post_cutoff"]) for r in assigned if r["slice"] == slice_id)
|
||||
assert all(abs(counts[key] - total / 3) < 1 for key, total in strata.items())
|
||||
public, private = d.write_draft(tmp_path, assigned, seed=19)
|
||||
public_text = public.read_text()
|
||||
shared = tomllib.loads(public_text)
|
||||
hidden = tomllib.loads(private.read_text())
|
||||
assert len(shared["repository"]) == 300 and len(hidden["repository"]) == 150
|
||||
assert sum(len(r["changes"]) for r in shared["repository"] if "changes" in r) == 1950
|
||||
for row in hidden["repository"]:
|
||||
assert row["url"] not in public_text and row["repository"] not in public_text
|
||||
pointer = next(p for p in shared["repository"] if p["id"] == row["id"])
|
||||
assert set(pointer) == {"id", "sha256", "slice", "private"}
|
||||
assert pointer["sha256"] == d.private_digest(row)
|
||||
changed = copy.deepcopy(row)
|
||||
changed["changes"][0]["head"] = "c" * 40
|
||||
assert d.private_digest(changed) != pointer["sha256"]
|
||||
assert private.is_relative_to(tmp_path / "benchmarks/unseen/private")
|
||||
with pytest.raises(FileExistsError):
|
||||
d.write_draft(tmp_path, assigned, seed=19)
|
||||
|
||||
|
||||
def test_draft_requires_full_size_and_valid_changes():
|
||||
with pytest.raises(ValueError, match="pool_size"):
|
||||
d.assign_slices(entries()[:299], seed=1)
|
||||
rows = entries()
|
||||
rows[0]["changes"].pop()
|
||||
with pytest.raises(ValueError, match="change_count"):
|
||||
d.assign_slices(rows, seed=1)
|
||||
|
||||
|
||||
def test_cache_is_outside_repository_deleted_and_partial(tmp_path):
|
||||
import subprocess
|
||||
|
||||
root, cache = tmp_path / "repo", tmp_path / "cache"
|
||||
root.mkdir()
|
||||
with pytest.raises(ValueError, match="cache_inside_repository"):
|
||||
d.Clones(root, root / "cache")
|
||||
calls = []
|
||||
|
||||
def runner(argv, **kwargs):
|
||||
calls.append(argv)
|
||||
if "clone" in argv:
|
||||
target = Path(argv[-1])
|
||||
(target / ".git").mkdir()
|
||||
(target / ".git/pack").write_bytes(b"123")
|
||||
return subprocess.CompletedProcess(argv, 0, b"", b"")
|
||||
|
||||
clones = d.Clones(root, cache, runner=runner)
|
||||
with clones.open("https://github.com/fixture/project") as git:
|
||||
assert git.path.exists()
|
||||
clone_path = git.path
|
||||
assert not clone_path.exists()
|
||||
assert "--filter=blob:none" in calls[0] and "--no-checkout" in calls[0]
|
||||
assert clones.counts["clones"] == 1 and clones.counts["clone_bytes"] == 3
|
||||
|
||||
|
||||
def test_clone_failure_sanitized_and_cleaned(tmp_path, capsys):
|
||||
import subprocess
|
||||
|
||||
def runner(argv, **kwargs):
|
||||
return subprocess.CompletedProcess(argv, 128, b"", b"private identity")
|
||||
|
||||
clones = d.Clones(tmp_path / "repo", tmp_path / "cache", runner=runner)
|
||||
with pytest.raises(d.InstrumentFailure), clones.open("https://github.com/fixture/project"):
|
||||
pass
|
||||
assert list((tmp_path / "cache").iterdir()) == []
|
||||
assert not capsys.readouterr().out
|
||||
|
||||
|
||||
class FakeAPI:
|
||||
calls = 0
|
||||
|
||||
def __init__(self, count=300):
|
||||
self.count = count
|
||||
self.licenses = []
|
||||
|
||||
def get_repo(self, name):
|
||||
self.calls += 1
|
||||
if name == "fixture/old":
|
||||
name = "fixture/project000"
|
||||
repo = {"full_name": name, "private": False, "size": 2, "clone_url": "https://github.com/" + name}
|
||||
if name == "fixture/fork":
|
||||
repo["source"] = {"full_name": "fixture/project001"}
|
||||
return repo
|
||||
|
||||
def get_commit(self, name, sha):
|
||||
self.calls += 1
|
||||
return {"sha": sha, "parents": [{"sha": "b" * 40}]}
|
||||
|
||||
def get_license(self, name, ref):
|
||||
self.licenses.append((name, ref))
|
||||
self.calls += 1
|
||||
# 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):
|
||||
self.calls += 1
|
||||
rows = list(self.advisories())[(page - 1) * 100 : page * 100]
|
||||
return {"rows": rows, "done": len(rows) < 100}
|
||||
|
||||
def advisories(self):
|
||||
for i in range(self.count):
|
||||
yield {
|
||||
"ghsa_id": f"GHSA-test-test-{i:04d}",
|
||||
"published_at": "2026-06-02T00:00:00Z",
|
||||
"cwes": [{"cwe_id": "CWE-89"}],
|
||||
"vulnerabilities": [{"package": {"ecosystem": "pip", "name": "fixture"}}],
|
||||
"references": [f"https://github.com/fixture/project{i:03d}/commit/" + "a" * 40],
|
||||
}
|
||||
|
||||
|
||||
class FakeOSV:
|
||||
calls = 0
|
||||
|
||||
def get(self, identifier):
|
||||
self.calls += 1
|
||||
return {}
|
||||
|
||||
|
||||
class FakeClones:
|
||||
def __init__(self):
|
||||
self.counts = Counter()
|
||||
self.commands = []
|
||||
|
||||
@contextmanager
|
||||
def open(self, url):
|
||||
self.counts["clones"] += 1
|
||||
yield self
|
||||
|
||||
def run(self, *args):
|
||||
self.commands.append(args)
|
||||
return ""
|
||||
|
||||
|
||||
def test_coordinator_end_to_end_fakes_pinned_license_and_no_names(tmp_path, monkeypatch, capsys):
|
||||
rows = entries()
|
||||
changes = copy.deepcopy(rows[0]["changes"])
|
||||
changes[0]["head"] = "c" * 40
|
||||
monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (changes, {"security": 4, "eligible": 40}, {"bytes": 12, "files": 1}))
|
||||
used, api, clones = e.UsedSet(), FakeAPI(), FakeClones()
|
||||
report = d.run(tmp_path, github=api, osv=FakeOSV(), clones=clones, used=used, seed=7, known_empty=set(used.sources))
|
||||
assert report["status"] == "draft" and report["changes"] == 3900
|
||||
assert report["license_split"] == {"private": 300}
|
||||
assert report["ordinary_exclusions"]["security"] == 1200
|
||||
assert len([ref for _, ref in api.licenses if ref == "c" * 40]) == 300
|
||||
public = (tmp_path.parent / (tmp_path.name + "-cache") / "journal/draft-p1.toml").read_text()
|
||||
assert "fixture/" not in public
|
||||
import json
|
||||
|
||||
assert "fixture/" not in json.dumps(report)
|
||||
assert capsys.readouterr().out == ""
|
||||
|
||||
|
||||
def test_used_rename_and_fork_resolution_and_shortfall(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (entries()[0]["changes"], {}, {"bytes": 1, "files": 1}))
|
||||
used, api, clones = e.UsedSet(), FakeAPI(2), FakeClones()
|
||||
used.add("pairs", {"fixture/old"}, 10)
|
||||
# project000 resolves from used alias; project001 is reached via used fork's source.
|
||||
used.add("pairs", {"fixture/fork"}, 10)
|
||||
report = d.run(tmp_path, github=api, osv=FakeOSV(), clones=clones, used=used, seed=7, known_empty=set(used.sources) - {"pairs"})
|
||||
assert report["status"] == "insufficient_repositories"
|
||||
assert report["rejections"] == {"used_pairs": 2}
|
||||
assert report["repositories"] == 0 and not clones.counts
|
||||
assert not (tmp_path / "benchmarks/unseen/draft-p1.toml").exists()
|
||||
|
||||
|
||||
def test_zero_source_is_instrument_failure_not_rejection(tmp_path, monkeypatch):
|
||||
def unread(*args, **kwargs):
|
||||
raise d.InstrumentFailure("zero_source_bytes")
|
||||
|
||||
monkeypatch.setattr(d, "repository_changes", unread)
|
||||
used = e.UsedSet()
|
||||
report = d.run(tmp_path, github=FakeAPI(1), osv=FakeOSV(), clones=FakeClones(), used=used, seed=1, known_empty=set(used.sources))
|
||||
assert report["status"] == "instrument_failure" and not report["rejections"]
|
||||
|
||||
|
||||
def test_cli_uses_environment_and_injected_stores(tmp_path, monkeypatch, capsys):
|
||||
from openultrasast import config
|
||||
from openultrasast.plane import memory
|
||||
|
||||
seen = []
|
||||
monkeypatch.setenv("GH_TOKEN", "fake-secret")
|
||||
monkeypatch.setattr(config, "load_dotenv", lambda path: None)
|
||||
monkeypatch.setattr(memory, "open_store", lambda value, **kw: seen.append((value, kw)) or object())
|
||||
monkeypatch.setattr(d.eligibility, "build_used", lambda *a, **kw: e.UsedSet())
|
||||
monkeypatch.setattr(d.source, "GitHub", lambda token: seen.append(token) or FakeAPI(0))
|
||||
monkeypatch.setattr(d.source, "OSV", FakeOSV)
|
||||
args = ["--root", str(tmp_path), "--cache", str(tmp_path.parent / "clone-cache")]
|
||||
for label in e.SOURCES:
|
||||
args += ["--known-empty", label]
|
||||
assert d.main(args) == 1
|
||||
output = capsys.readouterr().out
|
||||
assert "instrument_failure" in output and "fake-secret" not in output and "fixture/" not in output
|
||||
assert seen[-1] == "fake-secret"
|
||||
assert all(kwargs == {"read_only": True} for _, kwargs in seen[:-1])
|
||||
|
||||
|
||||
def test_repository_changes_excludes_all_advisory_sites(monkeypatch):
|
||||
calls = []
|
||||
|
||||
class FakeGit:
|
||||
def run(self, *args):
|
||||
calls.append(args)
|
||||
if args[0] == "ls-tree":
|
||||
return "app.py\n"
|
||||
return args[-1] if args[0] == "rev-parse" else ""
|
||||
|
||||
def blob(self, revision, path):
|
||||
return "def handler():\n return 1\n"
|
||||
|
||||
def parents(self, fix):
|
||||
return ["b" * 40]
|
||||
|
||||
candidate = s.Candidate(
|
||||
"fixture/project",
|
||||
"https://github.com/fixture/project",
|
||||
"a" * 40,
|
||||
"b" * 40,
|
||||
"GHSA-test-test-test",
|
||||
"pip",
|
||||
"injection",
|
||||
True,
|
||||
None,
|
||||
"MIT",
|
||||
"permissive",
|
||||
"b" * 40,
|
||||
"LICENSE",
|
||||
)
|
||||
monkeypatch.setattr(d.extract, "introducing", lambda *a: {"head": "c" * 40})
|
||||
monkeypatch.setattr(d.extract, "fix_sites", lambda git, fix, parent, **kw: [{"path": "app.py", "function": fix}])
|
||||
monkeypatch.setattr(d.extract, "ordinary_history", lambda *a: [])
|
||||
|
||||
def draw(rows, fixes, known, **kwargs):
|
||||
assert fixes == {"a" * 40, "d" * 40}
|
||||
assert known == {("app.py", "a" * 40), ("app.py", "d" * 40)}
|
||||
return [], {}
|
||||
|
||||
monkeypatch.setattr(d.extract, "draw_ordinary", draw)
|
||||
_, _, proof = d.repository_changes(FakeGit(), candidate, {"d" * 40}, seed=1)
|
||||
assert proof["bytes"] > 0
|
||||
assert len([c for c in calls if c[0] == "fetch"]) == 2
|
||||
|
||||
|
||||
def test_candidate_404_is_counted_but_service_failure_aborts(tmp_path):
|
||||
class Missing(FakeAPI):
|
||||
def get_commit(self, name, sha):
|
||||
raise e.RepositoryNotFound("not_found")
|
||||
|
||||
class Broken(FakeAPI):
|
||||
def get_commit(self, name, sha):
|
||||
raise ValueError("private URL must never escape")
|
||||
|
||||
for api, status in [(Missing(1), "insufficient_repositories"), (Broken(1), "instrument_failure")]:
|
||||
used = e.UsedSet()
|
||||
report = d.run(
|
||||
tmp_path,
|
||||
github=api,
|
||||
osv=FakeOSV(),
|
||||
clones=FakeClones(),
|
||||
used=used,
|
||||
seed=1,
|
||||
known_empty=set(used.sources),
|
||||
journal_path=tmp_path.parent / (tmp_path.name + type(api).__name__) / "journal",
|
||||
)
|
||||
assert report["status"] == status
|
||||
if status == "insufficient_repositories":
|
||||
assert report["rejections"] == {"repository_or_commit_not_found": 1}
|
||||
assert "private URL" not in str(report)
|
||||
|
||||
|
||||
def test_shortfall_keeps_ordinary_exclusion_counts(tmp_path, monkeypatch):
|
||||
def too_few(*args, **kwargs):
|
||||
exc = s.Rejected("ordinary_floor")
|
||||
exc.counts = {"eligible": 39, "size_cap": 10}
|
||||
raise exc
|
||||
|
||||
monkeypatch.setattr(d, "repository_changes", too_few)
|
||||
used = e.UsedSet()
|
||||
report = d.run(tmp_path, github=FakeAPI(1), osv=FakeOSV(), clones=FakeClones(), used=used, seed=1, known_empty=set(used.sources))
|
||||
assert report["ordinary_exclusions"] == {"eligible": 39, "size_cap": 10}
|
||||
|
||||
|
||||
def test_clone_disk_cap_cleans_up(tmp_path):
|
||||
import subprocess
|
||||
|
||||
def runner(argv, **kwargs):
|
||||
(Path(argv[-1]) / "big").write_bytes(b"12345")
|
||||
return subprocess.CompletedProcess(argv, 0, b"", b"")
|
||||
|
||||
clones = d.Clones(tmp_path / "repo", tmp_path / "cache", runner=runner, limit_bytes=4)
|
||||
with pytest.raises(d.InstrumentFailure, match="clone_size_cap"), clones.open("https://github.com/fixture/project"):
|
||||
pass
|
||||
assert list((tmp_path / "cache").iterdir()) == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("stage,n", [("advisories", 2), ("resolution", 103), ("eligibility", 17), ("extraction", 23), ("completion", 1)])
|
||||
def test_checkpoint_interrupt_resume_matches_uninterrupted(tmp_path, monkeypatch, stage, n):
|
||||
import json
|
||||
|
||||
journal_module = importlib.import_module("benchmarks.unseen.journal")
|
||||
# Large fake populations exercise replay; durability is checked separately below.
|
||||
monkeypatch.setattr(journal_module.os, "fsync", lambda fd: None)
|
||||
root = tmp_path / "repo"
|
||||
full, interrupted = tmp_path / "full/journal", tmp_path / "interrupted/journal"
|
||||
changes = entries()[0]["changes"]
|
||||
monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (changes, {"eligible": 40}, {"bytes": 12, "files": 1}))
|
||||
used = e.UsedSet()
|
||||
|
||||
def run(path, api=None, clones=None):
|
||||
return d.run(
|
||||
root,
|
||||
github=api or FakeAPI(),
|
||||
osv=FakeOSV(),
|
||||
clones=clones or FakeClones(),
|
||||
used=used,
|
||||
seed=7,
|
||||
known_empty=set(used.sources),
|
||||
journal_path=path,
|
||||
)
|
||||
|
||||
assert run(full)["status"] == "draft"
|
||||
original = d.Journal.put
|
||||
|
||||
def interrupt(self, current, key, value):
|
||||
original(self, current, key, value)
|
||||
if current == stage and len(self.rows[current]) == n:
|
||||
raise KeyboardInterrupt
|
||||
|
||||
monkeypatch.setattr(d.Journal, "put", interrupt)
|
||||
with pytest.raises(KeyboardInterrupt):
|
||||
run(interrupted)
|
||||
assert not (interrupted / "draft-p1.toml").exists()
|
||||
before = {p.name: p.read_bytes() for p in interrupted.glob("*.jsonl")}
|
||||
progress = json.loads((interrupted / "progress.json").read_text())
|
||||
assert progress["counts"][stage] == n
|
||||
assert progress["last_unit"]["stage"] == stage
|
||||
assert progress["elapsed_seconds"] >= 0
|
||||
assert "rate_limit" in progress
|
||||
monkeypatch.setattr(d.Journal, "put", original)
|
||||
clones = FakeClones()
|
||||
assert run(interrupted, clones=clones)["status"] == "draft"
|
||||
assert clones.counts["clones"] == 300 - progress["counts"]["extraction"]
|
||||
for name in ("draft-p1.toml", "private-draft-p1.toml"):
|
||||
assert (full / name).read_bytes() == (interrupted / name).read_bytes()
|
||||
assert all((interrupted / name).read_bytes().startswith(data) for name, data in before.items())
|
||||
assert not root.exists() # no names, manifests, or other records in the checkout
|
||||
api, clones = FakeAPI(), FakeClones()
|
||||
assert run(interrupted, api, clones)["status"] == "draft"
|
||||
assert api.calls == 0 and clones.counts == {}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("parameter", ["seed", "used_set_digest", "clone_limit_bytes", "design_extract"])
|
||||
def test_resume_refuses_changed_inputs_before_network(tmp_path, monkeypatch, parameter):
|
||||
path, root = tmp_path / "cache/journal", tmp_path / "repo"
|
||||
used = e.UsedSet()
|
||||
kwargs = dict(
|
||||
root=root,
|
||||
github=FakeAPI(),
|
||||
osv=FakeOSV(),
|
||||
clones=FakeClones(),
|
||||
used=used,
|
||||
seed=7,
|
||||
known_empty=set(used.sources),
|
||||
journal_path=path,
|
||||
max_hours=0,
|
||||
)
|
||||
assert d.run(**kwargs)["status"] == "stopped"
|
||||
if parameter == "seed":
|
||||
kwargs["seed"] = 8
|
||||
elif parameter == "used_set_digest":
|
||||
used.add("pairs", {"fixture/new"}, 12)
|
||||
elif parameter == "clone_limit_bytes":
|
||||
kwargs["clones"].limit_bytes = 12
|
||||
else:
|
||||
original = d.inputs
|
||||
monkeypatch.setattr(d, "inputs", lambda *a: {**original(*a), "design_extract": "changed"})
|
||||
report = d.run(**kwargs)
|
||||
assert report["status"] == "resume_refused"
|
||||
assert report["changed_parameters"] == [parameter]
|
||||
assert kwargs["github"].calls == 0
|
||||
|
||||
|
||||
def test_stop_between_units_and_progress_rate_state(tmp_path, monkeypatch):
|
||||
import json
|
||||
from types import SimpleNamespace
|
||||
|
||||
path, root = tmp_path / "cache/journal", tmp_path / "repo"
|
||||
api = FakeAPI(1)
|
||||
api.transport = SimpleNamespace(rate_limit={"remaining": "27", "reset": "1234"}, attempts=0)
|
||||
used = e.UsedSet()
|
||||
original = d.Journal.put
|
||||
|
||||
def stop(self, stage, key, value):
|
||||
original(self, stage, key, value)
|
||||
if stage == "eligibility":
|
||||
(self.path / "STOP").touch()
|
||||
|
||||
monkeypatch.setattr(d.Journal, "put", stop)
|
||||
clones = FakeClones()
|
||||
report = d.run(root, github=api, osv=FakeOSV(), clones=clones, used=used, seed=1, known_empty=set(used.sources), journal_path=path)
|
||||
assert report["status"] == "stopped" and report["reason"] == "STOP"
|
||||
assert clones.counts == {}
|
||||
progress = json.loads((path / "progress.json").read_text())
|
||||
assert progress["status"] == "stopped" and progress["counts"]["eligibility"] == 1
|
||||
assert progress["rate_limit"] == api.transport.rate_limit
|
||||
assert not (path / "draft-p1.toml").exists()
|
||||
|
||||
|
||||
def test_max_hours_stops_after_completed_page(tmp_path):
|
||||
import json
|
||||
|
||||
now = [0.0]
|
||||
|
||||
class TimedAPI(FakeAPI):
|
||||
def advisory_page(self, page):
|
||||
result = super().advisory_page(page)
|
||||
now[0] += 3601
|
||||
return result
|
||||
|
||||
path = tmp_path / "cache/journal"
|
||||
api, used = TimedAPI(), e.UsedSet()
|
||||
report = d.run(
|
||||
tmp_path / "repo",
|
||||
github=api,
|
||||
osv=FakeOSV(),
|
||||
clones=FakeClones(),
|
||||
used=used,
|
||||
seed=1,
|
||||
known_empty=set(used.sources),
|
||||
journal_path=path,
|
||||
max_hours=1,
|
||||
clock=lambda: now[0],
|
||||
)
|
||||
assert report["status"] == "stopped" and report["reason"] == "max_hours"
|
||||
assert api.calls == 1
|
||||
progress = json.loads((path / "progress.json").read_text())
|
||||
assert progress["counts"]["advisories"] == 1
|
||||
assert progress["elapsed_seconds"] == 3601
|
||||
|
||||
|
||||
def test_journal_cannot_be_inside_repository(tmp_path):
|
||||
used = e.UsedSet()
|
||||
api = FakeAPI()
|
||||
report = d.run(
|
||||
tmp_path,
|
||||
github=api,
|
||||
osv=FakeOSV(),
|
||||
clones=FakeClones(),
|
||||
used=used,
|
||||
seed=1,
|
||||
known_empty=set(used.sources),
|
||||
journal_path=tmp_path / "journal",
|
||||
)
|
||||
assert report["status"] == "instrument_failure" and api.calls == 0
|
||||
assert not (tmp_path / "journal").exists()
|
||||
|
||||
|
||||
def test_journal_fsync_lock_and_corruption(tmp_path, monkeypatch):
|
||||
import json
|
||||
|
||||
j = importlib.import_module("benchmarks.unseen.journal")
|
||||
original = j.os.fsync
|
||||
synced = []
|
||||
|
||||
def fsync(fd):
|
||||
synced.append(fd)
|
||||
original(fd)
|
||||
|
||||
monkeypatch.setattr(j.os, "fsync", fsync)
|
||||
path = tmp_path / "journal"
|
||||
with j.Journal(path).open({"seed": 1}) as journal:
|
||||
before = len(synced)
|
||||
journal.put("advisories", "1", {"rows": [], "done": True})
|
||||
# Appended JSONL, directory, atomic progress file, directory.
|
||||
assert len(synced) - before == 4
|
||||
assert json.loads((path / "progress.json").read_text())["counts"]["advisories"] == 1
|
||||
with pytest.raises(ValueError, match="journal_busy"), j.Journal(path).open({"seed": 1}):
|
||||
pass
|
||||
with (path / "advisories.jsonl").open("ab") as handle:
|
||||
handle.write(b'{"id":')
|
||||
with pytest.raises(ValueError, match="journal_incomplete_record"), j.Journal(path).open({"seed": 1}):
|
||||
pass
|
||||
@@ -0,0 +1,314 @@
|
||||
"""Offline extraction: an injected local Git client over a synthetic temporary history."""
|
||||
|
||||
import importlib
|
||||
import subprocess
|
||||
import sys
|
||||
from dataclasses import replace
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
e = importlib.import_module("benchmarks.unseen.extract")
|
||||
s = importlib.import_module("benchmarks.unseen.source")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def history(tmp_path):
|
||||
calls = []
|
||||
|
||||
def runner(argv, **kwargs):
|
||||
assert not {"clone", "fetch", "push", "pull"} & set(argv)
|
||||
calls.append(argv)
|
||||
return subprocess.run(argv, **kwargs)
|
||||
|
||||
git = e.Git(tmp_path, runner=runner)
|
||||
git.run("init", "-q")
|
||||
git.run("config", "user.name", "Fixture")
|
||||
git.run("config", "user.email", "[email protected]")
|
||||
|
||||
def commit(text, message="ordinary", extra=None):
|
||||
(tmp_path / "app.py").write_text(text)
|
||||
for path, content in (extra or {}).items():
|
||||
(tmp_path / path).write_text(content)
|
||||
git.run("add", ".")
|
||||
git.run("commit", "-q", "-m", message)
|
||||
return git.run("rev-parse", "HEAD").strip()
|
||||
|
||||
root = commit("def handler(value):\n return value\n")
|
||||
git.run("tag", "v1.0")
|
||||
intro = commit("def handler(value):\n return eval(value)\n")
|
||||
git.run("tag", "v1.1")
|
||||
parent = commit("def handler(value):\n return eval( value )\n", "format")
|
||||
fix = commit("def handler(value):\n return int(value)\n", "security fix")
|
||||
candidate = s.Candidate(
|
||||
"fixture/project",
|
||||
"https://github.com/fixture/project",
|
||||
fix,
|
||||
parent,
|
||||
"GHSA-test-test-test",
|
||||
"pip",
|
||||
"injection",
|
||||
True,
|
||||
"1.1",
|
||||
"MIT",
|
||||
"permissive",
|
||||
parent,
|
||||
"LICENSE",
|
||||
)
|
||||
return git, commit, candidate, root, intro, parent, calls
|
||||
|
||||
|
||||
def test_introducing_with_reformat_walkback(history):
|
||||
git, _, candidate, root, intro, _, calls = history
|
||||
change = e.introducing(git, candidate)
|
||||
assert change["method"] == "introducing"
|
||||
assert (change["base"], change["head"]) == (root, intro)
|
||||
assert change["function"] == "handler"
|
||||
assert change["family"] == "injection" and change["post_cutoff"]
|
||||
assert change["files"][0]["head_lines"] == [[2, 2]]
|
||||
assert change["bias"] and change["advisory"] == candidate.advisory
|
||||
assert any("-w" in args and "-M" in args and "-C" in args for args in calls)
|
||||
assert git.bytes_read > 0
|
||||
|
||||
|
||||
def test_bracket_rejection_uses_fallback(history):
|
||||
git, _, candidate, _, _, parent, _ = history
|
||||
change = e.introducing(git, replace(candidate, first_affected="1.0"))
|
||||
assert change["method"] == "last_touch" and change["head"] == parent
|
||||
assert change["fallback_reasons"]["version_bracket"] >= 1
|
||||
|
||||
|
||||
def test_size_cap_and_merge_rejection(history):
|
||||
git, _, candidate, _, _, _, _ = history
|
||||
|
||||
class Oversize:
|
||||
def __getattr__(self, name):
|
||||
return getattr(git, name)
|
||||
|
||||
def stats(self, base, head):
|
||||
return [("app.py", 2001, 0)]
|
||||
|
||||
assert e.introducing(Oversize(), candidate)["method"] == "last_touch"
|
||||
|
||||
class Merge(Oversize):
|
||||
def parents(self, sha):
|
||||
parents = git.parents(sha)
|
||||
return parents + ["f" * 40] if sha != candidate.fix else parents
|
||||
|
||||
with pytest.raises(s.Rejected, match="no_range"):
|
||||
e.introducing(Merge(), candidate)
|
||||
|
||||
|
||||
def test_pure_addition_blames_function_declaration(history):
|
||||
git, commit, candidate, _, _, _, _ = history
|
||||
git.run("checkout", "-q", candidate.parent)
|
||||
parent = commit("def handler(value):\n return eval(value)\n\ndef added(value):\n return value\n")
|
||||
fix = commit("def handler(value):\n return eval(value)\n\ndef added(value):\n assert value\n return value\n")
|
||||
change = e.introducing(git, replace(candidate, fix=fix, parent=parent, first_affected=None, family="access_control"))
|
||||
assert change["head"] == parent and change["method"] == "introducing"
|
||||
assert change["function"] == "added"
|
||||
|
||||
|
||||
def test_git_failures_are_sanitized_and_not_empty(tmp_path):
|
||||
def failed(argv, **kwargs):
|
||||
return subprocess.CompletedProcess(argv, 128, b"", b"private repository secret")
|
||||
|
||||
with pytest.raises(e.InstrumentFailure, match="git_exit_128"):
|
||||
e.Git(tmp_path, runner=failed).run("log")
|
||||
|
||||
|
||||
def test_walkback_is_bounded_and_file_cap_falls_back(history):
|
||||
git, _, candidate, _, _, parent, _ = history
|
||||
|
||||
class FileCap:
|
||||
def __getattr__(self, name):
|
||||
return getattr(git, name)
|
||||
|
||||
def stats(self, base, head):
|
||||
return [("app.py", 1, 0)] * 51
|
||||
|
||||
result = e.introducing(FileCap(), candidate)
|
||||
assert result["method"] == "last_touch"
|
||||
assert result["fallback_reasons"]["size_cap"]
|
||||
|
||||
class Walk:
|
||||
calls = 0
|
||||
|
||||
def __getattr__(self, name):
|
||||
return getattr(git, name)
|
||||
|
||||
def blame(self, revision, path, line):
|
||||
self.calls += 1
|
||||
return parent, path, 2
|
||||
|
||||
def parents(self, revision):
|
||||
return [parent] if revision != candidate.fix else [candidate.parent]
|
||||
|
||||
walk = Walk()
|
||||
result = e.introducing(walk, candidate)
|
||||
assert walk.calls == 6
|
||||
assert result["fallback_reasons"]["walkback_limit"] == 1
|
||||
|
||||
|
||||
def test_root_rejected_and_head_function_resolved(history):
|
||||
git, _, candidate, root, _, _, _ = history
|
||||
|
||||
class Root:
|
||||
def __getattr__(self, name):
|
||||
return getattr(git, name)
|
||||
|
||||
def blame(self, revision, path, line):
|
||||
return root, path, 1
|
||||
|
||||
result = e.introducing(Root(), candidate)
|
||||
assert result["method"] == "last_touch"
|
||||
assert result["fallback_reasons"]["merge_or_root"] == 1
|
||||
assert result["function_lines"] == [1, 2]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path,text",
|
||||
[
|
||||
("a.php", "<?php\nfunction handle($x) {\n return $x;\n}\n"),
|
||||
("a.js", "function handle(x) {\n return x;\n}\n"),
|
||||
("a.ts", "const handle = (x) => {\n return x;\n}\n"),
|
||||
("A.java", "class A {\n public String handle(String x) {\n return x;\n }\n}\n"),
|
||||
],
|
||||
)
|
||||
def test_supported_function_boundaries(path, text):
|
||||
ranges = e.functions(path, text)
|
||||
assert len(ranges) == 1 and ranges[0][0] == "handle"
|
||||
assert ranges[0][2] < len(text.splitlines()) + 1
|
||||
|
||||
|
||||
def test_ambiguous_parent_is_not_proof_of_absence(history):
|
||||
git, commit, candidate, _, _, _, _ = history
|
||||
duplicate = (
|
||||
"class A:\n def handler(self, value):\n return eval(value)\n\n"
|
||||
"class B:\n def handler(self, value):\n return value\n"
|
||||
)
|
||||
commit(duplicate)
|
||||
parent = commit("class A:\n def handler(self, value):\n return eval( value )\n")
|
||||
fix = commit("class A:\n def handler(self, value):\n return int(value)\n")
|
||||
selected = replace(candidate, fix=fix, parent=parent, first_affected=None)
|
||||
|
||||
class BlameLast:
|
||||
def __getattr__(self, name):
|
||||
return getattr(git, name)
|
||||
|
||||
def blame(self, revision, path, line):
|
||||
return parent, path, 3
|
||||
|
||||
result = e.introducing(BlameLast(), selected)
|
||||
assert result["method"] == "last_touch"
|
||||
assert result["fallback_reasons"]["ambiguous_parent"] == 1
|
||||
|
||||
|
||||
def ordinary(index, **overrides):
|
||||
row = {
|
||||
"head": f"{index:040x}",
|
||||
"parents": ["f" * 40],
|
||||
"timestamp": index * 100,
|
||||
"message": "update query token session valid",
|
||||
"author": "Human",
|
||||
"files": [{"path": "app.py", "added": 5, "removed": 0, "head_lines": [[2, 6]], "functions": ["ordinary"], "generated": False}],
|
||||
}
|
||||
row.update(overrides)
|
||||
return row
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"message",
|
||||
[
|
||||
"security fix",
|
||||
"CVE-2026",
|
||||
"GHSA-test",
|
||||
"vulnerability",
|
||||
"injection fix",
|
||||
"XSS",
|
||||
"CSRF",
|
||||
"SSRF",
|
||||
"exploit",
|
||||
"sanitize",
|
||||
"escaping",
|
||||
"Revert changes",
|
||||
"This reverts commit abc",
|
||||
"bump package",
|
||||
"upgrade package from 1 to 2",
|
||||
],
|
||||
)
|
||||
def test_ordinary_message_exclusions(message):
|
||||
assert e.ordinary_rejection(ordinary(1, message=message), set(), set()) is not None
|
||||
|
||||
|
||||
def test_narrow_filter_and_every_other_exclusion():
|
||||
assert e.ordinary_rejection(ordinary(1), set(), set()) is None
|
||||
assert e.ordinary_rejection(ordinary(1, parents=["a", "b"]), set(), set()) == "merge_or_root"
|
||||
assert e.ordinary_rejection(ordinary(1, author="dependabot[bot]"), set(), set()) == "dependency"
|
||||
assert e.ordinary_rejection(ordinary(1, author="renovate"), set(), set()) == "dependency"
|
||||
assert e.ordinary_rejection(ordinary(1), {f"{1:040x}"}, set()) == "advisory_fix"
|
||||
assert e.ordinary_rejection(ordinary(1), set(), {("app.py", "ordinary")}) == "vulnerable_function"
|
||||
row = ordinary(1)
|
||||
row["files"][0]["added"] = 1001
|
||||
assert e.ordinary_rejection(row, set(), set()) == "size_cap"
|
||||
for path, reason in [
|
||||
("README.md", "no_source"),
|
||||
("package.json", "manifest_only"),
|
||||
("package-lock.json", "manifest_only"),
|
||||
("vendor/a.py", "generated"),
|
||||
("node_modules/a.js", "generated"),
|
||||
("dist/a.js", "generated"),
|
||||
("app.min.js", "generated"),
|
||||
]:
|
||||
row = ordinary(1)
|
||||
row["files"][0]["path"] = path
|
||||
assert e.ordinary_rejection(row, set(), set()) == reason
|
||||
row = ordinary(1)
|
||||
row["files"][0]["generated"] = True
|
||||
assert e.ordinary_rejection(row, set(), set()) == "generated"
|
||||
|
||||
|
||||
def test_draw_seed_proportional_cells_floor_and_thirds():
|
||||
rows = [ordinary(i) for i in range(120)]
|
||||
for i, row in enumerate(rows):
|
||||
row["files"][0]["added"] = [5, 20, 100, 500][i % 4]
|
||||
drawn, counts = e.draw_ordinary(rows, set(), set(), seed=19)
|
||||
assert len(drawn) == 12 and len({r["head"] for r in drawn}) == 12
|
||||
assert {r["time_third"] for r in drawn} == {0, 1, 2}
|
||||
assert len({(r["time_third"], r["size_bucket"]) for r in drawn}) == 12
|
||||
assert drawn == e.draw_ordinary(list(reversed(rows)), set(), set(), seed=19)[0]
|
||||
assert counts["eligible"] == 120
|
||||
with pytest.raises(s.Rejected, match="ordinary_floor"):
|
||||
e.draw_ordinary(rows[:39], set(), set(), seed=19)
|
||||
# Thirds come from the history's time span, not rank among eligible changes.
|
||||
with pytest.raises(s.Rejected, match="time_thirds"):
|
||||
e.draw_ordinary([ordinary(i) for i in range(40)] + [ordinary(10000, message="security")], set(), set(), seed=1)
|
||||
|
||||
|
||||
def test_git_ordinary_collection_generated_attributes_and_headers(history):
|
||||
git, commit, _, _, _, _, _ = history
|
||||
commit("# Generated by fixture\nx = 2\n", extra={".gitattributes": "other.py linguist-generated=true\n", "other.py": "x = 1\n"})
|
||||
rows = e.ordinary_history(git, "HEAD")
|
||||
assert rows and rows[0]["files"]
|
||||
files = {f["path"]: f for f in rows[0]["files"]}
|
||||
assert files["app.py"]["generated"] and files["other.py"]["generated"]
|
||||
assert files["app.py"]["head_lines"]
|
||||
|
||||
|
||||
def test_attribute_unset_and_mixed_generated_source(history):
|
||||
git, commit, _, _, _, _, _ = history
|
||||
sha = commit("x = 1\n", extra={".gitattributes": "*.py linguist-generated\napp.py -linguist-generated\n"})
|
||||
assert not e.generated(git, sha, "app.py", "x = 1\n")
|
||||
row = ordinary(1)
|
||||
row["files"].append({**row["files"][0], "path": "dist/b.js", "generated": True})
|
||||
assert e.ordinary_rejection(row, set(), set()) is None
|
||||
|
||||
|
||||
def test_unbalanced_cells_remain_proportional_and_seed_changes_draw():
|
||||
rows = [ordinary(i) for i in range(120)]
|
||||
for i, row in enumerate(rows):
|
||||
row["files"][0]["added"] = 5 if i % 4 else 100
|
||||
draw, _ = e.draw_ordinary(rows, set(), set(), seed=1)
|
||||
assert len([r for r in draw if r["size_bucket"] == 0]) == 9
|
||||
assert draw != e.draw_ordinary(rows, set(), set(), seed=2)[0]
|
||||
@@ -0,0 +1,182 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user