"""Structural synthetic protocol tests; these are not ASR/retrieval proof.""" import json from pathlib import Path import unittest import tempfile import sys import os import wave import importlib.util import hashlib from run import segments, run_owned, recognized_window, decode_command, decoded_frames, primary_audio_range from score import score, require_eligible class ContractTests(unittest.TestCase): def test_harness_source_hash_is_captured_before_file_changes(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "run-copy.py" original = Path(__file__).with_name("run.py").read_bytes() path.write_bytes(original) spec = importlib.util.spec_from_file_location("speech_start_hash_test", path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) path.write_bytes(original + b"\n# changed after startup\n") self.assertEqual(module.HARNESS_SHA256_AT_START, hashlib.sha256(original).hexdigest()) self.assertNotEqual(module.HARNESS_SHA256_AT_START, module.sha256(path)) def test_scoring_rejects_incomplete_or_reduced_pool(self): for report in [{}, {"complete": False, "quality_score_eligible": True}, {"complete": True, "quality_score_eligible": False}, {"complete": True, "quality_score_eligible": True, "excluded_windows": [{}]}]: with self.assertRaises(ValueError): require_eligible(report) require_eligible({"complete": True, "quality_score_eligible": True, "excluded_windows": []}) def test_primary_audio_duration_and_nonzero_start_ignore_longer_video(self): probe = {"streams": [{"index": 0, "codec_type": "video", "duration": "90"}, {"index": 1, "codec_type": "audio", "start_time": "0.125", "duration": "31"}]} self.assertEqual(primary_audio_range(probe), (125, 31125, 1)) @unittest.skipUnless(os.environ.get("SEARCHI_TEST_FFMPEG"), "Requires verified CPU FFmpeg") def test_actual_video_longer_than_audio(self): ffmpeg = Path(os.environ["SEARCHI_TEST_FFMPEG"]) ffprobe = ffmpeg.with_name("ffprobe.exe" if os.name == "nt" else "ffprobe") with tempfile.TemporaryDirectory() as parent: scratch = Path(parent) source = scratch / "longer-video.mkv" run_owned([str(ffmpeg), "-nostdin", "-hide_banner", "-v", "error", "-threads", "1", "-f", "lavfi", "-i", "color=c=black:s=16x16:r=1:d=33", "-itsoffset", "0.125", "-f", "lavfi", "-i", "anullsrc=r=48000:cl=mono:d=31", "-c:v", "ffv1", "-c:a", "pcm_s16le", "-threads", "1", "-output_ts_offset", "5", "-y", str(source)], scratch, "mux", [(source, 4 * 1024 * 1024)], 30) run_owned([str(ffprobe), "-v", "error", "-show_entries", "stream=index,codec_type,duration,start_time:stream_tags=DURATION:format=duration", "-of", "json", str(source)], scratch, "probe", [], 30) probe = json.loads((scratch / "probe.log").read_text()) audio = next(row for row in probe["streams"] if row["codec_type"] == "audio") # Matroska stores duration in tags; test conversion is explicit, not a format fallback. if "duration" not in audio: h, m, sec = audio["tags"]["DURATION"].split(":") audio["duration"] = int(h) * 3600 + int(m) * 60 + float(sec) - float(audio["start_time"]) start, end, _ = primary_audio_range(probe) self.assertEqual((start, end), (5125, 36125)) self.assertGreater(float(probe["format"]["duration"]), end / 1000) wav = scratch / "tail.wav" run_owned(decode_command(ffmpeg, source, 35125, 1000, wav, audio["index"]), scratch, "tail", [(wav, 970000)], 30) self.assertGreaterEqual(decoded_frames(wav, 1000), 15984) @unittest.skipUnless(os.environ.get("SEARCHI_TEST_FFMPEG"), "Requires verified CPU FFmpeg") def test_actual_ffmpeg_full_window_and_partial_tail_sample_coverage(self): ffmpeg = Path(os.environ["SEARCHI_TEST_FFMPEG"]) with tempfile.TemporaryDirectory() as parent: scratch = Path(parent) for rate in [44100, 48000]: with self.subTest(rate=rate): source = scratch / f"source-{rate}.wav" with wave.open(str(source), "wb") as audio: audio.setnchannels(1) audio.setsampwidth(2) audio.setframerate(rate) audio.writeframes(b"\x00\x00" * (rate * 31)) for offset, length, expected in [(0, 30000, 480000), (30000, 1000, 16000)]: output = scratch / "decoded.wav" run_owned(decode_command(ffmpeg, source, offset, length, output), scratch, "samplecoverage", [(output, 970000)], 30) self.assertEqual(decoded_frames(output, length), expected) def test_threefold_truncation_rejected(self): with tempfile.TemporaryDirectory() as parent: path = Path(parent) / "truncated.wav" with wave.open(str(path), "wb") as audio: audio.setnchannels(1) audio.setsampwidth(2) audio.setframerate(16000) audio.writeframes(b"\x00\x00" * 160000) with self.assertRaises(ValueError): decoded_frames(path, 30000) def test_invalid_actual_json_is_retained_and_next_window_parses(self): invalid = {"transcription": [{"offsets": {"from": 0, "to": 30010}, "text": "synthetic test"}]} valid = {"transcription": [{"offsets": {"from": 0, "to": 10}, "text": "synthetic test"}]} rows = [recognized_window(raw, offset, 30000, 1, 2) for raw, offset in [(invalid, 0), (valid, 30000)]] self.assertEqual(rows[0]["actual_whisper_json"], invalid) self.assertEqual(rows[0]["segments"], []) self.assertIn("validation_error", rows[0]) self.assertNotIn("validation_error", rows[1]) self.assertEqual(rows[1]["segments"][0]["start_ms"], 30000) def test_timeout_and_output_overflow_terminate_owned_child(self): for name, code, error, timeout in [ ("timeout", "import time; time.sleep(10)", TimeoutError, 0.1), ("overflow", "import sys,time; sys.stdout.write('x'*5000000); sys.stdout.flush(); time.sleep(10)", ValueError, 5), ]: with self.subTest(name=name), tempfile.TemporaryDirectory() as parent: checkpoint = Path(parent) / "checkpoint.json" checkpoint.write_text('{"complete":false,"rows":[1]}') with tempfile.TemporaryDirectory(dir=parent) as scratch: with self.assertRaises(error): run_owned([sys.executable, "-c", code], Path(scratch), name, [], timeout) diagnostics = list(Path(parent).glob(f"speech-{name}-failure-*.json")) self.assertEqual(len(diagnostics), 1) report = json.loads(diagnostics[0].read_text()) self.assertIsNotNone(report["exit_code"]) self.assertEqual(report["error_type"], error.__name__) self.assertLessEqual(Path(report["bounded_log"]).stat().st_size, 4 * 1024 * 1024) self.assertEqual(json.loads(checkpoint.read_text())["rows"], [1]) def test_failure_diagnostic_survives_scratch_cleanup(self): with tempfile.TemporaryDirectory() as parent: with tempfile.TemporaryDirectory(dir=parent) as scratch: with self.assertRaises(RuntimeError): run_owned([sys.executable, "-c", "print('structural failure'); raise SystemExit(3)"], Path(scratch), "contract", [], 5) diagnostics = list(Path(parent).glob("speech-contract-failure-*.json")) self.assertEqual(len(diagnostics), 1) report = json.loads(diagnostics[0].read_text()) self.assertEqual(report["exit_code"], 3) self.assertGreater(report["elapsed_ms"], 0) self.assertIn("structural failure", Path(report["bounded_log"]).read_text()) def test_offsets_applied_once(self): actual = segments({"transcription": [{"offsets": {"from": 1500, "to": 2400}, "text": "test"}]}, 30000, 30000) self.assertEqual(actual[0]["start_ms"], 31500) self.assertEqual(actual[0]["end_ms"], 32400) def test_reject_invalid_times_and_text(self): for start, end, text in [(False, 2, "x"), (-1, 2, "x"), (2, 1, "x"), (0, 30001, "x"), (0, 1, "x" * 4097)]: with self.assertRaises(ValueError): segments({"transcription": [{"offsets": {"from": start, "to": end}, "text": text}]}, 0, 30000) def test_reject_nonmonotonic_times(self): with self.assertRaises(ValueError): segments({"transcription": [{"offsets": {"from": start, "to": start + 1}, "text": "x"} for start in [10, 2]]}, 0, 30000) def test_missing_rankings_fail(self): fixture = json.loads((Path(__file__).resolve().parents[2] / "fixtures/media/fixture.json").read_text()) with self.assertRaises(ValueError): score(fixture, {}) def test_point_is_distinct_from_overlap(self): fixture = json.loads((Path(__file__).resolve().parents[2] / "fixtures/media/fixture.json").read_text()) queries = [query for query in fixture["queries"] if query["category"] == "video-speech"] rankings = {query["id"]: [{"asset": "clip04", "start_ms": 0, "end_ms": 100000}] for query in queries} report = score(fixture, rankings) self.assertEqual(report["point_hits"], 0) self.assertEqual(report["overlap_hits"], 8) if __name__ == "__main__": unittest.main()