import contextlib import copy import io from pathlib import Path import tempfile import unittest import wave from chunks import CONTEXT, ENCODING_MARGIN, merge, plan, speaker_mapping, stitch from subtitles import convert def word(text, start, end=None): return {"word": text, "start": start, "end": start + 0.4 if end is None else end} def make_audio(path, duration, pauses=()): rate = 16000 samples = bytearray(b"\x10\x27" * round(rate * duration)) for start, end in pauses: first, last = round(start * rate) * 2, round(end * rate) * 2 samples[first:last] = bytes(last - first) with wave.open(str(path), "wb") as audio: audio.setparams((1, 2, rate, 0, "NONE", "not compressed")) audio.writeframes(samples) class ChunkPlanTests(unittest.TestCase): def planned(self, duration, pauses=(), limit=60): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "audio.wav" make_audio(path, duration, pauses) return plan(path, limit) def test_short_audio_is_one_upload(self): self.assertEqual(self.planned(20), [{"start": 0, "end": 20, "seam": 0}]) def test_continuous_speech_is_bounded_and_has_no_gaps(self): for duration in (59.9, 60, 70, 120, 140): with self.subTest(duration=duration): chunks = self.planned(duration) self.assertEqual(chunks[0]["start"], 0) self.assertEqual(chunks[-1]["end"], duration) for chunk in chunks: self.assertGreater(chunk["end"], chunk["start"]) self.assertLessEqual(chunk["end"] - chunk["start"], 60 - ENCODING_MARGIN + 1e-8) for previous, current in zip(chunks, chunks[1:]): self.assertAlmostEqual(previous["end"] - current["start"], CONTEXT * 2) self.assertAlmostEqual(current["seam"] - current["start"], CONTEXT) def test_uses_last_pause_before_limit(self): chunks = self.planned(130, [(30, 31), (53, 54)]) self.assertAlmostEqual(chunks[1]["seam"], 53.5) self.assertAlmostEqual(chunks[0]["end"], 56.5) self.assertAlmostEqual(chunks[1]["start"], 50.5) def test_short_quiet_sound_does_not_count_as_pause(self): chunks = self.planned(70, [(53, 53.1)]) self.assertAlmostEqual(chunks[1]["seam"], 56.9) def test_all_silence_still_makes_progress(self): chunks = self.planned(140, [(0, 140)]) self.assertEqual(chunks[-1]["end"], 140) self.assertLess(len(chunks), 10) self.assertTrue(all(0 < c["end"] - c["start"] <= 60 for c in chunks)) def test_invalid_limit_and_empty_audio(self): for limit in (0, 59, 1801, float("inf"), float("nan")): with self.subTest(limit=limit), self.assertRaises(ValueError): self.planned(1, limit=limit) with self.assertRaisesRegex(ValueError, "no samples"): self.planned(0) class ChunkMergeTests(unittest.TestCase): def test_speaker_mapping_uses_shared_words_even_when_provider_ids_swap(self): left = [dict(word(text, start), speaker=speaker) for text, start, speaker in ( ("one", 58, 0), ("two", 58.5, 0), ("three", 60, 1), ("four", 60.5, 1), )] right = [dict(w, speaker=1 - w["speaker"]) for w in left] self.assertEqual(speaker_mapping(left, right, 60), {1: 0, 0: 1}) def test_speaker_mapping_rejects_missing_sparse_and_conflicting_evidence(self): left = [dict(word(text, start), speaker=0) for text, start in ( ("one", 58), ("two", 58.5), ("three", 60), ("four", 60.5), )] self.assertEqual(speaker_mapping(left, [dict(left[0], speaker=5)], 60), {}) self.assertEqual(speaker_mapping(left, [dict(word("elsewhere", 90), speaker=0)], 60), {}) conflicting = [dict(w, speaker=i // 2) for i, w in enumerate(left)] self.assertEqual(speaker_mapping(left, conflicting, 60), {}) self.assertEqual(speaker_mapping(conflicting, left, 60), {}) repeated = [dict(word("yes", 60), speaker=0)] self.assertEqual(speaker_mapping(repeated, repeated * 2, 60), {}) def test_overlap_keeps_speaker_identity_through_three_requests(self): chunks = [{"start": 0, "end": 63, "seam": 0}, {"start": 57, "end": 120, "seam": 60}, {"start": 114, "end": 140, "seam": 117}] responses = [ {"words": [dict(word(text, start), speaker=7) for text, start in (("First", 1), ("shared", 59), ("words", 60))]}, {"words": [dict(word(text, start), speaker=2) for text, start in (("shared", 2), ("words", 3), ("more", 59), ("speech", 60))]}, {"words": [dict(word(text, start), speaker=9) for text, start in (("more", 2), ("speech", 3), ("Last", 20))]}, ] original = copy.deepcopy(responses) with contextlib.redirect_stderr(io.StringIO()) as errors: result = merge(chunks, responses) self.assertEqual(result["text"], "First shared words more speech Last") self.assertEqual({w["speaker"] for w in result["words"]}, {0}) self.assertEqual(errors.getvalue(), "") self.assertEqual(responses, original) def test_boundary_word_survives_once_despite_timestamp_drift(self): left = [word("before", 8), word("boundary,", 9.8, 10.4), word("after", 10.5)] right = [word("Boundary", 10.0, 10.6), word("after", 10.6), word("later", 14)] result = stitch(left, right, 10) self.assertEqual([w["word"] for w in result], ["before", "boundary,", "after", "later"]) def test_words_cut_at_upload_edges_are_discarded(self): left = [word("Before", 5), word("shared", 9), word("words", 10), word("incom-", 12.8)] right = [word("-fore", 7), word("shared", 9.1), word("words", 10.1), word("incomplete", 12.6)] result = stitch(left, right, 10) self.assertEqual([w["word"] for w in result], ["Before", "shared", "words", "incomplete"]) def test_intentional_repeated_words_are_not_removed(self): left = [word("yes", start, start + 0.1) for start in (9.4, 9.8, 10.2, 10.6)] right = copy.deepcopy(left) + [word("done", 14)] result = stitch(left, right, 10) self.assertEqual([w["word"] for w in result], ["yes"] * 4 + ["done"]) def test_identical_text_outside_overlap_is_not_deduplicated(self): left = [word("again", 1)] right = [word("again", 20)] self.assertEqual(stitch(left, right, 10), left + right) def test_no_match_uses_whole_words_and_warns(self): left = [word("before", 8), word("left", 9.5, 10.1), word("discard", 11)] right = [word("discard", 8), word("right", 10.1), word("after", 14)] with contextlib.redirect_stderr(io.StringIO()) as errors: result = stitch(left, right, 10) self.assertEqual([w["word"] for w in result], ["before", "left", "right", "after"]) self.assertIn("joining by timestamps", errors.getvalue()) def test_silent_overlap_does_not_warn(self): with contextlib.redirect_stderr(io.StringIO()) as errors: self.assertEqual(stitch([word("before", 1)], [word("after", 20)], 10), [word("before", 1), word("after", 20)]) self.assertEqual(errors.getvalue(), "") def test_offsets_three_chunks_and_keeps_srt_numbering(self): chunks = [{"start": 0, "end": 63, "seam": 0}, {"start": 57, "end": 120, "seam": 60}, {"start": 114, "end": 140, "seam": 117}] responses = [ {"words": [word("First.", 1), word("Join.", 59.8, 60.2)]}, {"words": [word("Join.", 2.9, 3.3), word("Second.", 20), word("Next.", 59.8, 60.2)]}, {"words": [word("Next.", 2.9, 3.3), word("Last.", 20)]}, ] original = copy.deepcopy(responses) result = merge(chunks, responses) self.assertEqual(result["text"], "First. Join. Second. Next. Last.") self.assertEqual([w["start"] for w in result["words"]], [1, 59.8, 77, 116.8, 134]) self.assertEqual(responses, original) srt = convert(result) self.assertIn("5\n00:02:14,000", srt) def test_regression_preserves_order_and_scopes_speakers_per_chunk(self): chunks = [{"start": 0, "end": 63, "seam": 0}, {"start": 57, "end": 100, "seam": 60}] first = dict(word("First", 10), speaker=0) second = dict(word("Second", 9), speaker=0) last = dict(word("Last", 20), speaker=0) result = merge(chunks, [{"words": [first, second]}, {"words": [last]}]) self.assertEqual([w["start"] for w in result["words"]], [10, 10, 77]) self.assertNotEqual(result["words"][0]["speaker"], result["words"][-1]["speaker"]) def test_missing_or_malformed_words_fail(self): chunks = [{"start": 0, "end": 30, "seam": 0}] for response in ({"text": "missing"}, {"text": "missing", "words": []}, {"text": "missing", "words": [word(" ", 1)]}, {"text": None, "words": []}, {"words": [word("invalid", 100)]}, {"words": [word("invalid", -1)]}, {"error": "bad response"}, []): with self.subTest(response=response), self.assertRaisesRegex(ValueError, "Chunk 1"): merge(chunks, [response]) with self.assertRaises(ValueError): merge(chunks, []) self.assertEqual(merge(chunks, [{"text": "", "words": []}])["text"], "") if __name__ == "__main__": unittest.main()