115 lines
6.2 KiB
Python
115 lines
6.2 KiB
Python
"""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
|
|
from run import segments, run_owned, recognized_window, decode_command, decoded_frames
|
|
from score import score
|
|
|
|
|
|
class ContractTests(unittest.TestCase):
|
|
@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()
|