core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Merge pull request #887 from AssetOverflow/feat/generalization-cache-verifier

Browse source
This commit is contained in:
Shay 2026-06-23 07:17:25 -07:00 committed by GitHub
commit 3100f84e73
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 586 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
evals/generalization/__init__.py
View file

@ -8,10 +8,20 @@ from evals.generalization.manifest_schema import (
load_and_validate_manifest,
validate_manifest_data,
)
from evals.generalization.cache_verifier import (
GENERALIZATION_CACHE_VERIFIER_POLICY_VERSION,
CacheVerificationRecord,
CacheVerificationReport,
verify_local_generalization_cache,
)
__all__ = [
"GeneralizationBenchmarkManifest",
"ManifestValidationError",
"load_and_validate_manifest",
"validate_manifest_data",
"GENERALIZATION_CACHE_VERIFIER_POLICY_VERSION",
"CacheVerificationRecord",
"CacheVerificationReport",
"verify_local_generalization_cache",
]

126
evals/generalization/cache_verifier.py Normal file
View file

@ -0,0 +1,126 @@
"""Verifier for local generalization benchmark cache directories."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from evals.generalization.manifest_schema import (
load_and_validate_manifest,
ManifestValidationError,
)
GENERALIZATION_CACHE_VERIFIER_POLICY_VERSION = "generalization_cache_verifier.v1"
@dataclass(frozen=True, slots=True)
class CacheVerificationRecord:
"""Readiness and verification details for a single generalization benchmark cache."""
dataset: str
manifest_path: str
local_cache: str
exists: bool
license_ready: bool
checksum_ready: bool
runnable: bool
reason_codes: tuple[str, ...]
@dataclass(frozen=True, slots=True)
class CacheVerificationReport:
"""Aggregated cache verification report across all manifests."""
policy_version: str
records: tuple[CacheVerificationRecord, ...]
all_runnable: bool
reason_codes: tuple[str, ...]
def verify_local_generalization_cache(
repo_root: Path,
manifests_dir: Path,
require_present: bool = False,
) -> CacheVerificationReport:
"""Verify all generalization benchmark manifests and local cache directory readiness.
Args:
repo_root: Path to the root of the git repository.
manifests_dir: Path to the directory containing YAML manifests.
require_present: If True, raises ValueError if any local cache directory is missing.
Returns:
A CacheVerificationReport containing verification records for all manifests.
"""
if not manifests_dir.exists() or not manifests_dir.is_dir():
raise ValueError(f"Manifests directory does not exist: {manifests_dir}")
manifest_files = sorted(manifests_dir.glob("*.yaml"))
records: list[CacheVerificationRecord] = []
benchmarks_dir = (repo_root / ".data/benchmarks").resolve()
for path in manifest_files:
manifest = load_and_validate_manifest(path)
local_cache = manifest.local_cache
# Check path traversal and escaping
# resolved cache directory path must lie strictly within .data/benchmarks/
cache_path = (repo_root / local_cache).resolve()
try:
cache_path.relative_to(benchmarks_dir)
except ValueError as exc:
raise ManifestValidationError(
f"Path traversal detected: local_cache {local_cache!r} in {path.name} "
f"resolves outside .data/benchmarks/ ({cache_path})"
) from exc
# Check directory existence (no files inside should be read/opened)
exists = cache_path.is_dir()
if require_present and not exists:
raise ValueError(
f"Cache directory for dataset {manifest.dataset} does not exist: {cache_path}"
)
license_ready = manifest.license != "TODO_VERIFY_BEFORE_CACHE"
checksum_ready = manifest.sha256 != "TODO_AFTER_DOWNLOAD"
runnable = license_ready and checksum_ready and exists
reason_codes_list: list[str] = []
if not exists:
reason_codes_list.append("CACHE_ABSENT")
if not license_ready:
reason_codes_list.append("LICENSE_UNRESOLVED")
if not checksum_ready:
reason_codes_list.append("CHECKSUM_UNRESOLVED")
try:
rel_manifest_path = str(path.relative_to(repo_root))
except ValueError:
rel_manifest_path = str(path)
records.append(
CacheVerificationRecord(
dataset=manifest.dataset,
manifest_path=rel_manifest_path,
local_cache=local_cache,
exists=exists,
license_ready=license_ready,
checksum_ready=checksum_ready,
runnable=runnable,
reason_codes=tuple(reason_codes_list),
)
)
all_runnable = all(r.runnable for r in records)
report_reasons = set()
for r in records:
report_reasons.update(r.reason_codes)
return CacheVerificationReport(
policy_version=GENERALIZATION_CACHE_VERIFIER_POLICY_VERSION,
records=tuple(records),
all_runnable=all_runnable,
reason_codes=tuple(sorted(report_reasons)),
)

98
scripts/benchmarks/verify_generalization_cache.py Executable file
View file

@ -0,0 +1,98 @@
#!/usr/bin/env python3
"""CLI script to verify local generalization benchmark cache directories."""
from __future__ import annotations
import argparse
import json
import sys
from dataclasses import asdict
from pathlib import Path
# Resolve repository root and add to sys.path to support imports
script_path = Path(__file__).resolve()
repo_root = script_path.parent.parent.parent
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))
from evals.generalization.cache_verifier import verify_local_generalization_cache # noqa: E402
def main() -> None:
# Explicit rejection check for download/fetch flags before argparse
forbidden = {"--download", "--fetch", "--pull"}
for arg in sys.argv[1:]:
if arg in forbidden:
print(
f"Error: Flag {arg} is not supported. Fetching or downloading dataset content is forbidden by policy.",
file=sys.stderr,
)
sys.exit(1)
parser = argparse.ArgumentParser(
description="Verify local generalization benchmark cache."
)
parser.add_argument(
"--metadata-only",
action="store_true",
help="Only validate manifests metadata without requiring cache presence.",
)
parser.add_argument(
"--require-present",
action="store_true",
help="Require all local cache directories to exist.",
)
parser.add_argument(
"--json",
action="store_true",
help="Print report in deterministic JSON format.",
)
args = parser.parse_args()
script_path = Path(__file__).resolve()
repo_root = script_path.parent.parent.parent
manifests_dir = repo_root / "evals" / "generalization" / "manifests"
try:
report = verify_local_generalization_cache(
repo_root=repo_root,
manifests_dir=manifests_dir,
require_present=args.require_present,
)
except Exception as exc:
print(f"Verification Failed: {exc}", file=sys.stderr)
sys.exit(1)
if args.json:
report_dict = asdict(report)
print(json.dumps(report_dict, indent=2, sort_keys=True))
else:
print(
f"Generalization Cache Verification Report (Policy: {report.policy_version})"
)
print("=" * 80)
for record in report.records:
status = "RUNNABLE" if record.runnable else "NOT RUNNABLE"
print(f"Dataset: {record.dataset}")
print(f" Manifest: {record.manifest_path}")
print(f" Cache Path: {record.local_cache}")
print(f" Exists: {record.exists}")
print(f" License Ready: {record.license_ready}")
print(f" Checksum Ready: {record.checksum_ready}")
print(f" Status: {status}")
if record.reason_codes:
print(f" Reasons: {', '.join(record.reason_codes)}")
print("-" * 80)
print(f"All datasets runnable: {report.all_runnable}")
# If --require-present is set, exit non-zero if any dataset is not runnable
if args.require_present and not report.all_runnable:
sys.exit(1)
sys.exit(0)
if __name__ == "__main__":
main()

352
tests/test_generalization_cache_verifier.py Normal file
View file

@ -0,0 +1,352 @@
"""Tests for the local generalization benchmark cache verifier."""
from __future__ import annotations
import builtins
import json
import subprocess
import sys
from pathlib import Path
import pytest
import yaml
from evals.generalization.cache_verifier import (
verify_local_generalization_cache,
GENERALIZATION_CACHE_VERIFIER_POLICY_VERSION,
)
from evals.generalization.manifest_schema import ManifestValidationError
from scripts.benchmarks.verify_generalization_cache import main as cli_main
def write_test_manifest(
manifests_dir: Path,
dataset: str = "TEST",
license: str = "TODO_VERIFY_BEFORE_CACHE",
sha256: str = "TODO_AFTER_DOWNLOAD",
local_cache: str = ".data/benchmarks/test/",
) -> Path:
"""Helper to write a synthetic manifest YAML file."""
data = {
"dataset": dataset,
"purpose": "sealed_audit_not_training",
"description": "Test description",
"source": "hf://test/test",
"license": license,
"version": "pinned",
"split": "test",
"sha256": sha256,
"local_cache": local_cache,
"repo_policy": "manifest_only",
"inspection_policy": "aggregate_reports_only",
"mutation_policy": "no_direct_pack_policy_operator_mutation",
}
path = manifests_dir / f"{dataset.lower()}.yaml"
path.write_text(yaml.dump(data), encoding="utf-8")
return path
def test_metadata_only_succeeds_with_absent_cache(tmp_path: Path) -> None:
"""metadata-only succeeds with absent cache."""
repo_root = tmp_path
manifests_dir = tmp_path / "manifests"
manifests_dir.mkdir()
(repo_root / ".data" / "benchmarks").mkdir(parents=True)
write_test_manifest(manifests_dir, dataset="TEST_ABSENT")
report = verify_local_generalization_cache(
repo_root=repo_root, manifests_dir=manifests_dir, require_present=False
)
assert len(report.records) == 1
record = report.records[0]
assert record.dataset == "TEST_ABSENT"
assert record.exists is False
assert record.runnable is False
assert "CACHE_ABSENT" in record.reason_codes
assert report.all_runnable is False
def test_require_present_refuses_absent_cache(tmp_path: Path) -> None:
"""require-present refuses absent cache."""
repo_root = tmp_path
manifests_dir = tmp_path / "manifests"
manifests_dir.mkdir()
(repo_root / ".data" / "benchmarks").mkdir(parents=True)
write_test_manifest(manifests_dir, dataset="TEST_ABSENT")
with pytest.raises(ValueError, match="does not exist"):
verify_local_generalization_cache(
repo_root=repo_root, manifests_dir=manifests_dir, require_present=True
)
def test_todo_license_makes_runnable_false(tmp_path: Path) -> None:
"""TODO license makes runnable=false."""
repo_root = tmp_path
manifests_dir = tmp_path / "manifests"
manifests_dir.mkdir()
(repo_root / ".data" / "benchmarks").mkdir(parents=True)
cache_dir = repo_root / ".data" / "benchmarks" / "test"
cache_dir.mkdir(parents=True)
valid_sha = "a" * 64
write_test_manifest(
manifests_dir,
dataset="TEST_TODO_LICENSE",
license="TODO_VERIFY_BEFORE_CACHE",
sha256=valid_sha,
)
report = verify_local_generalization_cache(
repo_root=repo_root, manifests_dir=manifests_dir, require_present=False
)
record = report.records[0]
assert record.exists is True
assert record.license_ready is False
assert record.checksum_ready is True
assert record.runnable is False
assert "LICENSE_UNRESOLVED" in record.reason_codes
def test_todo_sha256_makes_runnable_false(tmp_path: Path) -> None:
"""TODO sha256 makes runnable=false."""
repo_root = tmp_path
manifests_dir = tmp_path / "manifests"
manifests_dir.mkdir()
(repo_root / ".data" / "benchmarks").mkdir(parents=True)
cache_dir = repo_root / ".data" / "benchmarks" / "test"
cache_dir.mkdir(parents=True)
write_test_manifest(
manifests_dir,
dataset="TEST_TODO_SHA",
license="MIT",
sha256="TODO_AFTER_DOWNLOAD",
)
report = verify_local_generalization_cache(
repo_root=repo_root, manifests_dir=manifests_dir, require_present=False
)
record = report.records[0]
assert record.exists is True
assert record.license_ready is True
assert record.checksum_ready is False
assert record.runnable is False
assert "CHECKSUM_UNRESOLVED" in record.reason_codes
def test_resolved_license_valid_sha_present_cache_runnable(tmp_path: Path) -> None:
"""resolved license + valid sha256 + present cache makes runnable=true."""
repo_root = tmp_path
manifests_dir = tmp_path / "manifests"
manifests_dir.mkdir()
(repo_root / ".data" / "benchmarks").mkdir(parents=True)
cache_dir = repo_root / ".data" / "benchmarks" / "test"
cache_dir.mkdir(parents=True)
valid_sha = "b" * 64
write_test_manifest(
manifests_dir,
dataset="TEST_RUNNABLE",
license="Apache-2.0",
sha256=valid_sha,
)
report = verify_local_generalization_cache(
repo_root=repo_root, manifests_dir=manifests_dir, require_present=True
)
record = report.records[0]
assert record.exists is True
assert record.license_ready is True
assert record.checksum_ready is True
assert record.runnable is True
assert len(record.reason_codes) == 0
assert report.all_runnable is True
def test_local_cache_outside_benchmarks_refuses(tmp_path: Path) -> None:
"""local_cache outside .data/benchmarks refuses."""
repo_root = tmp_path
manifests_dir = tmp_path / "manifests"
manifests_dir.mkdir()
(repo_root / ".data" / "benchmarks").mkdir(parents=True)
# Write a manifest directly with invalid local_cache
data = {
"dataset": "BAD_CACHE",
"purpose": "sealed_audit_not_training",
"description": "Test description",
"source": "hf://test/test",
"license": "MIT",
"version": "pinned",
"split": "test",
"sha256": "c" * 64,
"local_cache": ".data/other/test/", # outside .data/benchmarks/
"repo_policy": "manifest_only",
"inspection_policy": "aggregate_reports_only",
"mutation_policy": "no_direct_pack_policy_operator_mutation",
}
path = manifests_dir / "bad_cache.yaml"
path.write_text(yaml.dump(data), encoding="utf-8")
with pytest.raises(ManifestValidationError):
verify_local_generalization_cache(
repo_root=repo_root, manifests_dir=manifests_dir, require_present=False
)
def test_path_traversal_local_cache_refuses(tmp_path: Path) -> None:
"""path traversal local_cache refuses."""
repo_root = tmp_path
manifests_dir = tmp_path / "manifests"
manifests_dir.mkdir()
(repo_root / ".data" / "benchmarks").mkdir(parents=True)
# Write a manifest with path traversal escaping via ..
data = {
"dataset": "TRAVERSAL",
"purpose": "sealed_audit_not_training",
"description": "Test description",
"source": "hf://test/test",
"license": "MIT",
"version": "pinned",
"split": "test",
"sha256": "c" * 64,
"local_cache": ".data/benchmarks/../other/", # path traversal
"repo_policy": "manifest_only",
"inspection_policy": "aggregate_reports_only",
"mutation_policy": "no_direct_pack_policy_operator_mutation",
}
path = manifests_dir / "traversal.yaml"
path.write_text(yaml.dump(data), encoding="utf-8")
with pytest.raises(ManifestValidationError, match="Path traversal detected"):
verify_local_generalization_cache(
repo_root=repo_root, manifests_dir=manifests_dir, require_present=False
)
def test_cli_rejects_download_flags(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""CLI rejects --download / --fetch / --pull."""
for flag in ("--download", "--fetch", "--pull"):
monkeypatch.setattr(sys, "argv", ["verify_generalization_cache.py", flag])
with pytest.raises(SystemExit) as excinfo:
cli_main()
assert excinfo.value.code != 0
captured = capsys.readouterr()
assert "not supported" in captured.err
def test_json_output_deterministic(tmp_path: Path) -> None:
"""JSON output is deterministic."""
repo_root = tmp_path
manifests_dir = tmp_path / "manifests"
manifests_dir.mkdir()
(repo_root / ".data" / "benchmarks").mkdir(parents=True)
write_test_manifest(manifests_dir, dataset="B_DATASET")
write_test_manifest(manifests_dir, dataset="A_DATASET")
report = verify_local_generalization_cache(
repo_root=repo_root, manifests_dir=manifests_dir, require_present=False
)
from dataclasses import asdict
json1 = json.dumps(asdict(report), indent=2, sort_keys=True)
json2 = json.dumps(asdict(report), indent=2, sort_keys=True)
assert json1 == json2
# Verify sorting by filename (dataset key) is deterministic
records = report.records
assert records[0].dataset == "A_DATASET"
assert records[1].dataset == "B_DATASET"
def test_does_not_open_example_files(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Verification does not open/read example files inside the cache."""
repo_root = tmp_path
manifests_dir = tmp_path / "manifests"
manifests_dir.mkdir()
(repo_root / ".data" / "benchmarks").mkdir(parents=True)
cache_dir = repo_root / ".data" / "benchmarks" / "test"
cache_dir.mkdir(parents=True)
sentinel_file = cache_dir / "example.jsonl"
sentinel_file.write_text("dummy content", encoding="utf-8")
write_test_manifest(manifests_dir, dataset="TEST_SENTINEL")
read_paths: list[Path] = []
# Spy on Path read methods
orig_read_text = Path.read_text
def mock_read_text(self: Path, *args: any, **kwargs: any) -> str:
read_paths.append(self)
return orig_read_text(self, *args, **kwargs)
orig_read_bytes = Path.read_bytes
def mock_read_bytes(self: Path, *args: any, **kwargs: any) -> bytes:
read_paths.append(self)
return orig_read_bytes(self, *args, **kwargs)
orig_open = Path.open
def mock_open(self: Path, *args: any, **kwargs: any) -> any:
read_paths.append(self)
return orig_open(self, *args, **kwargs)
monkeypatch.setattr(Path, "read_text", mock_read_text)
monkeypatch.setattr(Path, "read_bytes", mock_read_bytes)
monkeypatch.setattr(Path, "open", mock_open)
# Patch builtin open
orig_builtin_open = builtins.open
def mock_builtin_open(file: any, *args: any, **kwargs: any) -> any:
if isinstance(file, (str, Path)):
read_paths.append(Path(file))
return orig_builtin_open(file, *args, **kwargs)
monkeypatch.setattr(builtins, "open", mock_builtin_open)
verify_local_generalization_cache(
repo_root=repo_root, manifests_dir=manifests_dir, require_present=False
)
# Ensure no read path resides inside the cache directory
for p in read_paths:
resolved_p = p.resolve()
assert not resolved_p.is_relative_to(
cache_dir.resolve()
), f"Read file inside cache: {resolved_p}"
def test_cli_subprocess_metadata_only_json() -> None:
"""Run CLI in metadata-only and json mode via subprocess against actual manifests."""
result = subprocess.run(
[
sys.executable,
"scripts/benchmarks/verify_generalization_cache.py",
"--metadata-only",
"--json",
],
capture_output=True,
text=True,
check=True,
)
assert result.returncode == 0
report = json.loads(result.stdout)
assert report["policy_version"] == GENERALIZATION_CACHE_VERIFIER_POLICY_VERSION
assert isinstance(report["records"], list)
assert len(report["records"]) > 0