mirror of
https://github.com/yusufipk/dikte.git
synced 2026-09-11 19:06:11 +00:00
yusufipek
2f59029261
Twenty-two files at the top of the tree was the first thing anybody saw of this repository. They are one package now, imported relatively, and the three scripts that are not the front door moved under scripts/. install.sh stays where the README has always said it is. What starts the application is dikte/__main__.py: python3 -m dikte runs it, and so does naming the file, which is what the launcher symlink, both .desktop files, the macOS bundle and every registered shortcut do. Run by path there is no package around it, so it puts the checkout on sys.path itself. The installers now keep the keys you chose when they are given none, which is what an update is: update.sh no longer has to read them out and pass them back. An updater from before this commit cannot read them at all, so the one thing it can say, the default key with an empty discard key, is read as "nothing was asked for" rather than obeyed. That guard can go once nobody is updating across this commit.
296 lines
13 KiB
Python
296 lines
13 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 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_is_ignored(self):
|
||
pipeline = worker.Pipeline(self.config())
|
||
pipeline._thread = mock.Mock(is_alive=lambda: True)
|
||
self.assertTrue(pipeline.busy)
|
||
with mock.patch.object(worker.threading, "Thread") as thread:
|
||
pipeline.run("/tmp/nope.wav", 1.0)
|
||
thread.assert_not_called()
|
||
|
||
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()
|