Skip to content

Commit 2101d75

Browse files
committed
test: cover transcript state behavior
1 parent e02fcd6 commit 2101d75

1 file changed

Lines changed: 74 additions & 0 deletions

File tree

tests/test_audio_transcriber.py

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
import time
2+
import unittest
3+
import sys
4+
import types
5+
from datetime import datetime
6+
7+
sys.modules.setdefault("pyaudiowpatch", types.SimpleNamespace(paInt16=8, PyAudio=object))
8+
9+
from AudioTranscriber import AudioTranscriber, TRANSLATION_SILENCE_DELAY
10+
11+
12+
class FakeSource:
13+
SAMPLE_RATE = 16000
14+
SAMPLE_WIDTH = 2
15+
channels = 1
16+
17+
18+
class FakeModel:
19+
def translate_to_indonesian(self, text):
20+
return f"ID: {text}"
21+
22+
23+
class AudioTranscriberTests(unittest.TestCase):
24+
def make_transcriber(self):
25+
return AudioTranscriber(FakeSource(), FakeSource(), FakeModel())
26+
27+
def wait_for_entry(self, transcriber, predicate, timeout=1.0):
28+
deadline = time.monotonic() + timeout
29+
while time.monotonic() < deadline:
30+
entries = transcriber.get_entries()
31+
if entries and predicate(entries[0]):
32+
return entries[0]
33+
time.sleep(0.01)
34+
self.fail("Timed out waiting for transcript entry state")
35+
36+
def test_merge_transcript_keeps_overlapping_words_once(self):
37+
merged = AudioTranscriber._merge_transcript(
38+
"hello this is a live transcription",
39+
"live transcription test",
40+
)
41+
42+
self.assertEqual(merged, "hello this is a live transcription test")
43+
44+
def test_finalize_segment_translates_after_silence(self):
45+
transcriber = self.make_transcriber()
46+
transcriber.update_transcript("Speaker", "hello world", datetime.now())
47+
48+
with transcriber.lock:
49+
segment = transcriber.segment_state["Speaker"]
50+
segment["last_activity"] = time.monotonic() - TRANSLATION_SILENCE_DELAY - 0.1
51+
generation = segment["timer_generation"]
52+
53+
transcriber.finalize_segment("Speaker", generation)
54+
55+
entry = self.wait_for_entry(
56+
transcriber,
57+
lambda item: not item["translation_pending"],
58+
)
59+
self.assertEqual(entry["translation"], "ID: hello world")
60+
self.assertEqual(entry["translation_state"], "done")
61+
62+
def test_clear_transcript_resets_public_buffers(self):
63+
transcriber = self.make_transcriber()
64+
transcriber.update_transcript("You", "testing one two", datetime.now())
65+
66+
transcriber.clear_transcript_data()
67+
68+
self.assertEqual(transcriber.get_entries(), [])
69+
self.assertEqual(transcriber.mic_buffer, "")
70+
self.assertEqual(transcriber.speaker_buffer, "")
71+
72+
73+
if __name__ == "__main__":
74+
unittest.main()

0 commit comments

Comments
 (0)