mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Marius CiepluchandClaude Opus 4.8
197fe8a19c
Group 3 was unimplemented, so there was no way to turn the draft into a frozen pool. Add: - bisect_guard.py: stages the cache draft and bisects it against the reserved-repository guard, run as a black box (pytest -k referenced, stdout/stderr discarded, exit code only; an exit other than 0/1 aborts as an instrument failure). Drops guard-tripping candidates and reports the dropped COUNT only -- never a name or URL. Guard runner is injectable for offline tests. - freeze.py: writes canonical pool-p1.toml (+ private counterpart, reusing draft's public/private split and private_digest) and freeze-p1.json (used-set digest, per-source counts, guard result, dropped count, label-check counts, freeze digest = sha256 of the canonical public manifest). A matching retry is a no-op; any changed frozen artifact requires a new pool id (never an overwrite). - protocol-p1.md: the frozen scoring/statistics protocol from design sections 3 and 4. Offline only; the guard subprocess is the sole subprocess and is mocked in tests. 42 new tests pass (incl the group-1.3 guard digest check on a fixture pool); neighbouring unseen suites green; ruff clean. New files only -- does not touch the journal-pinned draft/extract modules, so a running draw is unaffected. Co-Authored-By: Claude Opus 4.8 <[email protected]>
198 lines
8.6 KiB
Python
198 lines
8.6 KiB
Python
"""Freeze fixture pools without reading reservations or running subprocesses."""
|
|
|
|
import hashlib
|
|
import importlib
|
|
import json
|
|
import tomllib
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from test_unseen_bisect import b, d, draft_pair
|
|
|
|
f = importlib.import_module("benchmarks.unseen.freeze")
|
|
|
|
|
|
def prepared(tmp_path):
|
|
root, draft, rows = draft_pair(tmp_path)
|
|
b.run(root, draft, guard_runner=lambda root: 0)
|
|
protocol = root / "benchmarks/unseen/protocol-p1.md"
|
|
protocol.write_bytes(Path("benchmarks/unseen/protocol-p1.md").read_bytes())
|
|
return root, rows
|
|
|
|
|
|
def used():
|
|
return {"sha256": "a" * 64, "sources": {"pairs": {"count": 7}, "memory_s3": {"count": 3}}}
|
|
|
|
|
|
def test_freeze_split_receipt_and_existing_guard_digest(tmp_path, capsys, caplog, monkeypatch):
|
|
root, rows = prepared(tmp_path)
|
|
record = f.run(root, used_report=used())
|
|
directory = root / "benchmarks/unseen"
|
|
pool = directory / "pool-p1.toml"
|
|
public = tomllib.loads(pool.read_text())
|
|
private = tomllib.loads((directory / "private/pool-p1.toml").read_text())
|
|
assert public["status"] == private["status"] == "frozen"
|
|
assert record["used_set_digest"] == "a" * 64
|
|
assert record["source_counts"] == {"pairs": 7, "memory_s3": 3}
|
|
assert record["label_check_counts"] == {"confirmed": 3, "refactor": 3, "unclear": 3, "unchecked": 0}
|
|
assert record["guard_result"] == 0 and record["dropped_count"] == 0
|
|
assert record["protocol_digest"] == hashlib.sha256((directory / "protocol-p1.md").read_bytes()).hexdigest()
|
|
assert record["freeze_digest"] == hashlib.sha256(d.canonical(public)).hexdigest()
|
|
assert json.loads((directory / "freeze-p1.json").read_text()) == record
|
|
for row in rows:
|
|
assert row["repository"] not in json.dumps(record) and row["url"] not in json.dumps(record)
|
|
if row["license_class"] == "private":
|
|
assert row["url"] not in pool.read_text() and row["repository"] not in pool.read_text()
|
|
pointer = next(p for p in public["repository"] if p["id"] == row["id"])
|
|
assert pointer == {"id": row["id"], "slice": row["slice"], "private": True, "sha256": d.private_digest(row)}
|
|
assert row in private["repository"]
|
|
else:
|
|
assert row in public["repository"]
|
|
assert (directory / "private/pool-p1.toml").stat().st_mode & 0o777 == 0o600
|
|
guard = importlib.import_module("test_independent_population")
|
|
monkeypatch.setattr(guard, "ROOT", root)
|
|
guard.test_frozen_pool_manifest_digest()
|
|
assert capsys.readouterr() == ("", "") and not caplog.records
|
|
|
|
|
|
def test_digest_and_retry_ignore_toml_key_order(tmp_path):
|
|
root, _ = prepared(tmp_path)
|
|
original = f.run(root, used_report=used())
|
|
pool = root / "benchmarks/unseen/pool-p1.toml"
|
|
text = pool.read_text()
|
|
lines = text.splitlines()
|
|
pool.write_text("\n".join([lines[2], lines[1], lines[0], *lines[3:]]) + "\n")
|
|
before = pool.read_bytes()
|
|
assert f.run(root, used_report=used()) == original
|
|
assert pool.read_bytes() == before
|
|
|
|
|
|
@pytest.mark.parametrize("target", ["pool-p1.toml", "private/pool-p1.toml", "freeze-p1.json"])
|
|
def test_numeric_type_change_is_a_different_frozen_artifact(tmp_path, target):
|
|
root, _ = prepared(tmp_path)
|
|
f.run(root, used_report=used())
|
|
path = root / "benchmarks/unseen" / target
|
|
text = path.read_text()
|
|
if path.suffix == ".toml":
|
|
modified = text.replace("seed = 19", "seed = 19.0")
|
|
else:
|
|
modified = text.replace('"guard_result": 0', '"guard_result": false')
|
|
assert text != modified
|
|
path.write_text(modified)
|
|
with pytest.raises(FileExistsError, match="new_pool"):
|
|
f.run(root, used_report=used())
|
|
assert path.read_text() == modified
|
|
|
|
|
|
@pytest.mark.parametrize("target", ["pool-p1.toml", "private/pool-p1.toml", "freeze-p1.json", "protocol-p1.md"])
|
|
def test_refuses_different_frozen_files(tmp_path, target):
|
|
root, _ = prepared(tmp_path)
|
|
f.run(root, used_report=used())
|
|
directory = root / "benchmarks/unseen"
|
|
path = directory / target
|
|
if path.suffix == ".toml":
|
|
path.write_text(path.read_text().replace('status = "frozen"', 'status = "modified"'))
|
|
elif path.suffix == ".json":
|
|
record = json.loads(path.read_text())
|
|
record["dropped_count"] += 1
|
|
path.write_text(json.dumps(record))
|
|
else:
|
|
path.write_text("changed protocol")
|
|
before = {p: p.read_bytes() for p in directory.rglob("*") if p.is_file()}
|
|
with pytest.raises(FileExistsError, match="new_pool"):
|
|
f.run(root, used_report=used())
|
|
assert all(p.read_bytes() == value for p, value in before.items())
|
|
|
|
|
|
@pytest.mark.parametrize("target", ["draft-p1.toml", "private/draft-p1.toml", "bisect-p1.json"])
|
|
def test_refuses_stale_or_red_guard_receipt(tmp_path, target):
|
|
root, _ = prepared(tmp_path)
|
|
path = root / "benchmarks/unseen" / target
|
|
if target.endswith(".json"):
|
|
receipt = json.loads(path.read_text())
|
|
receipt["guard_result"] = 1
|
|
path.write_text(json.dumps(receipt))
|
|
else:
|
|
before = path.read_text()
|
|
after = before.replace('"head" = "', '"head" = "0', 1)
|
|
assert before != after
|
|
path.write_text(after)
|
|
with pytest.raises((ValueError, b.InstrumentFailure)):
|
|
f.run(root, used_report=used())
|
|
assert not (root / "benchmarks/unseen/pool-p1.toml").exists()
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["digest", "source_name", "count"])
|
|
def test_rejects_identity_bearing_metadata(tmp_path, mutation):
|
|
root, rows = prepared(tmp_path)
|
|
report = used()
|
|
if mutation == "digest":
|
|
report["sha256"] = rows[0]["url"]
|
|
elif mutation == "source_name":
|
|
report["sources"][rows[0]["repository"]] = {"count": 1}
|
|
else:
|
|
report["sources"]["pairs"]["count"] = rows[0]["url"]
|
|
with pytest.raises(ValueError):
|
|
f.run(root, used_report=report)
|
|
assert not (root / "benchmarks/unseen/freeze-p1.json").exists()
|
|
|
|
|
|
def test_missing_protocol_aborts_before_publication(tmp_path):
|
|
root, _ = prepared(tmp_path)
|
|
(root / "benchmarks/unseen/protocol-p1.md").unlink()
|
|
with pytest.raises(ValueError, match="protocol"):
|
|
f.run(root, used_report=used())
|
|
assert not (root / "benchmarks/unseen/pool-p1.toml").exists()
|
|
|
|
|
|
def test_drops_flow_into_freeze_counts_and_private_manifest(tmp_path):
|
|
root, draft, rows = draft_pair(tmp_path)
|
|
hidden = {rows[i]["repository"] for i in (1, 2)}
|
|
from test_unseen_bisect import staged
|
|
|
|
b.run(root, draft, guard_runner=lambda root: int(any(r["repository"] in hidden for r in staged(root))))
|
|
directory = root / "benchmarks/unseen"
|
|
(directory / "protocol-p1.md").write_bytes(Path("benchmarks/unseen/protocol-p1.md").read_bytes())
|
|
record = f.run(root, used_report=used())
|
|
assert record["dropped_count"] == 2
|
|
assert record["label_check_counts"] == {"confirmed": 3, "refactor": 2, "unclear": 2, "unchecked": 0}
|
|
assert len(tomllib.loads((directory / "pool-p1.toml").read_text())["repository"]) == 7
|
|
assert len(tomllib.loads((directory / "private/pool-p1.toml").read_text())["repository"]) == 3
|
|
|
|
|
|
def test_changed_private_labels_invalidate_guard_receipt(tmp_path):
|
|
root, _ = prepared(tmp_path)
|
|
path = root / "benchmarks/unseen/private/draft-p1.toml"
|
|
before = path.read_text()
|
|
after = before.replace('"label_check" = "refactor"', '"label_check" = "confirmed"', 1)
|
|
assert before != after
|
|
path.write_text(after)
|
|
with pytest.raises(ValueError, match="guard_receipt"):
|
|
f.run(root, used_report=used())
|
|
|
|
|
|
def test_utf8_digest_matches_guard_with_non_ascii_metadata(tmp_path, monkeypatch):
|
|
root, draft, _ = draft_pair(tmp_path)
|
|
data = tomllib.loads(draft.read_text())
|
|
data["repository"][0]["license_path"] = "licence-ä.txt"
|
|
draft.write_text(d.manifest(data["repository"], 19))
|
|
b.run(root, draft, guard_runner=lambda root: 0)
|
|
(root / "benchmarks/unseen/protocol-p1.md").write_bytes(Path("benchmarks/unseen/protocol-p1.md").read_bytes())
|
|
f.run(root, used_report=used())
|
|
guard = importlib.import_module("test_independent_population")
|
|
monkeypatch.setattr(guard, "ROOT", root)
|
|
guard.test_frozen_pool_manifest_digest()
|
|
|
|
|
|
def test_freeze_cli_redacts_failure_and_reports_only_record(tmp_path, capsys):
|
|
root, rows = prepared(tmp_path)
|
|
report = tmp_path / "used.json"
|
|
report.write_text(json.dumps(used()))
|
|
assert f.main(["--root", str(root), "--used-report", str(report)]) == 0
|
|
output = capsys.readouterr()
|
|
assert json.loads(output.out)["guard_result"] == 0 and output.err == ""
|
|
assert all(r["repository"] not in output.out for r in rows)
|
|
report.write_text(rows[0]["url"])
|
|
assert f.main(["--root", str(root), "--used-report", str(report)]) == 1
|
|
assert capsys.readouterr() == ('{"status": "freeze_refused"}\n', "")
|