Spaces:
Running
Running
File size: 3,457 Bytes
62f52ac | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | """CI validation for the Benchmark + Engine registries (Component 7).
Runs in GitHub Actions after the deterministic pytest suite. Kept dependency-free
(no supabase client) so it fails fast on structural drift:
- every catalog benchmark carries required registry fields (incl. C6 depth),
- benchmark names are unique across the catalog,
- every benchmark section is a known result section,
- every registered engine describes() a complete contract,
- every engine-declared benchmark reference exists in the catalog.
"""
from __future__ import annotations
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from app.engines import ENGINES
from app.services.benchmarks import _RESULT_SECTIONS, load_benchmark_files
REQUIRED_RECORD_FIELDS = (
"name", "category", "description", "input", "expected_output",
"tolerance", "ground_truth", "citation", "source", "stage",
"difficulty", "version",
)
DESCRIBE_KEYS = (
"name", "version", "tool", "tool_version", "databases", "parameters",
"citations", "benchmarks", "export_formats", "figure_formats",
)
# Pipeline result sections plus standalone tool sections used by the catalog.
KNOWN_SECTIONS = set(_RESULT_SECTIONS) | {"primers", "ngs", "docking", "md", "admet", "stats"}
def check_catalog() -> list[str]:
errors: list[str] = []
records = load_benchmark_files()
if not records:
return ["catalog is empty"]
names: list[str] = []
for r in records:
for field in REQUIRED_RECORD_FIELDS:
if field not in r or r.get(field) is None:
errors.append(f"{r.get('name', '<unnamed>')}: missing '{field}'")
if r.get("difficulty") not in {"easy", "medium", "hard"}:
errors.append(f"{r.get('name')}: bad difficulty {r.get('difficulty')!r}")
if isinstance(r.get("version"), int) and r["version"] < 1:
errors.append(f"{r.get('name')}: bad version {r.get('version')}")
if r.get("section") and r["section"] not in KNOWN_SECTIONS:
errors.append(f"{r.get('name')}: unknown section {r.get('section')}")
names.append(r.get("name", ""))
dups = {n for n in names if names.count(n) > 1}
if dups:
errors.append(f"duplicate benchmark names: {sorted(dups)}")
return errors
def check_engines() -> list[str]:
errors: list[str] = []
if not ENGINES:
return ["engine registry is empty"]
catalog_names = {r["name"] for r in load_benchmark_files()}
for name, engine in ENGINES.items():
if name != engine.name:
errors.append(f"registry key {name} != engine.name {engine.name}")
desc = engine.describe()
for key in DESCRIBE_KEYS:
if key not in desc:
errors.append(f"{name}: describe() missing '{key}'")
if not engine.export_formats or "json" not in engine.export_formats:
errors.append(f"{name}: json export unsupported")
for bm in engine.benchmarks:
if bm not in catalog_names:
errors.append(f"{name}: benchmark ref '{bm}' not in catalog")
return errors
def main() -> int:
errors = check_catalog() + check_engines()
if errors:
print("VALIDATION FAILED:")
for e in errors:
print(f" - {e}")
return 1
print("VALIDATION OK: catalog + engine registries consistent")
return 0
if __name__ == "__main__":
sys.exit(main()) |