1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
|
from pathlib import Path
import tempfile
import threading
import unittest
from parley.config import Config
from parley.controller import Controller
from parley.state import State
class FakeRecorder:
def __init__(self) -> None:
self.path = None
self.cancelled = False
def start(self, path: Path) -> None:
self.path = path
path.write_bytes(b"RIFF")
def stop(self) -> Path:
return self.path
def cancel(self) -> None:
self.cancelled = True
if self.path:
self.path.unlink(missing_ok=True)
class FakeTranscriber:
def transcribe(self, wav_path: Path, workspace: Path) -> str:
return "hello from controller"
def cancel(self) -> None:
pass
class ControllerTests(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
root = Path(self.temporary.name)
ffmpeg = root / "ffmpeg"
runtime = root / "transcribe-cli"
model = root / "model.gguf"
for path in (ffmpeg, runtime):
path.write_text("#!/bin/sh\n")
path.chmod(0o755)
model.touch()
self.config = Config(ffmpeg, runtime, model, root / "transcripts")
def tearDown(self) -> None:
self.temporary.cleanup()
def test_successful_operation_persists_before_callback(self) -> None:
ready = threading.Event()
received = []
controller = Controller(
self.config,
on_transcript=lambda text, path: (
received.append((text, path)),
ready.set(),
),
)
controller.recorder = FakeRecorder()
controller.transcriber = FakeTranscriber()
try:
controller.start()
self.assertEqual(controller.state, State.RECORDING)
controller.stop_and_transcribe()
self.assertTrue(ready.wait(2), "transcription callback was not delivered")
self.assertEqual(controller.state, State.IDLE)
text, path = received[0]
self.assertEqual(text, "hello from controller")
self.assertEqual(path.read_text(), "hello from controller\n")
self.assertEqual(path.with_suffix(".wav").read_bytes(), b"RIFF")
finally:
controller.close()
def test_cancel_recording_discards_operation(self) -> None:
controller = Controller(self.config)
recorder = FakeRecorder()
controller.recorder = recorder
try:
controller.start()
controller.cancel()
self.assertEqual(controller.state, State.IDLE)
self.assertTrue(recorder.cancelled)
finally:
controller.close()
if __name__ == "__main__":
unittest.main()
|
