searchi/tools/speech-vad/test_native.py

"""Actual CPU model contract checks; explicit asset paths required."""
import hashlib
import json
import os
from pathlib import Path
import struct
import subprocess
import tempfile
import unittest


class NativeContract(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.worker = Path(os.environ["SEARCHI_VAD_DIAGNOSTIC"])
        cls.model = Path(os.environ["SEARCHI_VAD_MODEL"])
        if hashlib.sha256(cls.model.read_bytes()).hexdigest() != "2aa269b785eeb53a82983a20501ddf7c1d9c48e33ab63a41391ac6c9f7fb6987":
            raise ValueError("Expected frozen Silero v6.2.0 export")

    def run_pcm(self, data):
        with tempfile.TemporaryDirectory(prefix="searchi-vad-contract-") as scratch:
            pcm = Path(scratch) / "samples.f32"
            pcm.write_bytes(data)
            return subprocess.run([str(self.worker), str(self.model), str(pcm)],
                                  capture_output=True, timeout=30, check=False)

    def test_real_model_silence_tail(self):
        result = self.run_pcm(b"\0" * (16001 * 4))
        self.assertEqual(result.returncode, 0, result.stderr.decode("utf-8", errors="replace"))
        report = json.loads(result.stdout)
        self.assertEqual(report["samples"], 16001)
        self.assertEqual(len(report["probabilities"]), 32)
        self.assertEqual(report["intervals_cs"], [])
        self.assertEqual(report["threads"], 1)
        self.assertIs(report["gpu"], False)
        for key in ("load_ms", "detect_ms", "segment_ms"):
            self.assertGreaterEqual(report[key], 0)

    def test_invalid_pcm(self):
        for data in (b"", b"x", struct.pack("<f", float("nan")), struct.pack("<f", float("inf"))):
            with self.subTest(data=data):
                result = self.run_pcm(data)
                self.assertEqual(result.returncode, 2)
                self.assertEqual(result.stdout, b"")

    def test_unknown_arguments_fail(self):
        result = subprocess.run([str(self.worker), "--unknown"], capture_output=True, timeout=5)
        self.assertEqual(result.returncode, 2)
        self.assertEqual(result.stdout, b"")

    def test_missing_model_is_failure_without_report(self):
        with tempfile.TemporaryDirectory(prefix="searchi-vad-missing-") as scratch:
            pcm = Path(scratch) / "silence.f32"
            pcm.write_bytes(b"\0" * 64000)
            result = subprocess.run([str(self.worker), str(Path(scratch) / "missing.bin"), str(pcm)],
                                    capture_output=True, timeout=5)
            self.assertEqual(result.returncode, 3)
            self.assertEqual(result.stdout, b"")

    def test_oversized_pcm_rejected_before_model_loading(self):
        with tempfile.TemporaryDirectory(prefix="searchi-vad-large-") as scratch:
            pcm = Path(scratch) / "oversized.f32"
            with pcm.open("wb") as stream:
                stream.truncate(16000 * 600 * 4 + 4)
            result = subprocess.run([str(self.worker), "missing.bin", str(pcm)],
                                    capture_output=True, timeout=5)
            self.assertEqual(result.returncode, 2)
            self.assertEqual(result.stdout, b"")

    def test_finite_decoder_overshoot_is_preserved(self):
        # A DC input is not speech evidence. This checks only the input contract:
        # valid floating-point decoder output must not be clipped or rejected.
        result = self.run_pcm(struct.pack("<f", 1.01) * 16000)
        self.assertEqual(result.returncode, 0, result.stderr.decode("utf-8", errors="replace"))
        self.assertEqual(json.loads(result.stdout)["samples"], 16000)


if __name__ == "__main__":
    unittest.main()