mirror of
https://github.com/yusufipk/dikte.git
synced 2026-09-11 19:06:11 +00:00
yusufipk
80a4832818
570 tests over the standard library's unittest, so there is nothing to install beyond the PyQt6 the application already needs. They reach neither the network, the microphone, nor the real ~/.config/dikte: urllib is faked at one function, the tools are faked at shutil.which, and every test is handed its own config and data directory. What they hold onto is what a change is most likely to move without meaning to. The request each provider is sent, field by field. A settings window that loads a value into a widget and writes it back, which is where a setting added to one half and not the other is silently reset. The dictation chain end to end: what is transcribed, what is pasted, what lands in the history, and what happens to the audio afterwards. A config file written by an older version. A meeting whose two channels heard the same sentence. 59 of them carry @linux_only, because they cover what Dikte is on this desktop rather than what it does: PipeWire, wl-clipboard, ydotool, KDE's shortcut file. The other 511 pass on any platform, and that line is worth holding as the ports arrive. CONTRIBUTING.md says how to run them, what support.py offers, and the three things about this codebase that trip up a new test.
438 lines
18 KiB
Python
438 lines
18 KiB
Python
"""The two providers, over a faked urllib.
|
|
|
|
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.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import unittest
|
|
|
|
import api
|
|
from tests.support import (
|
|
DikteTest,
|
|
fake_urlopen,
|
|
http_error,
|
|
multipart_fields,
|
|
raw_body,
|
|
sent_json,
|
|
url_error,
|
|
)
|
|
|
|
OPENAI = api.Target("openai", "OpenAI", "sk-test", api.OPENAI_URL, "gpt-4o-transcribe")
|
|
OPENROUTER = api.Target("openrouter", "OpenRouter", "sk-or-test",
|
|
api.OPENROUTER_URL, "openai/gpt-4o-transcribe")
|
|
|
|
|
|
class TimestampModel(unittest.TestCase):
|
|
def test_only_whisper_returns_segment_times(self):
|
|
self.assertEqual(api.timestamp_model("openai"), "whisper-1")
|
|
|
|
def test_openrouter_namespaces_the_id(self):
|
|
self.assertEqual(api.timestamp_model("openrouter"), "openai/whisper-1")
|
|
|
|
|
|
class Explain(DikteTest):
|
|
def error(self, status):
|
|
return api.explain(api.ApiError("HTTP", status), "OpenAI")
|
|
|
|
def test_a_rejected_key_points_at_the_settings(self):
|
|
for status in (401, 403):
|
|
with self.subTest(status=status):
|
|
message = str(self.error(status))
|
|
self.assertIn("OpenAI", message)
|
|
self.assertIn("Settings", message)
|
|
|
|
def test_no_credit(self):
|
|
self.assertIn("credit", str(self.error(402)))
|
|
|
|
def test_rate_limited(self):
|
|
self.assertIn("rate limiting", str(self.error(429)))
|
|
|
|
def test_anything_else_keeps_the_original_text(self):
|
|
explained = api.explain(api.ApiError("something broke", 500), "OpenRouter")
|
|
self.assertIn("something broke", str(explained))
|
|
self.assertEqual(explained.status, 500)
|
|
|
|
def test_the_status_is_carried_through(self):
|
|
self.assertEqual(self.error(429).status, 429)
|
|
|
|
|
|
class ExtractError(unittest.TestCase):
|
|
def test_the_usual_shape(self):
|
|
body = json.dumps({"error": {"message": "invalid model"}})
|
|
self.assertEqual(api._extract_error(body), "invalid model")
|
|
|
|
def test_an_error_that_is_a_plain_string(self):
|
|
self.assertEqual(api._extract_error(json.dumps({"error": "nope"})), "nope")
|
|
|
|
def test_an_error_object_with_no_message(self):
|
|
body = json.dumps({"error": {"code": 42}})
|
|
self.assertIn("42", api._extract_error(body))
|
|
|
|
def test_a_body_that_is_not_json(self):
|
|
self.assertEqual(api._extract_error("<html>502</html>"), "<html>502</html>")
|
|
|
|
def test_a_wall_of_html_is_cut_short(self):
|
|
self.assertEqual(len(api._extract_error("x" * 5000)), 300)
|
|
|
|
|
|
class Multipart(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")
|
|
|
|
def build(self, fields):
|
|
return api._multipart(fields, "file", self.wav)
|
|
|
|
def test_the_boundary_is_declared_and_used(self):
|
|
body, ctype = self.build([("model", "whisper-1")])
|
|
boundary = ctype.split("boundary=")[1]
|
|
self.assertTrue(ctype.startswith("multipart/form-data"))
|
|
self.assertIn(boundary.encode(), body)
|
|
self.assertTrue(body.endswith(f"--{boundary}--\r\n".encode()))
|
|
|
|
def test_a_field_is_named_and_carries_its_value(self):
|
|
body, _ = self.build([("model", "whisper-1")])
|
|
self.assertIn(b'name="model"', body)
|
|
self.assertIn(b"whisper-1", body)
|
|
|
|
def test_empty_fields_are_left_out(self):
|
|
body, _ = self.build([("model", "whisper-1"), ("language", ""),
|
|
("prompt", None)])
|
|
self.assertNotIn(b'name="language"', body)
|
|
self.assertNotIn(b'name="prompt"', body)
|
|
|
|
def test_the_file_goes_in_with_its_name_and_type(self):
|
|
body, _ = self.build([])
|
|
self.assertIn(b'filename="clip.wav"', body)
|
|
self.assertIn(b"Content-Type: audio/x-wav", body)
|
|
self.assertIn(b"RIFFfake", body)
|
|
|
|
def test_a_boundary_is_not_reused_between_requests(self):
|
|
first, _ = self.build([])
|
|
second, _ = self.build([])
|
|
self.assertNotEqual(first, second)
|
|
|
|
|
|
class Headers(unittest.TestCase):
|
|
def test_the_key_is_a_bearer_token(self):
|
|
self.assertEqual(api._headers("openai", "sk-test")["Authorization"],
|
|
"Bearer sk-test")
|
|
|
|
def test_openai_gets_no_extras(self):
|
|
self.assertNotIn("HTTP-Referer", api._headers("openai", "sk-test"))
|
|
|
|
def test_openrouter_is_told_who_is_calling(self):
|
|
headers = api._headers("openrouter", "sk-or-test")
|
|
self.assertEqual(headers["HTTP-Referer"], api.APP_URL)
|
|
self.assertEqual(headers["X-Title"], "Dikte")
|
|
|
|
def test_a_content_type_is_added_when_there_is_a_body(self):
|
|
headers = api._headers("openai", "k", "application/json")
|
|
self.assertEqual(headers["Content-Type"], "application/json")
|
|
|
|
|
|
class Transcribe(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")
|
|
|
|
def test_the_transcript_comes_back_stripped(self):
|
|
with fake_urlopen({"text": " hello there \n"}):
|
|
self.assertEqual(api.transcribe(OPENAI, self.wav), "hello there")
|
|
|
|
def test_it_goes_to_the_transcriptions_endpoint(self):
|
|
with fake_urlopen({"text": "hi"}) as calls:
|
|
api.transcribe(OPENAI, self.wav)
|
|
self.assertEqual(calls[0].full_url,
|
|
"https://api.openai.com/v1/audio/transcriptions")
|
|
|
|
def test_a_custom_base_url_is_honoured(self):
|
|
target = OPENAI._replace(base_url="http://localhost:8080/v1/")
|
|
with fake_urlopen({"text": "hi"}) as calls:
|
|
api.transcribe(target, self.wav)
|
|
self.assertEqual(calls[0].full_url,
|
|
"http://localhost:8080/v1/audio/transcriptions")
|
|
|
|
def test_the_model_and_the_format_are_sent(self):
|
|
with fake_urlopen({"text": "hi"}) as calls:
|
|
api.transcribe(OPENAI, self.wav)
|
|
fields = multipart_fields(calls[0])
|
|
self.assertEqual(fields["model"], "gpt-4o-transcribe")
|
|
self.assertEqual(fields["response_format"], "json")
|
|
|
|
def test_a_language_is_sent_but_auto_is_not(self):
|
|
with fake_urlopen({"text": "hi"}) as calls:
|
|
api.transcribe(OPENAI, self.wav, language="tr")
|
|
api.transcribe(OPENAI, self.wav, language="auto")
|
|
self.assertEqual(multipart_fields(calls[0])["language"], "tr")
|
|
self.assertNotIn("language", multipart_fields(calls[1]))
|
|
|
|
def test_the_glossary_goes_to_openai_only(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(OPENROUTER, self.wav, prompt="Paraşüt, OpenFrame")
|
|
self.assertIn("prompt", multipart_fields(calls[0]))
|
|
self.assertNotIn("prompt", multipart_fields(calls[1]))
|
|
|
|
def test_openrouter_is_attributed(self):
|
|
with fake_urlopen({"text": "hi"}) as calls:
|
|
api.transcribe(OPENROUTER, self.wav)
|
|
self.assertEqual(calls[0].get_header("X-title"), "Dikte")
|
|
|
|
def test_no_key_at_all(self):
|
|
with self.assertRaises(api.ApiError) as caught:
|
|
api.transcribe(OPENAI._replace(api_key=""), self.wav)
|
|
self.assertIn("OpenAI", str(caught.exception))
|
|
|
|
def test_an_empty_transcript_is_an_error(self):
|
|
with fake_urlopen({"text": " "}), self.assertRaises(api.ApiError):
|
|
api.transcribe(OPENAI, self.wav)
|
|
|
|
def test_a_rejected_key_is_explained_in_the_provider_s_name(self):
|
|
with fake_urlopen(http_error(401, '{"error": {"message": "bad key"}}')), \
|
|
self.assertRaises(api.ApiError) as caught:
|
|
api.transcribe(OPENROUTER, self.wav)
|
|
self.assertIn("OpenRouter", str(caught.exception))
|
|
self.assertEqual(caught.exception.status, 401)
|
|
|
|
def test_no_network(self):
|
|
with fake_urlopen(url_error("name or service not known")), \
|
|
self.assertRaises(api.ApiError) as caught:
|
|
api.transcribe(OPENAI, self.wav)
|
|
self.assertIn("connect", str(caught.exception))
|
|
|
|
def test_a_reply_that_is_not_json(self):
|
|
with fake_urlopen(raw_body("<html>bad gateway</html>")), \
|
|
self.assertRaises(api.ApiError) as caught:
|
|
api.transcribe(OPENAI, self.wav)
|
|
self.assertIn("parse", str(caught.exception))
|
|
|
|
|
|
class TranscribeSegments(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")
|
|
|
|
def reply(self, segments, text=""):
|
|
return {"segments": segments, "text": text}
|
|
|
|
def test_it_switches_to_the_model_that_has_timestamps(self):
|
|
with fake_urlopen(self.reply([{"start": 0, "end": 1, "text": "hi"}])) as calls:
|
|
api.transcribe_segments(OPENAI, self.wav)
|
|
fields = multipart_fields(calls[0])
|
|
self.assertEqual(fields["model"], "whisper-1")
|
|
self.assertEqual(fields["response_format"], "verbose_json")
|
|
self.assertEqual(fields["timestamp_granularities[]"], "segment")
|
|
|
|
def test_openrouter_uses_the_namespaced_id(self):
|
|
with fake_urlopen(self.reply([{"start": 0, "end": 1, "text": "hi"}])) as calls:
|
|
api.transcribe_segments(OPENROUTER, self.wav)
|
|
self.assertEqual(multipart_fields(calls[0])["model"], "openai/whisper-1")
|
|
|
|
def test_the_segments_come_back_as_numbers(self):
|
|
with fake_urlopen(self.reply([
|
|
{"start": "0.5", "end": "2.25", "text": " hello "},
|
|
{"start": 2.25, "end": 4.0, "text": "there"},
|
|
])):
|
|
segments = api.transcribe_segments(OPENAI, self.wav)
|
|
self.assertEqual(segments, [(0.5, 2.25, "hello"), (2.25, 4.0, "there")])
|
|
|
|
def test_empty_segments_are_dropped(self):
|
|
with fake_urlopen(self.reply([
|
|
{"start": 0, "end": 1, "text": " "},
|
|
{"start": 1, "end": 2, "text": "real"},
|
|
])):
|
|
self.assertEqual(api.transcribe_segments(OPENAI, self.wav),
|
|
[(1.0, 2.0, "real")])
|
|
|
|
def test_an_end_before_its_start_is_pulled_forward(self):
|
|
with fake_urlopen(self.reply([{"start": 5, "end": 1, "text": "hi"}])):
|
|
self.assertEqual(api.transcribe_segments(OPENAI, self.wav),
|
|
[(5.0, 5.0, "hi")])
|
|
|
|
def test_a_model_that_returned_no_segments_still_gives_its_text(self):
|
|
with fake_urlopen(self.reply([], text="the whole thing")):
|
|
self.assertEqual(api.transcribe_segments(OPENAI, self.wav),
|
|
[(0.0, 0.0, "the whole thing")])
|
|
|
|
def test_nothing_at_all(self):
|
|
with fake_urlopen(self.reply([], text="")), \
|
|
self.assertRaises(api.ApiError):
|
|
api.transcribe_segments(OPENAI, self.wav)
|
|
|
|
|
|
def chat_reply(content):
|
|
return {"choices": [{"message": {"content": content}}]}
|
|
|
|
|
|
class Cleanup(DikteTest):
|
|
def call(self, replies, **kwargs):
|
|
with fake_urlopen(replies) as calls:
|
|
result = api.cleanup("uh, hello", "sk-or-test", "some/model",
|
|
"you clean up text", **kwargs)
|
|
return result, calls
|
|
|
|
def test_the_cleaned_text_comes_back(self):
|
|
result, _ = self.call(chat_reply(" Hello. "))
|
|
self.assertEqual(result, "Hello.")
|
|
|
|
def test_it_goes_to_chat_completions(self):
|
|
_, calls = self.call(chat_reply("Hello."))
|
|
self.assertEqual(calls[0].full_url,
|
|
"https://openrouter.ai/api/v1/chat/completions")
|
|
|
|
def test_the_prompt_and_the_transcript_are_kept_apart(self):
|
|
_, calls = self.call(chat_reply("Hello."))
|
|
payload = sent_json(calls[0])
|
|
self.assertEqual(payload["messages"][0]["role"], "system")
|
|
self.assertEqual(payload["messages"][0]["content"], "you clean up text")
|
|
self.assertIn("<transcript>", payload["messages"][1]["content"])
|
|
self.assertIn("uh, hello", payload["messages"][1]["content"])
|
|
|
|
def test_the_temperature_is_pinned(self):
|
|
_, calls = self.call(chat_reply("Hello."))
|
|
self.assertEqual(sent_json(calls[0])["temperature"], 0)
|
|
|
|
def test_no_effort_asked_for_means_no_reasoning_block(self):
|
|
_, calls = self.call(chat_reply("Hello."))
|
|
self.assertNotIn("reasoning", sent_json(calls[0]))
|
|
|
|
def test_an_effort_is_passed_on_and_the_thinking_left_out(self):
|
|
_, calls = self.call(chat_reply("Hello."), reasoning="high")
|
|
self.assertEqual(sent_json(calls[0])["reasoning"],
|
|
{"effort": "high", "exclude": True})
|
|
|
|
def test_a_local_base_url(self):
|
|
_, calls = self.call(chat_reply("Hello."), base_url="http://localhost:1234/v1")
|
|
self.assertEqual(calls[0].full_url, "http://localhost:1234/v1/chat/completions")
|
|
|
|
def test_no_key(self):
|
|
with self.assertRaises(api.ApiError):
|
|
api.cleanup("hello", "", "some/model", "prompt")
|
|
|
|
def test_a_reply_with_no_choices_says_why(self):
|
|
with fake_urlopen({"error": {"message": "model is offline"}}), \
|
|
self.assertRaises(api.ApiError) as caught:
|
|
api.cleanup("hello", "k", "m", "p")
|
|
self.assertIn("model is offline", str(caught.exception))
|
|
|
|
def test_an_empty_answer(self):
|
|
with fake_urlopen(chat_reply(" ")), self.assertRaises(api.ApiError):
|
|
api.cleanup("hello", "k", "m", "p")
|
|
|
|
def test_a_rate_limit_is_explained(self):
|
|
with fake_urlopen(http_error(429)), \
|
|
self.assertRaises(api.ApiError) as caught:
|
|
api.cleanup("hello", "k", "m", "p")
|
|
self.assertIn("OpenRouter", str(caught.exception))
|
|
|
|
|
|
class Chat(DikteTest):
|
|
def test_the_history_is_sent_after_the_system_prompt(self):
|
|
history = [{"role": "user", "content": "book it"},
|
|
{"role": "assistant", "content": "done"}]
|
|
with fake_urlopen(chat_reply("moved it")) as calls:
|
|
api.chat(history + [{"role": "user", "content": "move it"}],
|
|
"k", "some/model", "you are an agent")
|
|
payload = sent_json(calls[0])
|
|
self.assertEqual(payload["messages"][0],
|
|
{"role": "system", "content": "you are an agent"})
|
|
self.assertEqual(payload["messages"][1:], history +
|
|
[{"role": "user", "content": "move it"}])
|
|
|
|
def test_no_temperature_is_forced_on_a_conversation(self):
|
|
with fake_urlopen(chat_reply("hi")) as calls:
|
|
api.chat([{"role": "user", "content": "hi"}], "k", "m", "p")
|
|
self.assertNotIn("temperature", sent_json(calls[0]))
|
|
|
|
def test_no_key(self):
|
|
with self.assertRaises(api.ApiError):
|
|
api.chat([], "", "m", "p")
|
|
|
|
def test_an_empty_answer(self):
|
|
with fake_urlopen(chat_reply("")), self.assertRaises(api.ApiError):
|
|
api.chat([{"role": "user", "content": "hi"}], "k", "m", "p")
|
|
|
|
|
|
class KeyStatus(DikteTest):
|
|
def test_a_key_with_no_limit(self):
|
|
with fake_urlopen({"data": {"limit": None, "usage": 3}}):
|
|
self.assertIn("no spending limit",
|
|
api.openrouter_key_status("sk-or-test"))
|
|
|
|
def test_a_key_with_a_limit_reports_both_numbers(self):
|
|
with fake_urlopen({"data": {"limit": 10, "usage": 2.5}}):
|
|
message = api.openrouter_key_status("sk-or-test")
|
|
self.assertIn("2.5", message)
|
|
self.assertIn("10", message)
|
|
|
|
def test_no_key(self):
|
|
with self.assertRaises(api.ApiError):
|
|
api.openrouter_key_status("")
|
|
|
|
def test_a_key_the_service_rejects(self):
|
|
with fake_urlopen(http_error(401)), \
|
|
self.assertRaises(api.ApiError) as caught:
|
|
api.openrouter_key_status("sk-or-bad")
|
|
self.assertEqual(caught.exception.status, 401)
|
|
|
|
|
|
class ModelLists(DikteTest):
|
|
def test_openrouter_returns_sorted_ids(self):
|
|
with fake_urlopen({"data": [{"id": "z/model"}, {"id": "a/model"}]}):
|
|
self.assertEqual(api.openrouter_models(), ["a/model", "z/model"])
|
|
|
|
def test_the_model_list_needs_no_key(self):
|
|
with fake_urlopen({"data": []}) as calls:
|
|
api.openrouter_models()
|
|
self.assertIsNone(calls[0].get_header("Authorization"))
|
|
|
|
def test_a_key_is_sent_when_there_is_one(self):
|
|
with fake_urlopen({"data": []}) as calls:
|
|
api.openrouter_models("sk-or-test")
|
|
self.assertEqual(calls[0].get_header("Authorization"), "Bearer sk-or-test")
|
|
|
|
def test_speech_models_are_asked_for_and_filtered_again(self):
|
|
"""A query parameter the API stops honouring must not leak the lot."""
|
|
with fake_urlopen({"data": [
|
|
{"id": "openai/whisper-1",
|
|
"architecture": {"output_modalities": ["transcription"]}},
|
|
{"id": "google/gemini-3.5-flash",
|
|
"architecture": {"output_modalities": ["text"]}},
|
|
{"id": "broken/model"},
|
|
]}) as calls:
|
|
models = api.openrouter_models(transcription=True)
|
|
self.assertIn("output_modalities=transcription", calls[0].full_url)
|
|
self.assertEqual(models, ["openai/whisper-1"])
|
|
|
|
def test_openai_narrows_to_the_audio_models(self):
|
|
with fake_urlopen({"data": [{"id": "gpt-4o"}, {"id": "whisper-1"},
|
|
{"id": "gpt-4o-transcribe"}]}):
|
|
self.assertEqual(api.openai_models("sk-test"),
|
|
["gpt-4o-transcribe", "whisper-1"])
|
|
|
|
def test_a_list_with_no_audio_models_is_shown_whole(self):
|
|
with fake_urlopen({"data": [{"id": "gpt-4o"}, {"id": "o3"}]}):
|
|
self.assertEqual(api.openai_models("sk-test"), ["gpt-4o", "o3"])
|
|
|
|
def test_openai_needs_a_key(self):
|
|
with self.assertRaises(api.ApiError):
|
|
api.openai_models("")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|