from __future__ import annotations import errno import hashlib import json import shutil import subprocess import sys import tempfile import unittest from pathlib import Path from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] class ReleaseGuardrailTest(unittest.TestCase): def _onnx_runtime_python(self) -> str | None: candidates = [sys.executable, str(ROOT / ".cache" / "venv313" / "bin" / "python")] for candidate in candidates: if not Path(candidate).is_file(): continue completed = subprocess.run( [candidate, "-c", "import numpy, onnxruntime"], cwd=ROOT, check=False, capture_output=True, text=True, ) if completed.returncode == 0: return candidate return None def test_fake_onnx_is_rejected_by_runtime_validation(self) -> None: from scripts.build_release import _validate_onnx_runtime with tempfile.TemporaryDirectory() as directory: model = Path(directory) / "fake.onnx" model.write_bytes(b"not-an-onnx-model") metadata = { "frontend": { "max_seconds": 1.0, "sample_rate": 16_000, "hop_length": 160, "n_mels": 80, }, "input_features_name": "log_mel", "frame_mask_name": "frame_mask", "endpoint_output_name": "endpoint_probability", } with self.assertRaises(SystemExit): _validate_onnx_runtime( model, metadata, { "input_names": ["log_mel", "frame_mask"], "output_names": ["endpoint_probability"], }, ) def test_release_swap_retries_transient_nonempty_backup_cleanup(self) -> None: from scripts.build_release import _replace_release_directory with tempfile.TemporaryDirectory() as directory: root = Path(directory) destination = root / "release" staged = root / "staged" destination.mkdir() staged.mkdir() (destination / "old.txt").write_text("old", encoding="utf-8") (staged / "new.txt").write_text("new", encoding="utf-8") real_rmtree = shutil.rmtree calls = 0 def flaky_rmtree(path: Path) -> None: nonlocal calls calls += 1 if calls == 1: raise OSError(errno.ENOTEMPTY, "simulated Finder race") real_rmtree(path) with patch("scripts.build_release.shutil.rmtree", side_effect=flaky_rmtree): _replace_release_directory(staged, destination) self.assertEqual(calls, 2) self.assertTrue((destination / "new.txt").is_file()) self.assertFalse((destination / "old.txt").exists()) def test_space_card_rejects_server_invalid_frontmatter(self) -> None: from scripts.build_release import ( _validate_space_app, _validate_space_card, _validate_space_requirements, ) valid = """--- title: Test Space colorFrom: yellow colorTo: indigo sdk: gradio python_version: "3.12" app_file: app.py short_description: Tiny endpoint detector --- """ with tempfile.TemporaryDirectory() as directory: card = Path(directory) / "README.md" card.write_text(valid, encoding="utf-8") _validate_space_card(card) card.write_text( valid.replace("colorFrom: yellow", "colorFrom: orange"), encoding="utf-8", ) with self.assertRaisesRegex(SystemExit, "colorFrom"): _validate_space_card(card) card.write_text( valid.replace("Tiny endpoint detector", "x" * 61), encoding="utf-8", ) with self.assertRaisesRegex(SystemExit, "at most 60"): _validate_space_card(card) card.write_text(valid.replace('python_version: "3.12"\n', ""), encoding="utf-8") with self.assertRaisesRegex(SystemExit, "python_version"): _validate_space_card(card) requirements = Path(directory) / "requirements.txt" requirements.write_text( "# ZeroGPU CPython 3.12\nnumpy==2.3.5\nonnxruntime==1.26.0\n" "soundfile==0.14.0\nspaces==0.51.1\n", encoding="utf-8", ) _validate_space_requirements(requirements) requirements.write_text("numpy==2.2.6\n", encoding="utf-8") with self.assertRaisesRegex(SystemExit, "CPython-3.12-compatible"): _validate_space_requirements(requirements) app = Path(directory) / "app.py" valid_app = """ try: import spaces except ModuleNotFoundError: spaces = None @spaces.GPU(duration=10) def analyze_turn(audio): return audio analyze.click(fn=analyze_turn, inputs=[], outputs=[]) """ app.write_text(valid_app, encoding="utf-8") _validate_space_app(app) app.write_text(valid_app.replace("@spaces.GPU(duration=10)\n", ""), encoding="utf-8") with self.assertRaisesRegex(SystemExit, "@spaces.GPU"): _validate_space_app(app) def test_synthetic_replay_must_be_labelled_and_hash_bound(self) -> None: from scripts.build_release import _validate_synthetic_replay controller = { "endpoint_threshold": 0.7, "long_pause_threshold": 0.5, "min_silence_ms": 200, "relax_after_ms": 800, "max_silence_ms": 1800, "required_confirmations": 1, } with tempfile.TemporaryDirectory() as directory: root = Path(directory) fixture = root / "fixture.jsonl" decisions = root / "decisions.jsonl" summary_path = root / "summary.json" fixture.write_text("fixture\n", encoding="utf-8") decisions.write_text("decision\n", encoding="utf-8") def evidence(path: Path) -> dict[str, int | str]: return { "bytes": path.stat().st_size, "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), } summary = { "format_version": 1, "evidence_scope": "synthetic_integration", "controller_config": controller, "input": evidence(fixture), "decisions": evidence(decisions), "duplicate_response_emissions": 0, } summary_path.write_text(json.dumps(summary), encoding="utf-8") _validate_synthetic_replay(fixture, decisions, summary_path, controller) summary["evidence_scope"] = "unqualified_sequence" summary_path.write_text(json.dumps(summary), encoding="utf-8") with self.assertRaisesRegex(SystemExit, "not labelled synthetic"): _validate_synthetic_replay(fixture, decisions, summary_path, controller) def test_benchmark_must_bind_released_artifact_and_scope(self) -> None: from scripts.build_release import _validate_benchmark evidence = {"bytes": 123, "sha256": "a" * 64} with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "onnx_benchmark.json" report = { "artifact_bytes": 123, "artifact_sha256": "a" * 64, "scope": "neural_model_only_log_mel_input", "threads": 1, "batch_size": 1, "measured_iterations": 200, "warm_latency_ms": {"p95": 1.0}, } path.write_text(json.dumps(report), encoding="utf-8") _validate_benchmark( path, evidence, expected_scope="neural_model_only_log_mel_input", ) report["artifact_sha256"] = "b" * 64 path.write_text(json.dumps(report), encoding="utf-8") with self.assertRaisesRegex(SystemExit, "stale"): _validate_benchmark(path, evidence) def test_final_release_rejects_empty_or_incomplete_official_metrics(self) -> None: from scripts.build_release import _validate_metrics metadata = { "development_only": False, "training_status": "final", "data_scope": "full train", "data_revision": "train-revision", "threshold": 0.7, } export_manifest = {"checkpoint": {"sha256": "a" * 64}} shell = { "split": "test", "official_test": True, "dataset_revision": "0500378e8ed6d38e37b016e24d261e8e6c6a6859", "freeze_manifest_sha256": "b" * 64, "development_only": False, "training_status": "final", "data_scope": "full train", "data_revision": "train-revision", "threshold": 0.7, "checkpoint_sha256": "a" * 64, } with self.assertRaisesRegex(SystemExit, "no measured metrics object"): _validate_metrics( shell, metadata, export_manifest, development_only=False, ) shell["metrics"] = { "count": 1, "positive_count": 1, "negative_count": 0, "tp": 1, "fp": 0, "tn": 0, "fn": 0, "threshold": 0.7, "roc_auc": 0.5, "average_precision": 1.0, "brier_score": 0.1, "log_loss": 0.2, } with self.assertRaisesRegex(SystemExit, "two-class evaluation"): _validate_metrics( shell, metadata, export_manifest, development_only=False, ) def test_development_metrics_are_never_labelled_as_test_metrics(self) -> None: from scripts.build_release import _deployment_source_paths runtime_python = self._onnx_runtime_python() fixture = ROOT / "artifacts" / "smoke" / "model.onnx" if runtime_python is None or not fixture.is_file(): self.skipTest("ONNX Runtime and the generated smoke fixture are required") with tempfile.TemporaryDirectory(prefix="release-guardrail-", dir=ROOT) as directory: working = Path(directory) model = working / "preview.onnx" metadata = working / "model_metadata.json" export_manifest = working / "export_manifest.json" resolved_config = working / "resolved_config.json" split_manifest = working / "split.jsonl" metrics = working / "validation_metrics.json" output = working / "release" shutil.copy2(fixture, model) model_sha256 = hashlib.sha256(model.read_bytes()).hexdigest() checkpoint_sha256 = "a" * 64 resolved_config.write_text("{}", encoding="utf-8") split_manifest.write_text('{"split": "train"}\n', encoding="utf-8") def evidence(path: Path) -> dict[str, int | str]: return { "path": path.relative_to(ROOT).as_posix(), "bytes": path.stat().st_size, "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), } metadata.write_text( json.dumps( { "model_name": "preview", "architecture": "tiny_tcn", "development_only": True, "training_status": "preview-only", "data_scope": "one shard", "data_revision": "abc123", "threshold": 0.73, "controller": { "endpoint_threshold": 0.73, "long_pause_threshold": 0.6, "min_silence_ms": 250.0, "relax_after_ms": 700.0, "max_silence_ms": 1800.0, "required_confirmations": 2, }, "parameter_count": 10, "frontend": { "sample_rate": 16000, "hop_length": 160, "n_mels": 80, "max_seconds": 1.0, }, "input_features_name": "log_mel", "frame_mask_name": "frame_mask", "endpoint_output_name": "endpoint_probability", "output_type": "probability", } ), encoding="utf-8", ) source_files = [evidence(path) for path in _deployment_source_paths()] source_files = sorted(source_files, key=lambda item: str(item["path"])) source_inventory_sha256 = hashlib.sha256( json.dumps(source_files, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() export_manifest.write_text( json.dumps( { "format_version": 2, "task": "audio-turn-end-detection", "model_type": "tiny_tcn", "input_names": ["log_mel", "frame_mask"], "output_names": ["endpoint_probability"], "files": { "fp32": { "bytes": model.stat().st_size, "sha256": model_sha256, } }, "parity": {"fp32_max_abs_error": 0.0}, "checkpoint": {"sha256": checkpoint_sha256}, "threshold": 0.73, "controller": { "endpoint_threshold": 0.73, "long_pause_threshold": 0.6, "min_silence_ms": 250.0, "relax_after_ms": 700.0, "max_silence_ms": 1800.0, "required_confirmations": 2, }, "parameter_count": 10, "resolved_config": evidence(resolved_config), "source_files": source_files, "source_inventory_sha256": source_inventory_sha256, "training_data": { "revision": "abc123", "scope": "one shard", "sources": { "train_source": evidence(split_manifest), "validation_source": evidence(split_manifest), }, }, "development_only": True, "training_status": "preview-only", "data_scope": "one shard", "data_revision": "abc123", } ), encoding="utf-8", ) metrics.write_text( json.dumps( { "split": "validation", "development_only": True, "training_status": "preview-only", "data_scope": "one shard", "data_revision": "abc123", "threshold": 0.73, "checkpoint_sha256": checkpoint_sha256, "metrics": { "count": 4, "positive_count": 2, "negative_count": 2, "tp": 1, "fp": 0, "tn": 2, "fn": 1, "threshold": 0.73, "roc_auc": 0.75, "average_precision": 0.8, "brier_score": 0.2, "log_loss": 0.5, }, } ), encoding="utf-8", ) command = [ runtime_python, str(ROOT / "scripts/build_release.py"), "--model", str(model), "--metadata", str(metadata), "--metrics", str(metrics), "--output", str(output), "--allow-development-artifact", ] completed = subprocess.run( command, cwd=ROOT, check=False, capture_output=True, text=True ) self.assertEqual(completed.returncode, 0, completed.stderr) model_release = output / "model" self.assertTrue((model_release / "development_metrics.json").is_file()) self.assertFalse((model_release / "test_metrics.json").exists()) manifest = json.loads((output / "release_manifest.json").read_text(encoding="utf-8")) self.assertEqual(manifest["metrics_scope"], "development") self.assertEqual(manifest["metrics_file"], "development_metrics.json") self.assertFalse(manifest["has_test_metrics"]) publish_command = [ runtime_python, str(ROOT / "scripts/publish_hf.py"), "--release-dir", str(output), "--allow-development-release", ] validated = subprocess.run( publish_command, cwd=ROOT, check=False, capture_output=True, text=True ) self.assertEqual(validated.returncode, 0, validated.stderr) manifest_path = output / "release_manifest.json" original_manifest = manifest_path.read_text(encoding="utf-8") mutable_manifest = json.loads(original_manifest) mutable_manifest["development_only"] = False manifest_path.write_text(json.dumps(mutable_manifest), encoding="utf-8") misclassified = subprocess.run( publish_command, cwd=ROOT, check=False, capture_output=True, text=True ) self.assertNotEqual(misclassified.returncode, 0) self.assertIn("status disagrees", misclassified.stderr) manifest_path.write_text(original_manifest, encoding="utf-8") with (model_release / "model.onnx").open("ab") as handle: handle.write(b"tampered") rejected = subprocess.run( publish_command, cwd=ROOT, check=False, capture_output=True, text=True ) self.assertNotEqual(rejected.returncode, 0) self.assertIn("integrity validation", rejected.stderr) if __name__ == "__main__": unittest.main()