tests/test_release_guardrails.py

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()