74 lines
3.5 KiB
Python
74 lines
3.5 KiB
Python
"""Planning/provenance regressions without loading inference models."""
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import gated_asr as gated
|
|
|
|
|
|
class GatedPlanning(unittest.TestCase):
|
|
def test_frozen_dtw_combination_preserves_sources_and_component_hashes(self):
|
|
plan, digest = gated.dtw_contract()
|
|
self.assertEqual(plan["sources"], list(gated.speech.ASSETS))
|
|
self.assertEqual(plan["vad_report_sha256"], gated.VAD_SHA)
|
|
self.assertEqual(len(digest), 64)
|
|
with patch.object(gated.speech, "sha256", return_value="changed"):
|
|
with self.assertRaises(ValueError):
|
|
gated.dtw_contract()
|
|
def test_absolute_offset_and_contiguous_split(self):
|
|
interval = gated.vad.source_intersection([10, 6110], 1000000, 5125)
|
|
windows = gated.interval_windows([interval], 1000000, 5125)
|
|
self.assertEqual([(w["offset_ms"], w["duration_ms"]) for w in windows],
|
|
[(5225, 30000), (35225, 30000), (65225, 1000)])
|
|
self.assertTrue(all(w["vad_source_intersection"] == interval for w in windows))
|
|
|
|
def test_disjoint_intervals_are_not_concatenated(self):
|
|
intervals = [gated.vad.source_intersection(raw, 16000, 0) for raw in ([0, 20], [80, 90])]
|
|
self.assertEqual([(w["offset_ms"], w["duration_ms"]) for w in gated.interval_windows(intervals, 16000, 0)],
|
|
[(0, 200), (800, 100)])
|
|
|
|
def test_tail_fraction_and_raw_padding_remain_explicit(self):
|
|
interval = gated.vad.source_intersection([90, 102], 16001, 0)
|
|
window, = gated.interval_windows([interval], 16001, 0)
|
|
self.assertEqual(window["duration_ms"], 100)
|
|
self.assertEqual(window["unrepresented_tail_samples"], 1)
|
|
self.assertEqual(window["vad_source_intersection"]["tail_padding_samples"], 319)
|
|
|
|
def test_overlap_and_tampered_mapping_fail(self):
|
|
intervals = [gated.vad.source_intersection(raw, 16000, 0) for raw in ([0, 50], [40, 60])]
|
|
with self.assertRaises(ValueError):
|
|
gated.interval_windows(intervals, 16000, 0)
|
|
intervals[0]["source_absolute_ms"][0] = 1
|
|
with self.assertRaises(ValueError):
|
|
gated.interval_windows(intervals[:1], 16000, 0)
|
|
|
|
def test_empty_source_intersection_has_no_asr_window(self):
|
|
interval = gated.vad.source_intersection([101, 102], 16001, 0)
|
|
self.assertEqual(gated.interval_windows([interval], 16001, 0), [])
|
|
|
|
def test_frozen_real_evidence_is_complete_and_tamper_rejected(self):
|
|
root = Path(__file__).resolve().parents[2]
|
|
path = root / ".searchi/speech/vad-v6-results.json"
|
|
if not path.exists():
|
|
self.skipTest("Local frozen diagnostic evidence unavailable")
|
|
self.assertEqual(gated.speech.sha256(path), gated.VAD_SHA)
|
|
report = json.loads(path.read_bytes())
|
|
fixture = json.loads((root / "fixtures/media/fixture.json").read_bytes())
|
|
self.assertEqual(len(gated.validated_rows(report, fixture)), 5)
|
|
for mutate in (
|
|
lambda r: r["rows"].pop(),
|
|
lambda r: r["rows"][0]["native"].update(gpu=True),
|
|
lambda r: r["rows"][0]["native"]["probabilities"].pop(),
|
|
lambda r: r["rows"][0].update(source_sha256="bad"),
|
|
):
|
|
changed = copy.deepcopy(report)
|
|
mutate(changed)
|
|
with self.assertRaises(ValueError):
|
|
gated.validated_rows(changed, fixture)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|