mirror of
https://github.com/yusufipk/dikte.git
synced 2026-09-11 10:56:10 +00:00
yusufipek
be09cc3ac6
Pressing the shortcut while a transcript was being transcribed or cleaned up did nothing, and the thought you had while waiting was lost. The microphone is free the moment a recording stops, so the next dictation can now be spoken at once; the pipeline queues it and each one is finished, pasted and reported in the order it was spoken. The corner indicator stays with the recording under way rather than being wiped by the previous run's progress, and a stop that lands behind an unfinished run says it is waiting its turn.
325 lines
14 KiB
Python
325 lines
14 KiB
Python
"""The dictation chain end to end, with every outside call faked.
|
||
|
||
This is the one test that says what a dictation actually does: what gets sent,
|
||
what gets pasted, what is written to the history, and what happens to the audio
|
||
afterwards. A pull request that reorders any of it shows up here.
|
||
"""
|
||
|
||
import contextlib
|
||
import io
|
||
import os
|
||
import threading
|
||
import unittest
|
||
from unittest import mock
|
||
|
||
from dikte import api
|
||
from dikte import assistant
|
||
from dikte import config as cfg
|
||
from dikte import paste
|
||
from dikte import worker
|
||
from tests.support import DikteTest, make_wav, speech
|
||
|
||
|
||
class Chain(DikteTest):
|
||
def setUp(self):
|
||
super().setUp()
|
||
self.conf = self.config(openai_api_key="sk-test",
|
||
openrouter_api_key="sk-or-test")
|
||
self.wav = make_wav(self.path("clip.wav"), speech(2.0))
|
||
# The levels a real recording of that length would have handed over.
|
||
self.rms = [0.0005] * 40 + [0.2] * 20
|
||
|
||
def run_chain(self, ask=False, paste_override=None, duration=2.0,
|
||
transcript="uh, book it for Thursday",
|
||
cleaned="Book it for Thursday.",
|
||
cleanup_error=None, answer=("Booked.", ""), rms=None,
|
||
clipboard=b"what was there before", paste_error=None):
|
||
pipeline = worker.Pipeline(self.conf)
|
||
done, failures, stages, cancels = [], [], [], []
|
||
pipeline.finished.connect(lambda *args: done.append(args))
|
||
pipeline.failed.connect(failures.append)
|
||
pipeline.stage.connect(stages.append)
|
||
pipeline.cancelled.connect(lambda: cancels.append(True))
|
||
|
||
cleanup = (mock.Mock(side_effect=cleanup_error) if cleanup_error
|
||
else mock.Mock(return_value=cleaned))
|
||
calls = {}
|
||
# The chain reports its own failures on stderr, which a test run has no
|
||
# use for.
|
||
with contextlib.redirect_stderr(io.StringIO()), \
|
||
mock.patch.object(api, "transcribe", return_value=transcript) as tr, \
|
||
mock.patch.object(api, "cleanup", cleanup), \
|
||
mock.patch.object(assistant, "ask", return_value=answer) as ask_call, \
|
||
mock.patch.object(paste, "copy") as copy, \
|
||
mock.patch.object(paste, "copy_bytes") as copy_bytes, \
|
||
mock.patch.object(paste, "press") as press, \
|
||
mock.patch.object(paste, "read_clipboard",
|
||
return_value=clipboard) as read_clipboard, \
|
||
mock.patch.object(worker.time, "sleep", lambda seconds: None):
|
||
press.side_effect = paste_error
|
||
calls = {"transcribe": tr, "cleanup": cleanup, "ask": ask_call,
|
||
"copy": copy, "copy_bytes": copy_bytes, "press": press,
|
||
"read_clipboard": read_clipboard}
|
||
pipeline._work(self.wav, duration,
|
||
self.rms if rms is None else rms, ask, paste_override)
|
||
return {"done": done, "failures": failures, "stages": stages,
|
||
"cancelled": cancels, **calls}
|
||
|
||
# ---- the ordinary run -------------------------------------------------
|
||
|
||
def test_a_dictation_is_transcribed_cleaned_copied_and_pasted(self):
|
||
run = self.run_chain()
|
||
self.assertEqual(run["failures"], [])
|
||
self.assertEqual(run["done"][0],
|
||
("uh, book it for Thursday", "Book it for Thursday.", ""))
|
||
run["copy"].assert_called_once_with("Book it for Thursday.")
|
||
run["press"].assert_called_once_with(self.conf["paste_shortcut"])
|
||
|
||
def test_the_stages_are_named_as_they_happen(self):
|
||
run = self.run_chain()
|
||
self.assertEqual(run["stages"][:2], ["Transcribing…", "Cleaning up…"])
|
||
|
||
def test_cleanup_switched_off_pastes_what_was_heard(self):
|
||
self.conf["cleanup_enabled"] = False
|
||
run = self.run_chain()
|
||
run["cleanup"].assert_not_called()
|
||
run["copy"].assert_called_once_with("uh, book it for Thursday")
|
||
|
||
def test_auto_paste_switched_off_only_copies(self):
|
||
self.conf["auto_paste"] = False
|
||
self.conf["restore_clipboard"] = True
|
||
run = self.run_chain()
|
||
run["copy"].assert_called_once()
|
||
run["press"].assert_not_called()
|
||
run["read_clipboard"].assert_not_called()
|
||
|
||
def test_a_run_asked_for_from_a_terminal_pastes_nowhere(self):
|
||
"""The text comes back down the socket; the focused window is nobody's."""
|
||
run = self.run_chain(paste_override=False)
|
||
run["press"].assert_not_called()
|
||
run["copy"].assert_called_once()
|
||
|
||
def test_the_clipboard_is_put_back_afterwards(self):
|
||
self.conf["restore_clipboard"] = True
|
||
run = self.run_chain()
|
||
run["copy_bytes"].assert_called_once_with(b"what was there before")
|
||
|
||
def test_nothing_is_put_back_when_the_setting_is_off(self):
|
||
self.conf["restore_clipboard"] = False
|
||
run = self.run_chain()
|
||
run["copy_bytes"].assert_not_called()
|
||
|
||
def test_the_clipboard_is_put_back_when_the_keypress_fails(self):
|
||
self.conf["restore_clipboard"] = True
|
||
run = self.run_chain(paste_error=paste.PasteError("not trusted"))
|
||
self.assertIn("not trusted", run["failures"][0])
|
||
run["copy_bytes"].assert_called_once_with(b"what was there before")
|
||
|
||
def test_the_transcription_is_told_the_language_and_the_glossary(self):
|
||
self.conf["language"] = "tr"
|
||
self.conf["transcribe_prompt"] = "Paraşüt"
|
||
run = self.run_chain()
|
||
self.assertEqual(run["transcribe"].call_args.kwargs["language"], "tr")
|
||
self.assertEqual(run["transcribe"].call_args.kwargs["prompt"], "Paraşüt")
|
||
|
||
# ---- silence and stock phrases ----------------------------------------
|
||
|
||
def test_room_tone_costs_no_api_call(self):
|
||
run = self.run_chain(rms=[0.00001] * 60)
|
||
run["transcribe"].assert_not_called()
|
||
self.assertIn("No speech", run["failures"][0])
|
||
|
||
def test_the_silence_check_can_be_switched_off(self):
|
||
self.conf["skip_silent"] = False
|
||
run = self.run_chain(rms=[0.00001] * 60)
|
||
run["transcribe"].assert_called_once()
|
||
|
||
def test_a_stock_phrase_from_a_short_clip_is_thrown_away(self):
|
||
run = self.run_chain(duration=2.0, transcript="Altyazı M.K.")
|
||
self.assertIn("stock phrase", run["failures"][0])
|
||
run["copy"].assert_not_called()
|
||
|
||
def test_the_hallucination_filter_can_be_switched_off(self):
|
||
self.conf["filter_hallucinations"] = False
|
||
run = self.run_chain(duration=2.0, transcript="Altyazı M.K.")
|
||
run["copy"].assert_called_once()
|
||
|
||
# ---- when something goes wrong ----------------------------------------
|
||
|
||
def test_a_failed_cleanup_still_pastes_the_transcript(self):
|
||
run = self.run_chain(cleanup_error=api.ApiError("rate limited"))
|
||
_raw, text, warning = run["done"][0]
|
||
self.assertEqual(text, "uh, book it for Thursday")
|
||
self.assertIn("rate limited", warning)
|
||
run["copy"].assert_called_once_with("uh, book it for Thursday")
|
||
|
||
def test_a_failed_cleanup_is_never_silent(self):
|
||
"""A rejected key would otherwise look like dictation that works."""
|
||
run = self.run_chain(cleanup_error=api.ApiError("bad key"))
|
||
self.assertTrue(run["done"][0][2])
|
||
self.assertEqual(cfg.read_history()[0]["cleanup_error"], "bad key")
|
||
|
||
def test_a_failed_transcription_ends_the_run(self):
|
||
pipeline = worker.Pipeline(self.conf)
|
||
failures = []
|
||
pipeline.failed.connect(failures.append)
|
||
with mock.patch.object(api, "transcribe",
|
||
side_effect=api.ApiError("no credit")), \
|
||
mock.patch.object(paste, "copy") as copy:
|
||
pipeline._work(self.wav, 2.0, self.rms, False, None)
|
||
self.assertIn("no credit", failures[0])
|
||
copy.assert_not_called()
|
||
|
||
def test_a_clipboard_that_will_not_take_it(self):
|
||
pipeline = worker.Pipeline(self.conf)
|
||
failures = []
|
||
pipeline.failed.connect(failures.append)
|
||
with mock.patch.object(api, "transcribe", return_value="hello"), \
|
||
mock.patch.object(api, "cleanup", return_value="Hello."), \
|
||
mock.patch.object(paste, "read_clipboard", return_value=None), \
|
||
mock.patch.object(paste, "copy",
|
||
side_effect=paste.PasteError("no wl-copy")):
|
||
pipeline._work(self.wav, 2.0, self.rms, False, None)
|
||
self.assertIn("wl-copy", failures[0])
|
||
|
||
def test_an_unexpected_error_is_reported_rather_than_swallowed(self):
|
||
pipeline = worker.Pipeline(self.conf)
|
||
failures = []
|
||
pipeline.failed.connect(failures.append)
|
||
with mock.patch.object(api, "transcribe", side_effect=ValueError("oh dear")), \
|
||
mock.patch("traceback.print_exc"):
|
||
pipeline._work(self.wav, 2.0, self.rms, False, None)
|
||
self.assertIn("oh dear", failures[0])
|
||
|
||
# ---- handing it to an agent -------------------------------------------
|
||
|
||
def test_a_command_goes_to_the_agent_and_the_answer_comes_back(self):
|
||
run = self.run_chain(ask=True)
|
||
run["ask"].assert_called_once()
|
||
self.assertEqual(run["ask"].call_args.args[0], "uh, book it for Thursday")
|
||
run["copy"].assert_called_once_with("Booked.")
|
||
|
||
def test_a_command_is_not_cleaned_up_first_by_default(self):
|
||
"""The agent reads through the filler words without help."""
|
||
run = self.run_chain(ask=True)
|
||
run["cleanup"].assert_not_called()
|
||
|
||
def test_a_command_can_be_cleaned_up_if_you_want(self):
|
||
self.conf["assistant_cleanup"] = True
|
||
run = self.run_chain(ask=True)
|
||
run["cleanup"].assert_called_once()
|
||
self.assertEqual(run["ask"].call_args.args[0], "Book it for Thursday.")
|
||
|
||
def test_a_denied_tool_arrives_beside_the_answer(self):
|
||
run = self.run_chain(ask=True, answer=("Booked.", "It could not use: Bash"))
|
||
self.assertIn("Bash", run["done"][0][2])
|
||
|
||
def test_the_agent_has_its_own_paste_setting(self):
|
||
self.conf["assistant_paste"] = False
|
||
run = self.run_chain(ask=True)
|
||
run["press"].assert_not_called()
|
||
|
||
def test_a_command_that_was_cancelled(self):
|
||
pipeline = worker.Pipeline(self.conf)
|
||
cancels = []
|
||
pipeline.cancelled.connect(lambda: cancels.append(True))
|
||
with mock.patch.object(api, "transcribe", return_value="hello"), \
|
||
mock.patch.object(assistant, "ask", side_effect=assistant.Cancelled):
|
||
pipeline._work(self.wav, 2.0, self.rms, True, None)
|
||
self.assertEqual(cancels, [True])
|
||
|
||
def test_an_agent_that_is_not_installed(self):
|
||
pipeline = worker.Pipeline(self.conf)
|
||
failures = []
|
||
pipeline.failed.connect(failures.append)
|
||
with mock.patch.object(api, "transcribe", return_value="hello"), \
|
||
mock.patch.object(assistant, "ask",
|
||
side_effect=assistant.AssistantError("no claude")):
|
||
pipeline._work(self.wav, 2.0, self.rms, True, None)
|
||
self.assertIn("no claude", failures[0])
|
||
|
||
# ---- what is left behind ----------------------------------------------
|
||
|
||
def test_the_run_is_written_to_the_history(self):
|
||
self.run_chain()
|
||
row = cfg.read_history()[0]
|
||
self.assertEqual(row["raw"], "uh, book it for Thursday")
|
||
self.assertEqual(row["text"], "Book it for Thursday.")
|
||
self.assertEqual(row["duration"], 2.0)
|
||
self.assertEqual(row["model"], self.conf.transcribe_target().model)
|
||
self.assertEqual(row["mode"], "")
|
||
|
||
def test_a_command_is_recorded_as_one(self):
|
||
self.run_chain(ask=True)
|
||
row = cfg.read_history()[0]
|
||
self.assertEqual(row["mode"], "ask")
|
||
self.assertEqual(row["question"], "uh, book it for Thursday")
|
||
self.assertEqual(row["text"], "Booked.")
|
||
|
||
def test_the_history_is_kept_to_its_limit(self):
|
||
self.conf["history_limit"] = 2
|
||
for _ in range(4):
|
||
self.wav = make_wav(self.path("clip.wav"), speech(2.0))
|
||
self.run_chain()
|
||
self.assertEqual(len(cfg.read_history()), 2)
|
||
|
||
def test_the_recording_is_deleted_when_it_is_done_with(self):
|
||
self.run_chain()
|
||
self.assertFalse(os.path.exists(self.wav))
|
||
|
||
def test_the_recording_is_kept_when_the_setting_says_so(self):
|
||
self.conf["keep_audio"] = True
|
||
self.run_chain()
|
||
self.assertFalse(os.path.exists(self.wav))
|
||
self.assertEqual(len(list(cfg.RECORDINGS_DIR.iterdir())), 1)
|
||
|
||
def test_the_recording_goes_even_when_the_run_failed(self):
|
||
self.run_chain(rms=[0.00001] * 60)
|
||
self.assertFalse(os.path.exists(self.wav))
|
||
|
||
|
||
class Busy(DikteTest):
|
||
def test_a_second_run_while_one_is_going_waits_its_turn(self):
|
||
"""The microphone is free while a transcript is being cleaned up, so
|
||
the next dictation can already have been spoken by then. It has to run
|
||
once the first is done, in the order they were spoken, on one thread."""
|
||
pipeline = worker.Pipeline(self.config())
|
||
order = []
|
||
started, gate = threading.Event(), threading.Event()
|
||
|
||
def work(wav_path, *_rest):
|
||
order.append(wav_path)
|
||
started.set()
|
||
if wav_path == "first.wav":
|
||
gate.wait(5)
|
||
|
||
with mock.patch.object(pipeline, "_work", side_effect=work):
|
||
pipeline.run("first.wav", 1.0)
|
||
self.assertTrue(started.wait(5))
|
||
pipeline.run("second.wav", 1.0)
|
||
# Held, not dropped and not running beside the first.
|
||
self.assertEqual(order, ["first.wav"])
|
||
gate.set()
|
||
pipeline._thread.join(5)
|
||
self.assertEqual(order, ["first.wav", "second.wav"])
|
||
|
||
def test_a_run_arriving_after_the_queue_drained(self):
|
||
"""The worker thread ends with the queue; the next run brings one."""
|
||
pipeline = worker.Pipeline(self.config())
|
||
order = []
|
||
with mock.patch.object(pipeline, "_work",
|
||
side_effect=lambda wav, *rest: order.append(wav)):
|
||
pipeline.run("first.wav", 1.0)
|
||
pipeline._thread.join(5)
|
||
pipeline.run("second.wav", 1.0)
|
||
pipeline._thread.join(5)
|
||
self.assertEqual(order, ["first.wav", "second.wav"])
|
||
|
||
def test_the_chunk_length_matches_the_level_meter(self):
|
||
"""The silence thresholds are read in seconds, so the two must agree."""
|
||
self.assertAlmostEqual(worker.CHUNK_SECONDS, 1024 / 16000)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|