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
|
from pathlib import Path
import tempfile
import unittest
from unittest.mock import MagicMock, patch
from parley.config import Config
from parley.errors import RecordingError
from parley.recording import Recorder
class RecorderTests(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
root = Path(self.temporary.name)
self.output = root / "work" / "audio.wav"
self.config = Config(
ffmpeg=Path("/usr/bin/ffmpeg"),
transcribe_cli=root / "transcribe-cli",
model=root / "model.gguf",
transcript_dir=root / "transcripts",
)
def tearDown(self) -> None:
self.temporary.cleanup()
@patch("parley.recording.subprocess.Popen")
def test_start_and_stop(self, popen: MagicMock) -> None:
process = popen.return_value
process.poll.return_value = None
process.communicate.return_value = (b"", b"")
process.returncode = 0
recorder = Recorder(self.config)
recorder.start(self.output)
self.output.write_bytes(b"RIFF")
result = recorder.stop()
self.assertEqual(result, self.output)
self.assertFalse(recorder.is_recording)
process.communicate.assert_called_once_with(input=b"q\n", timeout=5.0)
command = popen.call_args.args[0]
self.assertIn("pulse", command)
self.assertIn("16000", command)
@patch("parley.recording.subprocess.Popen")
def test_cancel_discards_output(self, popen: MagicMock) -> None:
process = popen.return_value
process.communicate.return_value = (b"", b"")
process.returncode = 0
recorder = Recorder(self.config)
recorder.start(self.output)
self.output.write_bytes(b"partial")
recorder.cancel()
self.assertFalse(self.output.exists())
self.assertFalse(recorder.is_recording)
def test_stop_without_start_is_an_error(self) -> None:
with self.assertRaisesRegex(RecordingError, "no recording"):
Recorder(self.config).stop()
if __name__ == "__main__":
unittest.main()
|
