Merge master: the shortcut table, local models, and a Mac still in them

Four files disagreed, and all four the same way: master had turned things the
Mac branch wrote out by hand into one list to read from.

Shortcuts are the whole of it. master gave every binding a row in
hotkey.SHORTCUTS, so the Mac's DESKTOP_IDS is gone and CarbonHotkey reads the
desktop id off that row, which also gives the new cancel key a status line on a
Mac. Settings builds its four rows through master's _shortcut_row, and that one
now asks _install_buttons for Install and Remove, so macOS gets a combination
box and nothing to press, and everywhere else the button says the desktop's own
name. dikte.py starts the listener from the same table, on macOS whatever the
setting says: there is nothing installed for it to be a fallback to.

The rest is two imports and a paste list that lives in paste.Desktop now.
This commit is contained in:
yusufipk
2026-08-02 09:36:22 +03:00
41 changed files with 4888 additions and 470 deletions
+1 -1
View File
@@ -27,7 +27,7 @@ atexit.register(shutil.rmtree, _SANDBOX, True)
# A key sitting in the environment would otherwise reach the code that falls
# back to it, and the tests for "there is no key" would pass only on a machine
# without one.
for _var in ("OPENAI_API_KEY", "OPENROUTER_API_KEY"):
for _var in ("OPENAI_API_KEY", "GROQ_API_KEY", "OPENROUTER_API_KEY"):
os.environ.pop(_var, None)
# The interface language leaks through module-level state, so the tests fix it
+12
View File
@@ -18,6 +18,7 @@ import sys
import tempfile
import unittest
import urllib.error
import urllib.request
import wave
from unittest import mock
@@ -40,6 +41,12 @@ linux_only = unittest.skipUnless(
)
def _no_network(*args, **kwargs):
raise AssertionError(
"a test reached the network; wrap the call in support.fake_urlopen"
)
def _no_exec(*args, **kwargs):
raise AssertionError(
"a test reached os.execv, which would replace the test process with the "
@@ -78,6 +85,11 @@ class DikteTest(unittest.TestCase):
# with it and hang, so it fails loudly here instead.
self.patch_attr(os, "execv", _no_exec)
# Every way out of here goes through urllib, so closing it is enough to
# keep the suite offline. A test that means to answer a request patches
# this again through fake_urlopen.
self.patch_attr(urllib.request, "urlopen", _no_network)
# ---- helpers ---------------------------------------------------------
def path(self, *parts):
+267 -2
View File
@@ -3,13 +3,21 @@
Nothing here reaches the network. What is checked is the request that would have
gone out, because that is what a new provider changes and what an old one
notices: the URL, the headers, the fields of the multipart body, the JSON.
Stopping one is the exception. Cutting a request off is done to the socket it
is blocked on, and a faked urlopen has no socket to cut, so those tests talk to
a server of their own on the loopback interface.
"""
import http.server
import json
import os
import threading
import time
import unittest
import api
import ggml
from tests.support import (
DikteTest,
fake_urlopen,
@@ -21,6 +29,7 @@ from tests.support import (
)
OPENAI = api.Target("openai", "OpenAI", "sk-test", api.OPENAI_URL, "gpt-4o-transcribe")
GROQ = api.Target("groq", "Groq", "gsk-test", api.GROQ_URL, "whisper-large-v3-turbo")
OPENROUTER = api.Target("openrouter", "OpenRouter", "sk-or-test",
api.OPENROUTER_URL, "openai/gpt-4o-transcribe")
@@ -32,6 +41,18 @@ class TimestampModel(unittest.TestCase):
def test_openrouter_namespaces_the_id(self):
self.assertEqual(api.timestamp_model("openrouter"), "openai/whisper-1")
def test_groq_keeps_the_model_that_was_chosen(self):
"""Every model it transcribes with is a whisper, so all of them do times."""
self.assertEqual(api.timestamp_model("groq", "whisper-large-v3"),
"whisper-large-v3")
def test_groq_with_nothing_chosen_falls_back(self):
self.assertEqual(api.timestamp_model("groq"), "whisper-large-v3-turbo")
def test_the_others_ignore_what_was_chosen(self):
self.assertEqual(api.timestamp_model("openai", "gpt-4o-transcribe"),
"whisper-1")
class Explain(DikteTest):
def error(self, status):
@@ -176,13 +197,28 @@ class Transcribe(DikteTest):
self.assertEqual(multipart_fields(calls[0])["language"], "tr")
self.assertNotIn("language", multipart_fields(calls[1]))
def test_the_glossary_goes_to_openai_only(self):
def test_the_glossary_goes_everywhere_but_openrouter(self):
"""OpenRouter takes the field and throws it away, so spare it the bytes."""
with fake_urlopen({"text": "hi"}) as calls:
api.transcribe(OPENAI, self.wav, prompt="Paraşüt, OpenFrame")
api.transcribe(GROQ, self.wav, prompt="Paraşüt, OpenFrame")
api.transcribe(OPENROUTER, self.wav, prompt="Paraşüt, OpenFrame")
self.assertIn("prompt", multipart_fields(calls[0]))
self.assertNotIn("prompt", multipart_fields(calls[1]))
self.assertIn("prompt", multipart_fields(calls[1]))
self.assertNotIn("prompt", multipart_fields(calls[2]))
def test_groq_goes_to_groq(self):
with fake_urlopen({"text": "hi"}) as calls:
api.transcribe(GROQ, self.wav)
self.assertEqual(calls[0].full_url,
"https://api.groq.com/openai/v1/audio/transcriptions")
self.assertEqual(multipart_fields(calls[0])["model"], "whisper-large-v3-turbo")
def test_a_refused_groq_key_is_explained_in_groq_s_name(self):
with fake_urlopen(http_error(401, '{"error": {"message": "bad key"}}')), \
self.assertRaises(api.ApiError) as caught:
api.transcribe(GROQ, self.wav)
self.assertIn("Groq", str(caught.exception))
def test_openrouter_is_attributed(self):
with fake_urlopen({"text": "hi"}) as calls:
@@ -242,6 +278,12 @@ class TranscribeSegments(DikteTest):
api.transcribe_segments(OPENROUTER, self.wav)
self.assertEqual(multipart_fields(calls[0])["model"], "openai/whisper-1")
def test_groq_stays_on_the_model_it_was_given(self):
target = GROQ._replace(model="whisper-large-v3")
with fake_urlopen(self.reply([{"start": 0, "end": 1, "text": "hi"}])) as calls:
api.transcribe_segments(target, self.wav)
self.assertEqual(multipart_fields(calls[0])["model"], "whisper-large-v3")
def test_the_segments_come_back_as_numbers(self):
with fake_urlopen(self.reply([
{"start": "0.5", "end": "2.25", "text": " hello "},
@@ -432,6 +474,229 @@ class ModelLists(DikteTest):
with self.assertRaises(api.ApiError):
api.openai_models("")
def test_the_same_list_read_from_groq(self):
with fake_urlopen({"data": [{"id": "llama-3.3-70b"},
{"id": "whisper-large-v3"}]}) as calls:
models = api.openai_models("gsk-test", api.GROQ_URL, "Groq")
self.assertEqual(calls[0].full_url, "https://api.groq.com/openai/v1/models")
self.assertEqual(models, ["whisper-large-v3"])
def test_a_missing_groq_key_says_groq(self):
with self.assertRaises(api.ApiError) as caught:
api.openai_models("", api.GROQ_URL, "Groq")
self.assertIn("Groq", str(caught.exception))
if __name__ == "__main__":
unittest.main()
class FakeServer:
"""A ggml.Server as far as api.py is concerned."""
def __init__(self, url="http://127.0.0.1:9999/v1", fails="", log=""):
self.url = url
self.fails = fails
self.log = log
self.starts = 0
def serve(self):
self.starts += 1
if self.fails:
raise ggml.LocalError(self.fails)
return self.url
def error(self):
return self.log
LOCAL = api.Target("local", "Local whisper", "", "", "ggml-base.bin")
class TranscribeHere(DikteTest):
def setUp(self):
super().setUp()
self.wav = str(self.path("clip.wav"))
os.makedirs(self.root, exist_ok=True)
with open(self.wav, "wb") as fh:
fh.write(b"RIFFfake")
self.server = FakeServer()
self.patch_attr(ggml, "whisper", self.server)
def test_the_address_comes_from_the_server_it_starts(self):
with fake_urlopen({"text": "hello"}) as calls:
api.transcribe(LOCAL, self.wav)
self.assertEqual(self.server.starts, 1)
self.assertEqual(calls[0].full_url,
"http://127.0.0.1:9999/v1/audio/transcriptions")
def test_nothing_local_is_authorised(self):
with fake_urlopen({"text": "hello"}) as calls:
api.transcribe(LOCAL, self.wav)
self.assertNotIn("Authorization", calls[0].headers)
def test_a_server_that_will_not_start_is_the_error_shown(self):
self.patch_attr(ggml, "whisper", FakeServer(fails="no model downloaded"))
with self.assertRaises(api.ApiError) as caught:
api.transcribe(LOCAL, self.wav)
self.assertIn("no model downloaded", str(caught.exception))
def test_a_server_that_dies_mid_request_says_what_it_printed(self):
self.patch_attr(ggml, "whisper", FakeServer(log="out of memory"))
with fake_urlopen(url_error("connection reset")):
with self.assertRaises(api.ApiError) as caught:
api.transcribe(LOCAL, self.wav)
self.assertIn("out of memory", str(caught.exception))
def test_the_hint_reaches_whisper_as_its_initial_prompt(self):
with fake_urlopen({"text": "hi"}) as calls:
api.transcribe(LOCAL, self.wav, prompt="Dikte, Paraşüt")
self.assertEqual(multipart_fields(calls[0])["prompt"], "Dikte, Paraşüt")
def test_a_word_broken_over_two_lines_is_put_back_together(self):
# whisper.cpp cuts on tokens and writes one segment per line, which in
# Turkish lands inside a word about as often as between two.
with fake_urlopen({"text": "Onlar akraba değ\niller. Ve\n devamı."}):
# The line break inside a word leaves nothing in its place; the
# one between two words is where whisper's own leading space is.
self.assertEqual(api.transcribe(LOCAL, self.wav),
"Onlar akraba değiller. Ve devamı.")
def test_a_local_timeout_is_not_a_hosted_one(self):
# Nothing is being spent but time, and a long file on a machine without
# a graphics card takes a good deal of it.
with fake_urlopen({"text": "hi"}):
api.transcribe(LOCAL, self.wav, timeout=300)
self.assertGreaterEqual(api.LOCAL_TIMEOUT, 600)
def test_segments_that_continue_a_word_are_merged(self):
reply = {"segments": [
{"start": 0.0, "end": 1.0, "text": " Onlar akraba değ"},
{"start": 1.0, "end": 1.4, "text": "iller."},
{"start": 2.0, "end": 3.0, "text": " Başka bir cümle."},
]}
with fake_urlopen(reply):
out = api.transcribe_segments(LOCAL, self.wav)
self.assertEqual([text for _, _, text in out],
["Onlar akraba değiller.", "Başka bir cümle."])
self.assertEqual(out[0][1], 1.4) # the merged cue covers the whole word
def test_the_loaded_model_is_the_one_asked_for_again(self):
with fake_urlopen({"segments": [{"start": 0, "end": 1, "text": " hi"}]}) as calls:
api.transcribe_segments(LOCAL, self.wav)
self.assertEqual(multipart_fields(calls[0])["model"], "ggml-base.bin")
class Stopping(unittest.TestCase):
"""The Stop button, from the far end: a request already blocked on a reply.
The one that matters is a whisper on this machine, which answers minutes
after it was asked, so it is a real socket here rather than a fake urlopen.
Nothing leaves the loopback interface.
"""
def setUp(self):
answering = threading.Event()
class Slow(http.server.BaseHTTPRequestHandler):
def do_POST(self):
self.rfile.read(int(self.headers.get("Content-Length") or 0))
answering.set()
time.sleep(30) # the model, thinking
def log_message(self, *args):
pass
self.answering = answering
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Slow)
threading.Thread(target=self.server.serve_forever, daemon=True).start()
self.addCleanup(self.server.server_close)
self.addCleanup(self.server.shutdown)
self.url = f"http://127.0.0.1:{self.server.server_address[1]}/v1/x"
def post(self, aborter, out):
try:
api._request(self.url, b"{}", {}, timeout=30, aborter=aborter)
out.append("answered")
except BaseException as exc: # noqa: BLE001 - the type is the result
out.append(type(exc).__name__)
def test_a_request_waiting_on_a_reply_is_cut_off(self):
aborter, out = api.Aborter(), []
thread = threading.Thread(target=self.post, args=(aborter, out))
thread.start()
self.assertTrue(self.answering.wait(10))
aborter.abort()
thread.join(timeout=10)
self.assertFalse(thread.is_alive())
self.assertEqual(out, ["Aborted"])
def test_a_request_that_starts_after_the_stop_never_goes_out(self):
aborter, out = api.Aborter(), []
aborter.abort()
self.post(aborter, out)
self.assertEqual(out, ["Aborted"])
self.assertFalse(self.answering.is_set())
def test_without_one_the_request_is_the_plain_urllib_one(self):
"""Everything that is not stoppable keeps the opener it always had."""
with fake_urlopen({"text": "hi"}) as calls:
api._request(self.url, b"{}", {})
self.assertEqual(len(calls), 1)
class Sockets(unittest.TestCase):
"""The few lines urllib takes between making a connection and blocking on
it. A stop that lands in there must not leave the request waiting out its
hour-long local timeout."""
class FakeConn:
auto_open = 1
sock = None
closed = False
def close(self):
self.closed = True
def test_a_connection_opened_after_the_stop_is_refused(self):
sockets = api._Sockets()
sockets.cut()
with self.assertRaises(api.Aborted):
sockets.add(self.FakeConn())
def test_one_that_is_already_open_is_closed_where_it_stands(self):
sockets, conn = api._Sockets(), self.FakeConn()
sockets.add(conn)
sockets.cut()
self.assertTrue(conn.closed)
def test_one_with_no_socket_yet_is_stopped_from_making_another(self):
"""close() leaves auto_open on, and the next line would reconnect."""
sockets, conn = api._Sockets(), self.FakeConn()
sockets.add(conn)
sockets.cut()
self.assertEqual(conn.auto_open, 0)
class Aborter(unittest.TestCase):
def test_what_was_registered_is_run_once_the_stop_lands(self):
aborter, cut = api.Aborter(), []
with aborter.holding(lambda: cut.append(True)):
aborter.abort()
self.assertEqual(cut, [True])
def test_a_block_that_ended_is_not_cut_afterwards(self):
aborter, cut = api.Aborter(), []
with aborter.holding(lambda: cut.append(True)):
pass
aborter.abort()
self.assertEqual(cut, [])
def test_a_stop_that_already_landed_stops_the_next_step_too(self):
aborter = api.Aborter()
aborter.abort()
with self.assertRaises(api.Aborted):
aborter.check()
with self.assertRaises(api.Aborted):
with aborter.holding(lambda: None):
pass
+9 -6
View File
@@ -79,9 +79,12 @@ class Effort(unittest.TestCase):
self.assertEqual(assistant.CODEX_EFFORT["xhigh"], "high")
self.assertEqual(assistant.CODEX_EFFORT["max"], "high")
def test_claude_has_no_rung_below_low(self):
self.assertEqual(assistant.CLAUDE_EFFORT["none"], "low")
self.assertEqual(assistant.CLAUDE_EFFORT["minimal"], "low")
def test_neither_one_asks_for_a_rung_below_low(self):
# Claude has none; Codex has one, but calls it "minimal" on the older
# models and "none" on the newer ones, and refuses the wrong word.
for scale in (assistant.CLAUDE_EFFORT, assistant.CODEX_EFFORT):
self.assertEqual(scale["none"], "low")
self.assertEqual(scale["minimal"], "low")
def test_an_empty_setting_asks_for_nothing(self):
self.assertEqual(assistant.CLAUDE_EFFORT.get("", ""), "")
@@ -232,10 +235,10 @@ class SessionMissing(unittest.TestCase):
self.assertFalse(assistant._session_missing(text))
def test_the_last_line_is_the_one_worth_showing(self):
self.assertEqual(assistant._last_line("warning\n\nreal error\n"),
self.assertEqual(assistant.last_line("warning\n\nreal error\n"),
"real error")
self.assertEqual(assistant._last_line(""), "")
self.assertEqual(assistant._last_line(None), "")
self.assertEqual(assistant.last_line(""), "")
self.assertEqual(assistant.last_line(None), "")
class Conclude(DikteTest):
+286
View File
@@ -0,0 +1,286 @@
"""Who cleans the transcript up, and what they are asked.
The CLIs are faked at subprocess.run: what the tests read is the argument list
each one is given, where the answer is picked up from, and what happens to the
chain when the program is missing, slow or unhappy. The OpenRouter path is the
one that was always there and is checked here only for still being taken.
"""
import os
import subprocess
import unittest
from unittest import mock
import api
import cleanup
import ggml
from tests.support import DikteTest, fake_urlopen, sent_json, url_error
from tests.test_api import FakeServer, chat_reply
def fake_run(stdout="", code=0, stderr="", last_message=""):
"""Stand in for subprocess.run, writing the file Codex would have written."""
calls = []
def run(cmd, **kwargs):
calls.append(cmd)
if last_message and "-o" in cmd:
with open(cmd[cmd.index("-o") + 1], "w", encoding="utf-8") as fh:
fh.write(last_message)
return subprocess.CompletedProcess(cmd, code, stdout, stderr)
return mock.patch.object(subprocess, "run", side_effect=run), calls
class Provider(DikteTest):
def test_the_default_is_still_openrouter(self):
self.assertEqual(cleanup.provider(self.config()), "openrouter")
def test_a_provider_this_version_does_not_have(self):
self.assertEqual(
cleanup.provider(self.config(cleanup_provider="ollama")), "openrouter")
def test_each_one_is_recognised(self):
for name in cleanup.PROVIDERS:
with self.subTest(name=name):
self.assertEqual(
cleanup.provider(self.config(cleanup_provider=name)), name)
def test_what_each_one_runs(self):
self.assertEqual(cleanup.executable("claude"), "claude")
self.assertEqual(cleanup.executable("codex"), "codex")
self.assertEqual(cleanup.executable("openrouter"), "")
def test_the_model_named_in_the_history_is_the_one_that_did_it(self):
self.assertEqual(cleanup.model(self.config(cleanup_model="some/model")),
"some/model")
self.assertEqual(
cleanup.model(self.config(cleanup_provider="claude")), "haiku")
self.assertEqual(
cleanup.model(self.config(cleanup_provider="claude",
cleanup_claude_model="opus")), "opus")
# Codex on its own default has no model id to report, only a name.
self.assertEqual(
cleanup.model(self.config(cleanup_provider="codex")), "codex")
self.assertEqual(
cleanup.model(self.config(cleanup_provider="codex",
cleanup_codex_model="gpt-5.4")), "gpt-5.4")
class OpenRouter(DikteTest):
def test_it_is_still_one_request_with_the_settings_as_they_were(self):
conf = self.config(openrouter_api_key="sk-or-test",
cleanup_model="some/model", cleanup_reasoning="low")
with mock.patch.object(api, "cleanup", return_value="Done.") as call:
self.assertEqual(cleanup.run("uh, done", conf, "the rules"), "Done.")
text, key, model, prompt = call.call_args.args
self.assertEqual((text, key, model, prompt),
("uh, done", "sk-or-test", "some/model", "the rules"))
self.assertEqual(call.call_args.kwargs["reasoning"], "low")
def test_no_cli_is_started_for_it(self):
conf = self.config(openrouter_api_key="sk-or-test")
patcher, calls = fake_run(stdout="never")
with patcher, mock.patch.object(api, "cleanup", return_value="Done."):
cleanup.run("uh, done", conf, "the rules")
self.assertEqual(calls, [])
class ClaudeCode(DikteTest):
def setUp(self):
super().setUp()
self.conf = self.config(cleanup_provider="claude")
self.patch_attr(cleanup.shutil, "which", lambda name: f"/usr/bin/{name}")
def run_cleanup(self, text="uh, book it", **kwargs):
patcher, calls = fake_run(**kwargs)
with patcher:
answer = cleanup.run(text, self.conf, "the rules")
return answer, calls[0]
def test_the_transcript_goes_in_fenced_and_the_rules_go_in_as_the_prompt(self):
answer, cmd = self.run_cleanup(stdout="Book it.\n")
self.assertEqual(answer, "Book it.")
self.assertEqual(cmd[0], "claude")
self.assertIn("<transcript>\nuh, book it\n</transcript>", cmd)
self.assertEqual(cmd[cmd.index("--system-prompt") + 1], "the rules")
self.assertEqual(cmd[cmd.index("--model") + 1], "haiku")
def test_it_is_given_nothing_to_run_and_nothing_to_remember(self):
_, cmd = self.run_cleanup(stdout="Book it.")
self.assertEqual(cmd[cmd.index("--tools") + 1], "")
self.assertIn("--strict-mcp-config", cmd)
self.assertIn("--no-session-persistence", cmd)
def test_the_thinking_setting_is_carried_over_in_its_own_words(self):
self.conf["cleanup_reasoning"] = "none"
_, cmd = self.run_cleanup(stdout="Book it.")
self.assertEqual(cmd[cmd.index("--effort") + 1], "low")
def test_no_thinking_setting_means_no_flag(self):
_, cmd = self.run_cleanup(stdout="Book it.")
self.assertNotIn("--effort", cmd)
def test_a_model_of_your_own(self):
self.conf["cleanup_claude_model"] = "claude-sonnet-5"
_, cmd = self.run_cleanup(stdout="Book it.")
self.assertEqual(cmd[cmd.index("--model") + 1], "claude-sonnet-5")
def test_an_answer_of_nothing_is_a_failure_rather_than_an_empty_paste(self):
with self.assertRaises(cleanup.CleanupError):
self.run_cleanup(stdout=" \n")
def test_the_last_line_of_the_complaint_is_what_gets_shown(self):
with self.assertRaises(cleanup.CleanupError) as caught:
self.run_cleanup(code=1, stderr="a warning\nout of credit\n")
self.assertEqual(str(caught.exception), "out of credit")
def test_a_failure_is_the_same_kind_the_chain_already_catches(self):
# worker, the file transcriber and the meeting all keep the raw
# transcript when an ApiError comes out of here.
self.assertTrue(issubclass(cleanup.CleanupError, api.ApiError))
def test_a_program_that_is_not_installed_says_so_before_running_anything(self):
self.patch_attr(cleanup.shutil, "which", lambda name: "")
with self.assertRaises(cleanup.CleanupError) as caught:
self.run_cleanup(stdout="Book it.")
self.assertIn("claude", str(caught.exception))
def test_a_run_that_never_ends(self):
def run(cmd, **kwargs):
raise subprocess.TimeoutExpired(cmd, 180)
with mock.patch.object(subprocess, "run", side_effect=run):
with self.assertRaises(cleanup.CleanupError) as caught:
cleanup.run("uh, book it", self.conf, "the rules")
self.assertIn("180", str(caught.exception))
class Codex(DikteTest):
def setUp(self):
super().setUp()
self.conf = self.config(cleanup_provider="codex")
self.patch_attr(cleanup.shutil, "which", lambda name: f"/usr/bin/{name}")
def run_cleanup(self, text="uh, book it", **kwargs):
patcher, calls = fake_run(**kwargs)
with patcher:
answer = cleanup.run(text, self.conf, "the rules")
return answer, calls[0]
def test_the_rules_ride_in_front_of_the_transcript(self):
answer, cmd = self.run_cleanup(last_message="Book it.\n")
self.assertEqual(answer, "Book it.")
self.assertEqual(cmd[:2], ["codex", "exec"])
self.assertEqual(cmd[-1],
"the rules\n\n---\n\n<transcript>\nuh, book it\n</transcript>")
def test_the_answer_is_read_from_the_file_rather_than_the_noise_on_stdout(self):
answer, _ = self.run_cleanup(
stdout="workdir: /home\nmodel: gpt-5.4\ntokens used 400\n",
last_message="Book it.",
)
self.assertEqual(answer, "Book it.")
def test_that_file_does_not_stay_behind(self):
_, cmd = self.run_cleanup(last_message="Book it.")
self.assertFalse(os.path.exists(cmd[cmd.index("-o") + 1]))
def test_it_may_read_but_not_write_and_has_nobody_to_ask(self):
_, cmd = self.run_cleanup(last_message="Book it.")
self.assertEqual(cmd[cmd.index("--sandbox") + 1], "read-only")
self.assertIn('approval_policy="never"', cmd)
self.assertIn("--ephemeral", cmd)
def test_the_model_is_left_alone_until_one_is_typed_in(self):
_, cmd = self.run_cleanup(last_message="Book it.")
self.assertNotIn("-m", cmd)
self.conf["cleanup_codex_model"] = "gpt-5.4"
_, cmd = self.run_cleanup(last_message="Book it.")
self.assertEqual(cmd[cmd.index("-m") + 1], "gpt-5.4")
def test_the_thinking_setting_lands_on_the_nearest_rung_codex_has(self):
self.conf["cleanup_reasoning"] = "xhigh"
_, cmd = self.run_cleanup(last_message="Book it.")
self.assertIn('model_reasoning_effort="high"', cmd)
def test_an_answer_of_nothing(self):
with self.assertRaises(cleanup.CleanupError):
self.run_cleanup(stdout="tokens used 400", last_message="")
if __name__ == "__main__":
unittest.main()
class Here(DikteTest):
"""llama.cpp, answering the request OpenRouter answers."""
def setUp(self):
super().setUp()
self.conf = self.config(cleanup_provider="local",
local_llm_model="gemma.gguf")
self.server = FakeServer()
self.patch_attr(ggml, "llm", self.server)
def test_the_address_comes_from_the_server_it_starts(self):
with fake_urlopen(chat_reply("Done.")) as calls:
self.assertEqual(cleanup.run("uh, done", self.conf, "the rules"),
"Done.")
self.assertEqual(self.server.starts, 1)
self.assertEqual(calls[0].full_url,
"http://127.0.0.1:9999/v1/chat/completions")
def test_no_key_is_wanted_and_none_is_sent(self):
with fake_urlopen(chat_reply("Done.")) as calls:
cleanup.run("uh, done", self.conf, "the rules")
self.assertNotIn("Authorization", calls[0].headers)
def test_thinking_is_turned_off_in_the_words_llama_cpp_uses(self):
with fake_urlopen(chat_reply("Done.")) as calls:
cleanup.run("uh, done", self.conf, "the rules")
self.assertEqual(sent_json(calls[0])["chat_template_kwargs"],
{"enable_thinking": False})
def test_the_models_own_default_asks_for_nothing(self):
self.conf["local_llm_reasoning"] = ""
with fake_urlopen(chat_reply("Done.")) as calls:
cleanup.run("uh, done", self.conf, "the rules")
self.assertNotIn("chat_template_kwargs", sent_json(calls[0]))
def test_a_reply_longer_than_the_transcript_is_cut_off(self):
# A small model will repeat the transcript until the context is full,
# and every one of those tokens is a second of somebody waiting.
with fake_urlopen(chat_reply("Done.")) as calls:
cleanup.run("x" * 4000, self.conf, "the rules")
self.assertEqual(sent_json(calls[0])["max_tokens"], 4000)
def test_a_short_dictation_still_gets_room_to_answer(self):
with fake_urlopen(chat_reply("Done.")) as calls:
cleanup.run("uh, done", self.conf, "the rules")
self.assertEqual(sent_json(calls[0])["max_tokens"], 512)
def test_a_reply_that_was_all_thinking_names_the_setting_that_fixes_it(self):
reply = {"choices": [{"message": {"content": "", "reasoning": "hmm"}}]}
with fake_urlopen(reply), self.assertRaises(api.ApiError) as caught:
cleanup.run("uh, done", self.conf, "the rules")
self.assertIn("Thinking", str(caught.exception))
def test_a_server_that_will_not_start_is_the_error_shown(self):
self.patch_attr(ggml, "llm", FakeServer(fails="llama.cpp is not installed"))
with self.assertRaises(api.ApiError) as caught:
cleanup.run("uh, done", self.conf, "the rules")
self.assertIn("llama.cpp", str(caught.exception))
def test_a_server_that_dies_mid_request_says_what_it_printed(self):
self.patch_attr(ggml, "llm", FakeServer(log="out of memory"))
with fake_urlopen(url_error("connection reset")):
with self.assertRaises(api.ApiError) as caught:
cleanup.run("uh, done", self.conf, "the rules")
self.assertIn("out of memory", str(caught.exception))
def test_no_cli_is_started_for_it(self):
patcher, calls = fake_run(stdout="never")
with patcher, fake_urlopen(chat_reply("Done.")):
cleanup.run("uh, done", self.conf, "the rules")
self.assertEqual(calls, [])
+79 -1
View File
@@ -14,8 +14,9 @@ from unittest import mock
import cli
import config as cfg
import hotkey
import ipc
from tests.support import DikteTest
from tests.support import DikteTest, fake_urlopen
class Options:
@@ -144,6 +145,22 @@ class Parser(unittest.TestCase):
self.assertIsNone(opts.verb)
self.assertEqual(opts.func, cli.cmd_plain)
def test_every_global_shortcut_runs_a_verb_that_exists(self):
"""A shortcut registers a command line; a verb the parser never heard of
is a key that does nothing at all when it is pressed."""
for name, spec in hotkey.SHORTCUTS.items():
with self.subTest(name=name):
opts = self.parse(spec.verb)
self.assertTrue(callable(opts.func))
def test_every_shortcut_can_be_installed_and_removed_by_name(self):
for name in hotkey.SHORTCUTS:
with self.subTest(name=name):
self.assertEqual(self.parse("shortcut", "install", name).which,
name)
self.assertEqual(self.parse("shortcut", "remove", name).which,
name)
def test_every_verb_is_wired_to_something(self):
for verb in ("record", "toggle", "start", "stop", "cancel", "ask",
"session", "transcribe", "meeting", "meetings", "history",
@@ -316,6 +333,67 @@ class ConfigCommands(DikteTest):
{"cleanup", "subtitles", "meeting", "agent"})
class Providers(DikteTest):
"""The terminal reaches every provider the settings window does."""
def run_cmd(self, func, **values):
with captured() as (out, err):
code = func(Options(**values))
return code, out.getvalue(), err.getvalue()
def test_a_provider_the_settings_window_offers_is_a_choice_here_too(self):
parser = cli.build_parser()
for provider in cfg.TRANSCRIBERS:
with self.subTest(provider=provider):
opts = parser.parse_args(["models", "--provider", provider])
self.assertEqual(opts.provider, provider)
self.assertEqual(parser.parse_args(["test-key", provider]).which,
provider)
def test_the_model_list_is_read_from_the_chosen_provider(self):
self.write_config({"groq_api_key": "gsk-test"})
with fake_urlopen({"data": [{"id": "whisper-large-v3"}]}) as calls:
code, out, _ = self.run_cmd(cli.cmd_models, provider="groq",
transcription=False)
self.assertEqual(code, 0)
self.assertEqual(calls[0].full_url, "https://api.groq.com/openai/v1/models")
self.assertEqual(out.strip(), "whisper-large-v3")
def test_a_key_that_is_not_there_is_reported_under_its_own_name(self):
code, out, _ = self.run_cmd(cli.cmd_test_key, which="groq")
self.assertEqual(code, 1)
self.assertIn("groq", out)
self.assertIn("Groq", out)
class Doctor(DikteTest):
"""One pass over everything the settings window checks behind its buttons."""
def run_doctor(self, as_json=True, **settings):
self.write_config(settings)
with mock.patch.object(ipc, "send", return_value=None), \
captured() as (out, _err):
cli.cmd_doctor(Options(json=as_json))
return json.loads(out.getvalue()) if as_json else out.getvalue()
def test_cleanup_on_openrouter_is_a_question_about_the_key(self):
reply = self.run_doctor(cleanup_model="some/model")
self.assertEqual(reply["cleanup"]["provider"], "openrouter")
self.assertEqual(reply["cleanup"]["model"], "some/model")
self.assertIn("OpenRouter key, cleaning up on some/model",
self.run_doctor(as_json=False, cleanup_model="some/model"))
def test_cleanup_on_a_cli_is_a_question_about_the_program(self):
reply = self.run_doctor(cleanup_provider="codex",
cleanup_codex_model="gpt-5.4")
self.assertEqual(reply["cleanup"]["provider"], "codex")
self.assertEqual(reply["cleanup"]["model"], "gpt-5.4")
self.assertIn("codex", reply["programs"])
self.assertIn("codex, cleaning up on gpt-5.4",
self.run_doctor(as_json=False, cleanup_provider="codex",
cleanup_codex_model="gpt-5.4"))
class Finding(DikteTest):
def test_no_history_at_all(self):
self.assertIsNone(cli._find_history("last"))
+100 -3
View File
@@ -12,7 +12,9 @@ import unittest
from unittest import mock
import api
import cleanup
import config as cfg
import ggml
import i18n
import paste
from tests.support import DikteTest
@@ -126,10 +128,23 @@ class Keys(DikteTest):
def test_no_key_anywhere(self):
self.assertEqual(cfg.Config().openai_key(), "")
def test_every_provider_falls_back_to_the_variable_of_its_own_name(self):
with mock.patch.dict(os.environ, {"GROQ_API_KEY": "gsk-env"}):
self.assertEqual(cfg.Config().groq_key(), "gsk-env")
class TranscribeTarget(DikteTest):
def test_openai_by_default(self):
target = self.config(openai_api_key="sk-test").transcribe_target()
def test_this_machine_by_default(self):
target = cfg.Config().transcribe_target()
self.assertEqual(target.provider, "local")
self.assertEqual(target.api_key, "")
# Empty on purpose: the server picks a port when it starts, and reading
# a setting must not be what starts it.
self.assertEqual(target.base_url, "")
def test_openai_when_it_is_picked(self):
target = self.config(transcribe_provider="openai",
openai_api_key="sk-test").transcribe_target()
self.assertEqual(target.provider, "openai")
self.assertEqual(target.service, "OpenAI")
self.assertEqual(target.api_key, "sk-test")
@@ -146,8 +161,24 @@ class TranscribeTarget(DikteTest):
self.assertEqual(target.api_key, "sk-or-test")
self.assertEqual(target.model, "openai/whisper-1")
def test_groq_when_it_is_picked(self):
conf = self.config(transcribe_provider="groq", groq_api_key="gsk-test",
groq_transcribe_model="whisper-large-v3")
target = conf.transcribe_target()
self.assertEqual(target.provider, "groq")
self.assertEqual(target.service, "Groq")
self.assertEqual(target.api_key, "gsk-test")
self.assertEqual(target.base_url, api.GROQ_URL)
self.assertEqual(target.model, "whisper-large-v3")
def test_a_provider_this_version_has_never_heard_of(self):
"""A config written by a fork, or by a version that dropped one."""
target = self.config(transcribe_provider="deepgram").transcribe_target()
self.assertEqual(target.provider, "openai")
def test_a_self_hosted_endpoint(self):
conf = self.config(openai_base_url="http://localhost:8080/v1")
conf = self.config(transcribe_provider="openai",
openai_base_url="http://localhost:8080/v1")
self.assertEqual(conf.transcribe_target().base_url, "http://localhost:8080/v1")
@@ -459,3 +490,69 @@ class Directories(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class LocalCleanup(DikteTest):
def test_the_local_model_is_what_the_history_records(self):
conf = self.config(cleanup_provider="local",
local_llm_model="gemma-3-4b-it-Q4_K_M.gguf")
self.assertEqual(cleanup.provider(conf), "local")
self.assertEqual(cleanup.model(conf), "gemma-3-4b-it-Q4_K_M.gguf")
def test_it_needs_no_program_on_the_path(self):
# whisper.cpp and llama.cpp are fetched rather than installed, so unlike
# Claude Code and Codex there is no executable to look for.
self.assertEqual(cleanup.executable("local"), "")
def test_the_minutes_do_not_follow_the_cleanup_provider(self):
# A 4B model here will strip the filler words out of a dictation and
# will not write up an hour long meeting.
conf = self.config(cleanup_provider="local")
self.assertEqual(conf["meeting_model"], cfg.DEFAULTS["meeting_model"])
def test_only_the_cleanup_setting_asks_for_the_local_model(self):
self.assertFalse(cfg.Config().uses_local_llm())
self.assertTrue(self.config(cleanup_provider="local").uses_local_llm())
class ReadyToRun(DikteTest):
def setUp(self):
super().setUp()
self.patch_attr(ggml, "MODELS_DIR", self.path("models"))
def install(self, name):
path = ggml.whisper_model_path(name)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b"model")
def test_a_missing_program_is_not_ready(self):
with mock.patch("shutil.which", return_value=None):
self.install("ggml-base.bin")
conf = self.config(local_model="ggml-base.bin")
self.assertFalse(conf.transcribe_ready())
def test_a_missing_model_is_not_ready_either(self):
with mock.patch("shutil.which", return_value="/usr/bin/whisper-server"):
conf = self.config(local_model="ggml-base.bin")
self.assertFalse(conf.transcribe_ready())
def test_both_halves_in_place(self):
with mock.patch("shutil.which", return_value="/usr/bin/whisper-server"):
self.install("ggml-base.bin")
conf = self.config(local_model="ggml-base.bin")
self.assertTrue(conf.transcribe_ready())
def test_a_hosted_provider_is_ready_when_it_has_a_key(self):
conf = self.config(transcribe_provider="openai", openai_api_key="sk-test")
self.assertTrue(conf.transcribe_ready())
def test_the_settings_reach_the_servers(self):
conf = self.config(local_model="ggml-base.bin", local_threads=4,
local_gpu=False, local_llm_model="gemma.gguf",
local_llm_context=4096)
conf.apply_local()
self.addCleanup(ggml.whisper.configure, model="", threads=0, gpu=True)
self.assertEqual(ggml.whisper.settings()["model"], "ggml-base.bin")
self.assertEqual(ggml.whisper.settings()["threads"], 4)
self.assertFalse(ggml.whisper.settings()["gpu"])
self.assertEqual(ggml.llm.settings()["context"], 4096)
+36 -1
View File
@@ -7,6 +7,7 @@ made up a stamp nobody recorded.
"""
import contextlib
import time
import unittest
import wave
from unittest import mock
@@ -170,7 +171,7 @@ class Transcriber(DikteTest):
worker.failed.connect(failures.append)
worker.progress.connect(progress.append)
def to_wav(path, workdir):
def to_wav(path, workdir, aborter=None):
return make_wav(self.path("converted.wav"), tone(1.0))
with mock.patch.object(ft, "_to_wav", side_effect=to_wav), \
@@ -225,6 +226,40 @@ class Transcriber(DikteTest):
_, _, _, cleanup_call = self.run_chain(cleanup=True, transcript="")
cleanup_call.assert_not_called()
def test_a_stopped_run_is_not_a_failure(self):
def stopped(*args, **kwargs):
raise api.Aborted
done, failures, progress, _ = self.run_chain(fail=stopped)
self.assertEqual(failures, [])
self.assertEqual(done, [])
self.assertEqual(progress[-1], "Stopped.")
def test_the_request_is_handed_the_stop_to_watch(self):
worker = ft.FileTranscriber(self.conf)
with mock.patch.object(ft, "_to_wav", side_effect=lambda *a: self.source), \
mock.patch.object(ft.shutil, "which", return_value="/usr/bin/ffmpeg"), \
mock.patch.object(api, "transcribe", return_value="text") as call:
worker._work(self.source, False, False)
self.assertIs(call.call_args.kwargs["aborter"], worker._abort)
def test_stopping_a_local_run_stops_the_model_with_it(self):
"""Closing the socket is nothing to a process of ours: it would grind on
to the end of the chunk with nobody left to hand the answer to."""
worker = ft.FileTranscriber(self.conf)
worker._local = mock.Mock()
worker.stop()
self.assertTrue(worker._abort.aborted)
for _ in range(100):
if worker._local.stop.called:
break
time.sleep(0.01)
worker._local.stop.assert_called_once_with()
def test_a_run_that_is_over_leaves_the_model_alone(self):
worker = ft.FileTranscriber(self.conf)
worker.stop()
self.assertTrue(worker._abort.aborted)
def test_a_second_start_while_one_is_running_is_ignored(self):
worker = ft.FileTranscriber(self.conf)
worker._thread = mock.Mock(is_alive=lambda: True)
+664
View File
@@ -0,0 +1,664 @@
"""Fetching a program and a model, and keeping a server alive on them.
No network and no whisper.cpp: the downloads are answered from memory, and the
servers are stand-in scripts that take the same arguments and open their port
when they are told to, which is the only thing the code waits on.
"""
import contextlib
import hashlib
import io
import os
import signal
import sys
import tarfile
import textwrap
import threading
import time
from unittest import mock
import ggml
import hub
from tests.support import (DikteTest, fake_urlopen, http_error, json_body,
linux_only, url_error)
def body(data, length=None):
"""What urlopen hands back for a download: a reader with a length header."""
class Body:
def __init__(self):
self._buf = io.BytesIO(data)
self.headers = {"Content-Length":
str(len(data) if length is None else length)}
def read(self, count=-1):
return self._buf.read(count)
def __enter__(self):
return self
def __exit__(self, *_):
return False
return Body()
def item(name, data, url="https://example.invalid/f", sha=True):
return hub.Item(name, url, len(data),
hashlib.sha256(data).hexdigest() if sha else "")
@contextlib.contextmanager
def serving(release, archive):
"""Answer by what is being asked for rather than by what came before.
An install asks GitHub what the release is and then asks for one file out of
it, and the first of those two comes from the cache the second time around.
Answering in order would then hand the archive request the release listing.
"""
def opener(request, timeout=None):
url = request.full_url
if "api.github.com" in url:
return json_body(release)
return body(archive)
with mock.patch("urllib.request.urlopen", side_effect=opener) as calls:
yield calls
def tarball(entries):
"""A .tar.gz laid out the way the releases are: one directory of files."""
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
for name, content in entries.items():
info = tarfile.TarInfo(name)
info.size = len(content)
info.mode = 0o755
tar.addfile(info, io.BytesIO(content))
return buf.getvalue()
class Local(DikteTest):
"""A test with its own bin, models and cache directories."""
def setUp(self):
super().setUp()
self.patch_attr(ggml, "DATA_DIR", self.path("data"))
self.patch_attr(ggml, "BIN_DIR", self.path("data", "bin"))
self.patch_attr(ggml, "MODELS_DIR", self.path("data", "models"))
self.patch_attr(hub, "CACHE_DIR", self.path("cache"))
# --- downloading ----------------------------------------------------------
class Download(Local):
def test_it_lands_and_the_part_file_is_gone(self):
data = b"a model, more or less" * 100
target = self.path("data", "models", "m.bin")
with fake_urlopen(body(data)):
self.assertTrue(ggml.download(item("m.bin", data), target))
self.assertEqual(target.read_bytes(), data)
self.assertFalse(target.with_name("m.bin.part").exists())
def test_a_wrong_checksum_installs_nothing(self):
data = b"the bytes that arrived"
wrong = hub.Item("m.bin", "https://example.invalid/f", len(data), "f" * 64)
target = self.path("data", "models", "m.bin")
with fake_urlopen(body(data)):
with self.assertRaises(ggml.LocalError) as caught:
ggml.download(wrong, target)
self.assertIn("checksum", str(caught.exception))
self.assertFalse(target.exists())
self.assertFalse(target.with_name("m.bin.part").exists())
def test_a_body_shorter_than_its_header_installs_nothing(self):
data = b"half of it"
target = self.path("data", "models", "m.bin")
with fake_urlopen(body(data, length=len(data) * 2)):
with self.assertRaises(ggml.LocalError):
ggml.download(item("m.bin", data), target)
self.assertFalse(target.exists())
def test_a_file_with_no_published_checksum_is_refused(self):
# Everything fetched here is run or parsed by something written in C++,
# and GitHub did not always publish a digest.
data = b"a program, say"
target = self.path("data", "models", "m.bin")
with fake_urlopen(body(data)):
with self.assertRaises(ggml.LocalError) as caught:
ggml.download(item("m.bin", data, sha=False), target)
self.assertIn("checksum", str(caught.exception))
self.assertFalse(target.exists())
def test_nothing_is_asked_for_before_it_is_refused(self):
# The refusal is not worth a gigabyte of somebody's bandwidth first.
with fake_urlopen(body(b"never read")) as calls:
with self.assertRaises(ggml.LocalError):
ggml.download(item("m.bin", b"x", sha=False), self.path("m.bin"))
self.assertEqual(calls, [])
def test_stopping_leaves_nothing_behind(self):
data = b"x" * (ggml.DOWNLOAD_CHUNK * 3)
target = self.path("data", "models", "m.bin")
with fake_urlopen(body(data)):
landed = ggml.download(item("m.bin", data), target,
should_stop=lambda: True)
self.assertFalse(landed)
self.assertFalse(target.exists())
self.assertFalse(target.with_name("m.bin.part").exists())
def test_progress_is_reported_against_the_total(self):
data = b"y" * (ggml.DOWNLOAD_CHUNK + 5)
seen = []
with fake_urlopen(body(data)):
ggml.download(item("m.bin", data), self.path("data", "m.bin"),
on_progress=lambda done, total: seen.append((done, total)))
self.assertEqual(seen[-1], (len(data), len(data)))
self.assertGreater(len(seen), 1)
def test_a_refused_connection_says_which_file(self):
with fake_urlopen(url_error("no route to host")):
with self.assertRaises(ggml.LocalError) as caught:
ggml.download(item("m.bin", b"x"), self.path("data", "m.bin"))
self.assertIn("m.bin", str(caught.exception))
def test_an_http_error_is_not_written_to_disk(self):
target = self.path("data", "m.bin")
with fake_urlopen(http_error(404)):
with self.assertRaises(ggml.LocalError):
ggml.download(item("m.bin", b"x"), target)
self.assertFalse(target.exists())
# --- installing a program -------------------------------------------------
class InstallProgram(Local):
def setUp(self):
super().setUp()
# Built once, because the release listing has to publish its checksum
# and a tarball is not the same bytes twice.
self.archive = tarball({
"whisper-bin-ubuntu-x64/whisper-server": b"#!/bin/sh\nexit 0\n",
"whisper-bin-ubuntu-x64/libwhisper.so": b"not really a library",
})
def release(self, *names, archive=None):
digest = hashlib.sha256(self.archive if archive is None else archive)
return {"tag_name": "v1.9.1", "assets": [
{"name": name, "browser_download_url": f"https://example.invalid/{name}",
"size": 10, "digest": "sha256:" + digest.hexdigest()}
for name in names]}
def install(self, *names, archive=None):
self.patch_attr(ggml, "_arch", lambda: "x64")
blob = self.archive if archive is None else archive
with serving(self.release(*names, archive=blob), blob) as calls:
path = ggml.install_program(ggml.WHISPER)
return path, [call.args[0].full_url for call in calls.call_args_list]
def test_the_binary_and_its_libraries_land_together(self):
path, _ = self.install("whisper-bin-ubuntu-x64.tar.gz")
self.assertTrue(os.path.isfile(path))
self.assertTrue(os.access(path, os.X_OK))
self.assertTrue(os.path.isfile(os.path.join(os.path.dirname(path),
"libwhisper.so")))
def test_the_build_for_this_machine_is_the_one_fetched(self):
_, urls = self.install("whisper-bin-x64.zip", "whisper-bin-ubuntu-arm64.tar.gz",
"whisper-bin-ubuntu-x64.tar.gz")
self.assertTrue(urls[1].endswith("whisper-bin-ubuntu-x64.tar.gz"))
def test_a_release_with_nothing_for_this_machine_says_so(self):
self.patch_attr(ggml, "_arch", lambda: "x64")
with fake_urlopen(self.release("whisper-bin-Win32.zip")):
with self.assertRaises(ggml.LocalError) as caught:
ggml.install_program(ggml.WHISPER)
self.assertIn("this machine", str(caught.exception))
def test_what_was_installed_is_remembered(self):
path, _ = self.install("whisper-bin-ubuntu-x64.tar.gz")
self.assertEqual(ggml.installed_program(ggml.WHISPER), path)
self.assertEqual(ggml.installed_version(ggml.WHISPER), "v1.9.1")
def test_a_record_pointing_at_a_deleted_binary_counts_for_nothing(self):
path, _ = self.install("whisper-bin-ubuntu-x64.tar.gz")
os.unlink(path)
self.assertEqual(ggml.installed_program(ggml.WHISPER), "")
def test_the_archive_is_not_kept(self):
self.install("whisper-bin-ubuntu-x64.tar.gz")
left = list((self.path("data", "bin", "whisper")).glob("*.tar.gz"))
self.assertEqual(left, [])
def test_the_previous_version_is_swept_up(self):
self.install("whisper-bin-ubuntu-x64.tar.gz")
old = self.path("data", "bin", "whisper", "v1.9.0")
old.mkdir(parents=True)
(old / "whisper-server").write_bytes(b"older")
self.install("whisper-bin-ubuntu-x64.tar.gz")
self.assertFalse(old.exists())
def test_an_archive_without_the_binary_is_refused(self):
empty = tarball({"whisper-bin-ubuntu-x64/README": b"nothing here"})
with self.assertRaises(ggml.LocalError) as caught:
self.install("whisper-bin-ubuntu-x64.tar.gz", archive=empty)
self.assertIn("whisper-server", str(caught.exception))
def test_a_release_without_a_published_checksum_is_refused(self):
# GitHub did not always publish one, and whisper.cpp v1.8.0 and older
# still have none.
self.patch_attr(ggml, "_arch", lambda: "x64")
listing = {"tag_name": "v1.8.0", "assets": [
{"name": "whisper-bin-ubuntu-x64.tar.gz",
"browser_download_url": "https://example.invalid/w.tar.gz",
"size": 10}]}
with serving(listing, self.archive):
with self.assertRaises(ggml.LocalError) as caught:
ggml.install_program(ggml.WHISPER)
self.assertIn("checksum", str(caught.exception))
self.assertEqual(ggml.installed_program(ggml.WHISPER), "")
def test_an_archive_that_is_not_what_was_promised_installs_nothing(self):
listing = self.release("whisper-bin-ubuntu-x64.tar.gz")
other = tarball({"whisper-bin-ubuntu-x64/whisper-server": b"#!/bin/sh\nrm -rf\n"})
self.patch_attr(ggml, "_arch", lambda: "x64")
with serving(listing, other):
with self.assertRaises(ggml.LocalError) as caught:
ggml.install_program(ggml.WHISPER)
self.assertIn("checksum", str(caught.exception))
self.assertEqual(ggml.installed_program(ggml.WHISPER), "")
def test_an_archive_cannot_write_outside_the_directory_it_is_opened_in(self):
# An archive is not a trusted thing to unpack: a member named ../../ is
# how one writes over a file it was never given.
escape = tarball({"../../../escaped": b"should not land"})
path = self.path("data", "bin", "whisper", "v1.9.1")
with self.assertRaises(ggml.LocalError):
self.install("whisper-bin-ubuntu-x64.tar.gz", archive=escape)
self.assertFalse(self.path("escaped").exists())
self.assertFalse((path.parent.parent / "escaped").exists())
def test_a_symlink_out_of_the_directory_does_not_survive_either(self):
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
info = tarfile.TarInfo("whisper-bin-ubuntu-x64/whisper-server")
info.type, info.linkname = tarfile.SYMTYPE, "/etc/passwd"
tar.addfile(info)
with self.assertRaises(ggml.LocalError):
self.install("whisper-bin-ubuntu-x64.tar.gz", archive=buf.getvalue())
def test_everything_is_asked_for_over_tls(self):
for url in (hub.GITHUB_API, hub.HF_API, hub.HF_FILES):
with self.subTest(url=url):
self.assertTrue(url.startswith("https://"))
def test_llama_takes_the_vulkan_build_when_there_is_a_loader(self):
self.patch_attr(ggml, "_arch", lambda: "x64")
self.patch_attr(ggml, "_has_vulkan", lambda: True)
self.assertEqual(ggml._wanted_assets(ggml.LLAMA)[0],
"bin-ubuntu-vulkan-x64.tar.gz")
def test_llama_falls_back_to_the_plain_build_without_one(self):
self.patch_attr(ggml, "_arch", lambda: "x64")
self.patch_attr(ggml, "_has_vulkan", lambda: False)
self.assertEqual(ggml._wanted_assets(ggml.LLAMA), ("bin-ubuntu-x64.tar.gz",))
class WhichCopyRuns(Local):
def test_a_system_build_wins_over_a_downloaded_one(self):
self.patch_attr(ggml, "installed_program", lambda program: "/data/whisper-server")
with mock.patch("shutil.which", return_value="/usr/bin/whisper-server"):
self.assertEqual(ggml.program_path(ggml.WHISPER), "/usr/bin/whisper-server")
def test_the_downloaded_one_is_used_when_there_is_no_system_build(self):
self.patch_attr(ggml, "installed_program", lambda program: "/data/whisper-server")
with mock.patch("shutil.which", return_value=None):
self.assertEqual(ggml.program_path(ggml.WHISPER), "/data/whisper-server")
def test_a_setting_pointing_at_nothing_is_no_program(self):
self.assertEqual(ggml.program_path(ggml.WHISPER, "/nowhere/whisper-server"), "")
def test_a_setting_pointing_at_a_program_wins(self):
mine = self.path("mine")
mine.write_text("#!/bin/sh\n")
mine.chmod(0o755)
with mock.patch("shutil.which", return_value="/usr/bin/whisper-server"):
self.assertEqual(ggml.program_path(ggml.WHISPER, str(mine)), str(mine))
# --- the lists ------------------------------------------------------------
WHISPER_TREE = [
{"type": "file", "path": "ggml-base.bin", "size": 147951465,
"lfs": {"oid": "a" * 64}},
{"type": "file", "path": "ggml-large-v3-turbo-q5_0.bin", "size": 574041195,
"lfs": {"oid": "b" * 64}},
{"type": "file", "path": "ggml-base-encoder.mlmodelc.zip", "size": 37922638,
"lfs": {"oid": "c" * 64}},
{"type": "file", "path": "README.md", "size": 3196},
]
GGUF_TREE = [
{"type": "file", "path": "gemma-3-4b-it-Q4_K_M.gguf", "size": 2489000000,
"lfs": {"oid": "a" * 64}},
{"type": "file", "path": "gemma-3-4b-it-Q8_0.gguf", "size": 4130000000,
"lfs": {"oid": "b" * 64}},
{"type": "file", "path": "mmproj-model-f16.gguf", "size": 851000000,
"lfs": {"oid": "c" * 64}},
{"type": "file", "path": "mtp-gemma-4-E4B-it-Q4_0.gguf", "size": 59000000,
"lfs": {"oid": "d" * 64}},
{"type": "file", "path": "huge-00001-of-00009.gguf", "size": 40000000000,
"lfs": {"oid": "e" * 64}},
{"type": "file", "path": "README.md", "size": 100},
]
class Catalogue(Local):
def test_only_models_are_offered_and_the_small_ones_first(self):
with fake_urlopen(WHISPER_TREE):
models = ggml.whisper_models()
self.assertEqual([m.name for m in models],
["ggml-base.bin", "ggml-large-v3-turbo-q5_0.bin"])
def test_the_core_ml_encoders_are_not_models(self):
with fake_urlopen(WHISPER_TREE):
names = [m.name for m in ggml.whisper_models()]
self.assertNotIn("ggml-base-encoder.mlmodelc.zip", names)
def test_the_projector_and_the_draft_head_are_not_models(self):
with fake_urlopen(GGUF_TREE):
names = [q.name for q in ggml.llm_quants("ggml-org/gemma-3-4b-it-GGUF")]
self.assertEqual(names,
["gemma-3-4b-it-Q4_K_M.gguf", "gemma-3-4b-it-Q8_0.gguf"])
def test_a_model_split_across_files_is_left_out(self):
with fake_urlopen(GGUF_TREE):
names = [q.name for q in ggml.llm_quants("ggml-org/gemma-3-4b-it-GGUF")]
self.assertNotIn("huge-00001-of-00009.gguf", names)
def test_the_suggestions_come_first_and_the_rest_follow(self):
listing = [{"id": "ggml-org/something-new-GGUF"},
{"id": ggml.SUGGESTED_LLM[0]}]
with fake_urlopen(listing):
found = ggml.llm_repos()
self.assertEqual(found[0], ggml.SUGGESTED_LLM[0])
self.assertIn("ggml-org/something-new-GGUF", found)
def test_an_unreachable_list_still_offers_the_suggestions(self):
with fake_urlopen(url_error()):
self.assertEqual(ggml.llm_repos(), list(ggml.SUGGESTED_LLM))
def test_an_unreachable_whisper_list_is_an_error_worth_showing(self):
with fake_urlopen(url_error()):
with self.assertRaises(ggml.LocalError):
ggml.whisper_models()
def test_what_is_on_disk_is_read_from_disk(self):
self.assertEqual(ggml.installed_whisper_models(), [])
path = ggml.whisper_model_path("ggml-base.bin")
path.parent.mkdir(parents=True)
path.write_bytes(b"model")
self.assertEqual(ggml.installed_whisper_models(), ["ggml-base.bin"])
self.assertTrue(ggml.have_model(path))
def test_an_empty_file_is_not_a_model(self):
path = ggml.llm_model_path("ggml-org/x-GGUF/model.gguf")
path.parent.mkdir(parents=True)
path.write_bytes(b"")
self.assertFalse(ggml.have_model(path))
def test_a_model_is_named_by_its_file_not_its_repository(self):
self.assertEqual(ggml.llm_model_path("ggml-org/x-GGUF/model.gguf").name,
"model.gguf")
# --- keeping a server alive -----------------------------------------------
STAND_IN = textwrap.dedent("""
import http.server, sys, threading, time
args = sys.argv[1:]
def opt(name, default=""):
return args[args.index(name) + 1] if name in args else default
if "--die" in args:
print("could not load model: no such file")
sys.exit(2)
time.sleep(float(opt("--wait", "0")))
started = time.monotonic()
healthy_after = float(opt("--healthy-after", "0"))
class Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
ok = time.monotonic() - started >= healthy_after
self.send_response(200 if ok else 503)
self.end_headers()
self.wfile.write(b"{}")
def log_message(self, *a):
pass
server = http.server.HTTPServer((opt("--host"), int(opt("--port"))), Handler)
print("listening on " + opt("--port"), flush=True)
server.serve_forever()
""")
class Servers(Local):
def setUp(self):
super().setUp()
self.path("data").mkdir(parents=True, exist_ok=True)
# Named for the program and kept inside the data directory, because that
# is what the sweep looks for on a command line.
self.script = self.path("data", "whisper-server.py")
self.script.write_text(STAND_IN)
self.addCleanup(ggml.stop_all)
self.servers = []
def server(self, program=ggml.WHISPER, **settings):
defaults = {"extra": []}
defaults.update(settings)
made = ggml.Server(
program,
lambda values: [sys.executable, str(self.script)] + list(values["extra"]),
defaults,
)
self.servers.append(made)
self.addCleanup(made.stop)
return made
def test_a_started_server_hands_back_its_address(self):
server = self.server()
url = server.serve()
self.assertRegex(url, r"^http://127\.0\.0\.1:\d+/v1$")
self.assertTrue(server.running)
def test_the_second_call_does_not_start_a_second_one(self):
server = self.server()
first = server.serve()
self.assertEqual(server.serve(), first)
def test_a_settings_change_stops_what_was_running(self):
server = self.server()
server.serve()
server.configure(extra=["--wait", "0"])
self.assertFalse(server.running)
def test_the_new_settings_are_what_the_next_start_uses(self):
server = self.server()
server.serve()
server.configure(extra=["--healthy-after", "0"])
second = server.serve()
self.assertTrue(server.running)
self.assertTrue(second)
def test_a_program_that_dies_reports_what_it_printed(self):
server = self.server(extra=["--die"])
with self.assertRaises(ggml.LocalError) as caught:
server.serve()
self.assertIn("no such file", str(caught.exception))
self.assertFalse(server.running)
def test_a_model_that_is_still_loading_is_not_ready_yet(self):
# llama binds its port first and answers /health with 503 until the
# model is in memory, so the open port on its own is not the signal.
server = self.server(program=ggml.LLAMA, extra=["--healthy-after", "0.4"])
started = time.monotonic()
server.serve()
self.assertGreaterEqual(time.monotonic() - started, 0.4)
def test_a_start_that_never_becomes_ready_gives_up(self):
self.patch_attr(ggml, "STARTUP_TIMEOUT", 0.5)
server = self.server(program=ggml.LLAMA, extra=["--healthy-after", "30"])
with self.assertRaises(ggml.LocalError):
server.serve()
def test_stopping_leaves_nothing_running(self):
server = self.server()
server.serve()
server.stop()
self.assertFalse(server.running)
self.assertEqual(server.base_url(), "")
def test_the_last_thing_it_printed_is_available(self):
server = self.server()
server.serve()
self.assertIn("listening", server.error())
def test_asking_what_is_running_does_not_wait_for_a_start(self):
"""A model being loaded must not freeze the settings window.
The interface asks a running server what it is doing while a start is in
flight, and a lock held across the whole start would stop it dead.
"""
server = self.server(extra=["--wait", "0.6"])
answers = []
def start():
server.serve()
thread = __import__("threading").Thread(target=start)
thread.start()
try:
time.sleep(0.15)
began = time.monotonic()
answers.append(server.settings())
answers.append(server.running)
self.assertLess(time.monotonic() - began, 0.2)
finally:
thread.join(timeout=10)
@linux_only
def test_a_server_a_killed_dikte_left_behind_is_swept_up(self):
server = self.server()
server.serve()
# What a SIGKILL of Dikte leaves: the child still running, the pid file
# still on disk, and nothing left that knows about either.
proc, server._proc = server._proc, None
self.assertTrue(server.sweep())
self.assertEqual(proc.wait(timeout=5), -signal.SIGTERM)
@linux_only
def test_a_pid_that_belongs_to_something_else_is_left_alone(self):
server = self.server()
server._remember(os.getpid()) # this test runner, not a server
self.assertFalse(server.sweep())
def test_no_pid_file_is_nothing_to_sweep(self):
self.assertFalse(self.server().sweep())
def test_a_start_that_goes_wrong_takes_its_process_with_it(self):
started = []
def explode(inner, proc, port):
started.append(proc)
raise RuntimeError("something in the wait went wrong")
self.patch_attr(ggml.Server, "_wait_ready", explode)
server = self.server()
with self.assertRaises(RuntimeError):
server.serve()
# Nothing else holds a reference to it, so leaving it running would leak
# a loaded model with nobody left to ask it anything.
self.assertIsNotNone(started[0].poll())
self.assertFalse(server.sweep()) # and the pid file went with it
class Arguments(Local):
"""What the two command lines say, since neither program is here to say it."""
def setUp(self):
super().setUp()
self.binary = self.path("whisper-server")
self.binary.write_text("#!/bin/sh\n")
self.binary.chmod(0o755)
def whisper_model(self, name="ggml-base.bin"):
path = ggml.whisper_model_path(name)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b"model")
return name
def test_the_inference_path_is_the_one_api_py_builds(self):
args = ggml._whisper_args({"binary": str(self.binary), "gpu": True,
"threads": 0, "model": self.whisper_model()})
self.assertIn("--inference-path", args)
self.assertEqual(args[args.index("--inference-path") + 1],
"/v1/audio/transcriptions")
def test_detection_rather_than_english_when_nothing_is_asked_for(self):
args = ggml._whisper_args({"binary": str(self.binary), "gpu": True,
"threads": 0, "model": self.whisper_model()})
self.assertEqual(args[args.index("-l") + 1], "auto")
def test_the_graphics_card_is_turned_off_rather_than_asked_for(self):
settings = {"binary": str(self.binary), "gpu": False, "threads": 2,
"model": self.whisper_model()}
args = ggml._whisper_args(settings)
self.assertIn("-ng", args)
self.assertEqual(args[args.index("-t") + 1], "2")
def test_a_missing_model_is_a_message_about_settings(self):
with self.assertRaises(ggml.LocalError) as caught:
ggml._whisper_args({"binary": str(self.binary), "gpu": True,
"threads": 0, "model": "ggml-nothing.bin"})
self.assertIn("Settings", str(caught.exception))
def test_a_missing_program_says_so_before_a_missing_model(self):
with mock.patch("shutil.which", return_value=None):
with self.assertRaises(ggml.LocalError) as caught:
ggml._whisper_args({"binary": "", "gpu": True, "threads": 0,
"model": self.whisper_model()})
self.assertIn("whisper.cpp", str(caught.exception))
def test_the_layers_go_to_the_card_when_there_is_one(self):
model = ggml.llm_model_path("m.gguf")
model.parent.mkdir(parents=True, exist_ok=True)
model.write_bytes(b"gguf")
args = ggml._llm_args({"binary": str(self.binary), "gpu": True,
"threads": 0, "model": "m.gguf", "context": 4096})
self.assertEqual(args[args.index("-ngl") + 1], "99")
self.assertEqual(args[args.index("-c") + 1], "4096")
def test_no_card_means_no_layers_offloaded(self):
model = ggml.llm_model_path("m.gguf")
model.parent.mkdir(parents=True, exist_ok=True)
model.write_bytes(b"gguf")
args = ggml._llm_args({"binary": str(self.binary), "gpu": False,
"threads": 0, "model": "m.gguf", "context": 4096})
self.assertEqual(args[args.index("-ngl") + 1], "0")
class Sizes(DikteTest):
def test_bytes_are_written_the_way_a_download_is_talked_about(self):
self.assertEqual(ggml.human_size(512), "512 B")
self.assertEqual(ggml.human_size(574041195), "547.4 MB")
self.assertEqual(ggml.human_size(3_095_033_483), "2.9 GB")
+40
View File
@@ -6,6 +6,7 @@ import subprocess
import unittest
from unittest import mock
import config as cfg
import hotkey
from tests.support import DikteTest, FakeCompleted, linux_only
@@ -56,6 +57,28 @@ class ParseShortcut(unittest.TestCase):
self.assertEqual(hotkey.parse_shortcut(None), (None, None))
class Table(unittest.TestCase):
"""The one list of global shortcuts. The command line, the settings window
and install.sh read it instead of keeping a copy each, so what it has to
hold together is checked here rather than in three places."""
def test_every_shortcut_remembers_itself_in_a_real_setting(self):
for name, spec in hotkey.SHORTCUTS.items():
with self.subTest(name=name):
self.assertIn(spec.setting, cfg.DEFAULTS)
def test_no_two_share_a_desktop_entry(self):
ids = [spec.desktop_id for spec in hotkey.SHORTCUTS.values()]
self.assertEqual(len(ids), len(set(ids)))
def test_only_the_toggle_falls_back_to_a_key_of_its_own(self):
"""The rest are off until you pick one, and emptying the box is how you
turn them off again."""
self.assertEqual(hotkey.SHORTCUTS["toggle"].fallback, "Ctrl+Space")
self.assertEqual([name for name, spec in hotkey.SHORTCUTS.items()
if spec.fallback], ["toggle"])
class ModsMatch(unittest.TestCase):
"""The combination has to be exact, or Ctrl+Space fires on Ctrl+Shift+Space."""
@@ -122,6 +145,23 @@ class Bindings(DikteTest):
thread.assert_called_once()
self.assertEqual(len(listener._bindings[57]), 2)
def test_starting_and_discarding_do_not_fire_on_each_other(self):
"""The two defaults are one modifier apart on the same key code, so the
modifier set is the only thing keeping them apart."""
listener = hotkey.EvdevHotkey()
self.addCleanup(listener.stop)
with mock.patch.object(listener, "_open_devices", return_value=[99]), \
mock.patch.object(hotkey.threading, "Thread"):
listener.start({"toggle": "Ctrl+Space", "cancel": "Ctrl+Alt+Space"})
def fired(held):
return [name for mods, name in listener._bindings[57]
if hotkey.EvdevHotkey._mods_match(held, mods)]
self.assertEqual(fired({29}), ["toggle"]) # ctrl
self.assertEqual(fired({29, 56}), ["cancel"]) # ctrl + alt
self.assertEqual(fired({29, 42}), []) # ctrl + shift
class Chooser(DikteTest):
"""Which desktop is asked to register the shortcut."""
+184
View File
@@ -0,0 +1,184 @@
"""What GitHub and Hugging Face are asked, and what is believed of the answer."""
import json
import hub
from tests.support import DikteTest, fake_urlopen, http_error, url_error
RELEASE = {
"tag_name": "v1.9.1",
"assets": [
{"name": "whisper-bin-ubuntu-x64.tar.gz",
"browser_download_url": "https://example.invalid/ubuntu-x64.tar.gz",
"size": 9379235, "digest": "sha256:" + "a" * 64},
{"name": "whisper-bin-x64.zip",
"browser_download_url": "https://example.invalid/win.zip",
"size": 100, "digest": "sha256:" + "b" * 64},
{"name": "no-url-here.zip", "size": 1},
],
}
TREE = [
{"type": "file", "path": ".gitattributes", "size": 1477},
{"type": "file", "path": "ggml-base.bin", "size": 147951465,
"lfs": {"oid": "c" * 64, "size": 147951465}},
{"type": "directory", "path": "extra"},
{"type": "file", "path": "extra/ggml-tiny.bin", "size": 77691713,
"lfs": {"oid": "d" * 64, "size": 77691713}},
]
MODELS = [
{"id": "ggml-org/gemma-3-4b-it-GGUF", "downloads": 44606,
"lastModified": "2026-07-01T00:00:00.000Z"},
{"id": "ggml-org/gpt-oss-20b-GGUF", "downloads": 47975},
{"noid": True},
]
class Releases(DikteTest):
def setUp(self):
super().setUp()
self.patch_attr(hub, "CACHE_DIR", self.path("cache"))
def test_the_tag_and_the_assets_come_back(self):
with fake_urlopen(RELEASE) as calls:
tag, assets = hub.release("ggml-org/whisper.cpp")
self.assertEqual(tag, "v1.9.1")
self.assertEqual([a.name for a in assets],
["whisper-bin-ubuntu-x64.tar.gz", "whisper-bin-x64.zip"])
self.assertEqual(calls[0].full_url,
"https://api.github.com/repos/ggml-org/whisper.cpp/"
"releases/latest")
def test_the_sha256_prefix_is_dropped(self):
with fake_urlopen(RELEASE):
_, assets = hub.release("ggml-org/whisper.cpp")
self.assertEqual(assets[0].sha256, "a" * 64)
def test_a_tag_asks_for_that_tag(self):
with fake_urlopen(RELEASE) as calls:
hub.release("ggml-org/whisper.cpp", "v1.9.1")
self.assertTrue(calls[0].full_url.endswith("/releases/tags/v1.9.1"))
def test_a_release_with_no_assets_is_an_error(self):
with fake_urlopen({"tag_name": "v1", "assets": []}):
with self.assertRaises(hub.HubError):
hub.release("ggml-org/whisper.cpp")
def test_the_second_call_asks_nobody(self):
with fake_urlopen(RELEASE) as calls:
hub.release("ggml-org/whisper.cpp")
hub.release("ggml-org/whisper.cpp")
self.assertEqual(len(calls), 1)
def test_a_refresh_asks_again(self):
with fake_urlopen(RELEASE) as calls:
hub.release("ggml-org/whisper.cpp")
hub.release("ggml-org/whisper.cpp", refresh=True)
self.assertEqual(len(calls), 2)
def test_an_old_cache_beats_no_answer(self):
with fake_urlopen(RELEASE):
hub.release("ggml-org/whisper.cpp")
# Old enough that it would normally be fetched again, and no network
# to fetch it with.
for path in self.path("cache").iterdir():
os_utime(path)
with fake_urlopen(url_error()):
tag, assets = hub.release("ggml-org/whisper.cpp")
self.assertEqual(tag, "v1.9.1")
self.assertEqual(len(assets), 2)
def test_no_cache_and_no_network_says_so(self):
with fake_urlopen(url_error("no route to host")):
with self.assertRaises(hub.HubError) as caught:
hub.release("ggml-org/whisper.cpp")
self.assertIn("api.github.com", str(caught.exception))
def test_an_http_error_names_the_host_and_the_code(self):
with fake_urlopen(http_error(404, "nope")):
with self.assertRaises(hub.HubError) as caught:
hub.release("ggml-org/nothing")
self.assertIn("404", str(caught.exception))
class Files(DikteTest):
def setUp(self):
super().setUp()
self.patch_attr(hub, "CACHE_DIR", self.path("cache"))
def test_directories_are_left_out(self):
with fake_urlopen(TREE):
files = hub.files("ggerganov/whisper.cpp")
self.assertEqual([f.name for f in files],
[".gitattributes", "ggml-base.bin", "extra/ggml-tiny.bin"])
def test_the_url_is_the_one_that_serves_the_bytes(self):
with fake_urlopen(TREE):
files = hub.files("ggerganov/whisper.cpp")
self.assertEqual(
files[1].url,
"https://huggingface.co/ggerganov/whisper.cpp/resolve/main/ggml-base.bin")
def test_the_lfs_object_id_is_the_checksum(self):
with fake_urlopen(TREE):
files = hub.files("ggerganov/whisper.cpp")
self.assertEqual(files[1].sha256, "c" * 64)
self.assertEqual(files[1].size, 147951465)
def test_a_file_outside_lfs_has_no_checksum(self):
with fake_urlopen(TREE):
files = hub.files("ggerganov/whisper.cpp")
self.assertEqual(files[0].sha256, "")
def test_an_answer_that_is_not_a_list_is_an_error(self):
with fake_urlopen({"error": "Invalid username or password."}):
with self.assertRaises(hub.HubError):
hub.files("ggml-org/whisper.cpp")
class Repos(DikteTest):
def setUp(self):
super().setUp()
self.patch_attr(hub, "CACHE_DIR", self.path("cache"))
def test_it_asks_for_one_author_and_for_gguf(self):
with fake_urlopen(MODELS) as calls:
found = hub.repos(author="ggml-org")
self.assertIn("author=ggml-org", calls[0].full_url)
self.assertIn("filter=gguf", calls[0].full_url)
self.assertEqual([r.id for r in found],
["ggml-org/gemma-3-4b-it-GGUF", "ggml-org/gpt-oss-20b-GGUF"])
def test_a_missing_download_count_is_zero(self):
with fake_urlopen(MODELS):
found = hub.repos(author="ggml-org")
self.assertEqual(found[0].downloads, 44606)
self.assertEqual(found[1].updated, "")
def os_utime(path):
"""Backdate a cache file past its time to live."""
import os
import time
old = time.time() - hub.CACHE_TTL - 60
os.utime(path, (old, old))
class CacheOnDisk(DikteTest):
def setUp(self):
super().setUp()
self.patch_attr(hub, "CACHE_DIR", self.path("cache"))
def test_what_is_stored_is_what_came_back(self):
with fake_urlopen(RELEASE):
hub.release("ggml-org/whisper.cpp")
stored = [json.loads(p.read_text()) for p in self.path("cache").iterdir()]
self.assertEqual(stored[0]["tag_name"], "v1.9.1")
def test_a_cache_that_cannot_be_written_is_not_a_failure(self):
self.patch_attr(hub, "CACHE_DIR", self.path("nope", "deeper"))
self.path("nope").write_text("a file where a directory would go")
with fake_urlopen(RELEASE):
tag, _ = hub.release("ggml-org/whisper.cpp")
self.assertEqual(tag, "v1.9.1")
+207 -3
View File
@@ -13,7 +13,9 @@ from unittest import mock
from PyQt6.QtWidgets import QApplication, QMessageBox
import cleanup
import config as cfg
import hotkey
import overlay as overlay_module
import paste
import settings_ui
@@ -39,13 +41,27 @@ CHANGED = {
"filter_hallucinations": False,
"keep_audio": True,
"openai_api_key": "sk-test-key",
"groq_api_key": "gsk-test-key",
"openrouter_api_key": "sk-or-test-key",
"transcribe_provider": "openrouter",
"transcribe_model": "whisper-1",
"groq_transcribe_model": "whisper-large-v3",
"openrouter_transcribe_model": "openai/whisper-1",
"cleanup_enabled": False,
"cleanup_provider": "local",
"cleanup_model": "some/other-model",
"cleanup_claude_model": "opus",
"cleanup_codex_model": "gpt-5",
"cleanup_reasoning": "high",
"local_model": "ggml-small.bin",
"local_gpu": False,
"local_preload": False,
"local_threads": 6,
"local_llm_model": "gemma-3-4b-it-Q4_K_M.gguf",
"local_llm_repo": "ggml-org/gemma-4-E2B-it-GGUF",
"local_llm_gpu": False,
"local_llm_preload": True,
"local_llm_reasoning": "low",
"cleanup_prompt": "Only fix the punctuation.",
"file_cleanup_prompt": "Keep the stamps where they are.",
"transcribe_prompt": "Paraşüt, OpenFrame",
@@ -77,6 +93,7 @@ CHANGED = {
"file_timestamps": True,
"file_cleanup": False,
"shortcut": "Ctrl+Alt+Space",
"cancel_shortcut": "Meta+Shift+Space",
"evdev_hotkey": True,
"history_limit": 50,
}
@@ -105,7 +122,7 @@ class Settings(DikteTest):
self.path("kglobalshortcutsrc")))
def window(self, conf):
window = settings_ui.SettingsWindow(conf, "dikte toggle")
window = settings_ui.SettingsWindow(conf)
self.addCleanup(window.deleteLater)
self.addCleanup(window.close)
return window
@@ -133,6 +150,20 @@ class Settings(DikteTest):
with self.subTest(key=key):
self.assertEqual(stored[key], value)
def test_the_model_box_on_screen_belongs_to_whoever_cleans_up(self):
"""An OpenRouter id and a Claude alias are not the same field."""
window = self.window(cfg.Config())
boxes = {"openrouter": window.cleanup_model_row,
"claude": window.cleanup_claude_model,
"codex": window.cleanup_codex_model}
for provider, box in boxes.items():
with self.subTest(provider=provider):
window._select_data(window.cleanup_provider, provider)
shown = [name for name, other in boxes.items()
if not other.isHidden()]
self.assertEqual(shown, [provider])
self.assertFalse(box.isHidden())
def test_the_settings_the_window_does_not_show_are_left_alone(self):
"""A tab nobody wrote must not reset what the command line set."""
self.write_config({"silence_db": -42.0, "speech_margin_db": 15.0,
@@ -143,6 +174,37 @@ class Settings(DikteTest):
self.assertEqual(stored["speech_margin_db"], 15.0)
self.assertEqual(stored["openrouter_base_url"], "http://localhost:1234/v1")
def test_every_global_shortcut_has_a_row_of_its_own(self):
window = self.window(cfg.Config())
self.assertEqual(set(window._shortcut_rows), set(hotkey.SHORTCUTS))
def test_emptying_a_shortcut_turns_it_off_but_not_the_toggle(self):
"""The application is unusable without the toggle, so that one box
falls back. The rest stay empty, which is how they are switched off."""
conf = cfg.Config()
window = self.window(conf)
for box, _status, _missing in window._shortcut_rows.values():
box.setCurrentText("")
window._save()
self.assertEqual(conf["shortcut"], "Ctrl+Space")
self.assertEqual(conf["cancel_shortcut"], "")
self.assertEqual(conf["assistant_shortcut"], "")
self.assertEqual(conf["meeting_shortcut"], "")
def test_installing_the_discard_key_writes_its_own_entry(self):
conf = cfg.Config()
window = self.window(conf)
window._shortcut_rows["cancel"][0].setCurrentText("Meta+Shift+Space")
with mock.patch.object(settings_ui.hotkey, "install_shortcut",
return_value=(True, "saved")) as install:
window._install_shortcut("cancel")
combo, command = install.call_args.args
self.assertEqual(combo, "Meta+Shift+Space")
self.assertTrue(command.endswith(" cancel"))
self.assertEqual(install.call_args.kwargs["desktop_id"],
hotkey.CANCEL_DESKTOP_ID)
self.assertEqual(conf["cancel_shortcut"], "Meta+Shift+Space")
def test_a_prompt_left_at_its_default_is_stored_as_empty(self):
"""So that switching the interface language switches the prompt too."""
conf = cfg.Config()
@@ -154,14 +216,44 @@ class Settings(DikteTest):
def test_each_provider_keeps_its_own_transcription_model(self):
self.write_config({"transcribe_provider": "openai",
"transcribe_model": "gpt-4o-transcribe",
"groq_transcribe_model": "whisper-large-v3",
"openrouter_transcribe_model": "openai/whisper-1"})
conf = cfg.Config()
window = self.window(conf)
window.transcribe_provider.setCurrentIndex(
window.transcribe_provider.findData("openrouter"))
for provider in ("groq", "openrouter"):
window.transcribe_provider.setCurrentIndex(
window.transcribe_provider.findData(provider))
window._save()
self.assertEqual(conf["transcribe_provider"], "openrouter")
self.assertEqual(conf["transcribe_model"], "gpt-4o-transcribe")
self.assertEqual(conf["groq_transcribe_model"], "whisper-large-v3")
def test_the_provider_box_offers_every_provider_config_knows(self):
window = self.window(cfg.Config())
offered = [window.transcribe_provider.itemData(i)
for i in range(window.transcribe_provider.count())]
self.assertEqual(offered, ["local"] + list(cfg.TRANSCRIBERS))
def test_the_cleanup_box_offers_everyone_cleanup_py_dispatches_to(self):
window = self.window(cfg.Config())
offered = [window.cleanup_provider.itemData(i)
for i in range(window.cleanup_provider.count())]
self.assertEqual(sorted(offered), sorted(cleanup.PROVIDERS))
def test_the_answer_to_a_test_lands_under_the_key_it_was_asked_about(self):
"""One signal serves all three buttons, so it carries which one asked."""
window = self.window(cfg.Config())
window._on_test_done("groq", True, "it works")
button, answer = window._testers["groq"]
self.assertEqual(answer.text(), "✓ it works")
self.assertTrue(button.isEnabled())
self.assertEqual(window._testers["openai"][1].text(), "")
def test_a_key_lands_in_the_field_of_its_own_provider(self):
self.write_config({"groq_api_key": "gsk-mine"})
window = self.window(cfg.Config())
self.assertEqual(window.groq_key.text(), "gsk-mine")
self.assertEqual(window.openai_key.text(), "")
def test_saving_applies_the_lowered_history_limit_at_once(self):
for index in range(10):
@@ -183,6 +275,43 @@ class Settings(DikteTest):
window = self.window(cfg.Config())
self.assertEqual(window.windowTitle(), "Dikte Ayarları")
def test_the_audio_file_switches_are_kept_without_the_save_button(self):
"""They are ticked to transcribe one file, not to fill in a form."""
self.write_config({"file_timestamps": False, "file_cleanup": True})
window = self.window(cfg.Config())
window.file_timestamps.setChecked(True)
window.file_cleanup.setChecked(False)
stored = self.read_config_file()
self.assertTrue(stored["file_timestamps"])
self.assertFalse(stored["file_cleanup"])
def test_loading_the_audio_file_tab_is_not_taken_for_a_change(self):
self.write_config({"file_timestamps": True, "file_cleanup": False})
conf = cfg.Config()
with mock.patch.object(conf, "save") as save:
window = self.window(conf)
save.assert_not_called()
self.assertTrue(window.file_timestamps.isChecked())
self.assertFalse(window.file_cleanup.isChecked())
def test_the_run_button_comes_back_when_the_stop_lands(self):
"""In whichever language, since the worker says so through t() too."""
for language in ("auto", "tr"):
with self.subTest(language=language):
self.write_config({"ui_language": language})
window = self.window(cfg.Config())
window.file_run.setEnabled(False)
window._on_file_progress(settings_ui.t("Stopped."))
self.assertTrue(window.file_run.isEnabled())
def test_stop_leaves_nothing_to_press_twice(self):
window = self.window(cfg.Config())
with mock.patch.object(window.transcriber, "stop") as stop:
window.file_stop.setEnabled(True)
window._stop_file()
stop.assert_called_once_with()
self.assertFalse(window.file_stop.isEnabled())
class MacSettings(Settings):
"""The same window and the same round trip, standing on a Mac.
@@ -296,3 +425,78 @@ class Overlay(DikteTest):
if __name__ == "__main__":
unittest.main()
class LocalModels(DikteTest):
"""The download boxes, without a network and without either program."""
def window(self, conf):
window = settings_ui.SettingsWindow(conf)
self.addCleanup(window.deleteLater)
self.addCleanup(window.close)
return window
def test_it_opens_where_the_missing_model_is_fixed(self):
# Nothing can transcribe on a fresh install, which is why this window
# was opened at all.
window = self.window(cfg.Config())
self.assertEqual(window.tabs.currentIndex(), window.api_tab_index)
def test_it_opens_where_it_was_left_when_everything_works(self):
conf = self.config(transcribe_provider="openai", openai_api_key="sk-test")
self.assertEqual(self.window(conf).tabs.currentIndex(), 0)
def test_a_model_that_is_not_here_yet_survives_a_save(self):
# The box is filled from what is on this disk, so a model that was
# deleted from underneath is not in the list. Dropping it on save would
# quietly empty the setting instead of asking for the download again.
conf = self.config(local_model="ggml-large-v3-turbo-q5_0.bin")
with mock.patch.object(QMessageBox, "information"):
self.window(conf)._save()
self.assertEqual(conf["local_model"], "ggml-large-v3-turbo-q5_0.bin")
def test_nothing_is_fetched_for_a_window_nobody_opened(self):
# DikteTest closes the network, so a request would fail the test. The
# lists are asked for when the box is shown, not when it is built.
window = self.window(cfg.Config())
self.assertTrue(window.local_whisper._pending)
def test_a_model_bigger_than_two_gigabytes_counts_up_rather_than_down(self):
# Qt's int is C++'s 32-bit one, and a 2.3 GB model is more than fits in
# it: the count came out the far side negative, at "-1%".
box = self.window(cfg.Config()).local_llm
box._downloading = True
box._report(1_048_576, 2_489_757_856)
_app.processEvents()
self.assertIn("2.3 GB", box.status.text())
self.assertNotIn("-", box.status.text())
def test_a_long_model_name_is_not_cut_in_half(self):
# The list under a combo box takes the box's width and elides what does
# not fit, in the middle: "ggml-org/Qwen....7B-Base-GGUF".
box = self.window(cfg.Config()).local_llm
box.repo.addItem("ggml-org/a-model-with-a-name-that-runs-on-and-on-GGUF")
box._fit_popup(box.repo)
view = box.repo.view()
self.assertEqual(view.textElideMode(), settings_ui.Qt.TextElideMode.ElideNone)
widest = max(box.repo.fontMetrics().horizontalAdvance(box.repo.itemText(row))
for row in range(box.repo.count()))
self.assertGreaterEqual(view.minimumWidth(), widest)
def test_only_the_chosen_transcriber_is_on_screen(self):
window = self.window(self.config(transcribe_provider="openai"))
self.assertTrue(window.stt_form.isRowVisible(window.transcribe_model_row))
self.assertFalse(window.stt_form.isRowVisible(window.local_whisper))
window._select_data(window.transcribe_provider, "local")
self.assertFalse(window.stt_form.isRowVisible(window.transcribe_model_row))
self.assertTrue(window.stt_form.isRowVisible(window.local_whisper))
def test_only_the_chosen_cleaner_is_on_screen(self):
window = self.window(cfg.Config())
self.assertTrue(window.cleanup_form.isRowVisible(window.cleanup_model_row))
self.assertFalse(window.cleanup_form.isRowVisible(window.local_llm))
window._select_data(window.cleanup_provider, "local")
self.assertTrue(window.cleanup_form.isRowVisible(window.local_llm))
self.assertFalse(window.cleanup_form.isRowVisible(window.cleanup_model_row))
# Its own thinking box, because the two default to opposite things.
self.assertFalse(window.cleanup_form.isRowVisible(window.cleanup_reasoning))
+1 -1
View File
@@ -234,7 +234,7 @@ class Chain(DikteTest):
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_model"])
self.assertEqual(row["model"], self.conf.transcribe_target().model)
self.assertEqual(row["mode"], "")
def test_a_command_is_recorded_as_one(self):