mirror of
https://github.com/yusufipk/dikte.git
synced 2026-09-11 10:56:10 +00:00
yusufipk
2cfbbb2d99
whisper-server is started on --inference-path /v1/audio/transcriptions, which is exactly the path api.py already builds for the hosted providers, and llama-server answers /chat/completions the way OpenRouter does. So the local half is one more base URL rather than a second code path: worker.py, filetranscribe.py and meeting.py are untouched, and dictation, subtitles and meetings all work here on the first try. Three findings worth naming, none of them in the new code: whisper.cpp cuts segments on tokens, which in Turkish lands inside a word about as often as between two. Pasted raw that gives "akraba değ\niller."; in a subtitle it gives a cue reading "değ". Whisper marks the start of a word with a leading space, so a piece that does not begin with one continues the word above it. A small model will repeat the transcript until the context is full, and every one of those tokens is a second of somebody waiting: measured at 206 seconds, and 25 with a ceiling on the reply. Hosted models are left alone, where the same runaway is rare and a ceiling would cut the minutes short. A server outlives SIGTERM and SIGKILL holding its model in memory. Signals are now turned into an event Qt delivers, since Qt blocks in C where a Python handler never runs, and a pid file lets the next start sweep up what a SIGKILL left behind. The minutes keep their own provider rather than following cleanup's. The two jobs are not the same size: a 4B model here will strip the filler words out of a dictation and will not write up an hour long meeting. The suite runs offline now: a test that reaches the network says so instead of quietly going there.
610 lines
26 KiB
Python
610 lines
26 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
|
||
import ggml
|
||
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", "gpt-4o-transcribe"),
|
||
"whisper-1")
|
||
|
||
def test_openrouter_namespaces_the_id(self):
|
||
self.assertEqual(api.timestamp_model("openrouter", "openai/gpt-4o-transcribe"),
|
||
"openai/whisper-1")
|
||
|
||
def test_the_local_server_stays_on_the_model_it_loaded(self):
|
||
# Asking it for whisper-1 would name a model it has never heard of, and
|
||
# it is running whisper whatever the file is called.
|
||
self.assertEqual(api.timestamp_model("local", "ggml-base.bin"),
|
||
"ggml-base.bin")
|
||
|
||
|
||
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}}]}
|
||
|
||
|
||
def openrouter(model="some/model", key="sk-or-test", reasoning="",
|
||
base_url="https://openrouter.ai/api/v1"):
|
||
return api.Target("openrouter", "OpenRouter", key, base_url, model, reasoning)
|
||
|
||
|
||
class Cleanup(DikteTest):
|
||
def call(self, replies, target=None, **kwargs):
|
||
with fake_urlopen(replies) as calls:
|
||
result = api.cleanup(target or openrouter(), "uh, hello",
|
||
"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."),
|
||
target=openrouter(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."),
|
||
target=openrouter(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(openrouter(key=""), "hello", "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(openrouter(), "hello", "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(openrouter(), "hello", "p")
|
||
|
||
def test_a_rate_limit_is_explained(self):
|
||
with fake_urlopen(http_error(429)), \
|
||
self.assertRaises(api.ApiError) as caught:
|
||
api.cleanup(openrouter(), "hello", "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()
|
||
|
||
|
||
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")
|
||
LOCAL_LLM = api.Target("local-llm", "Local model", "", "", "gemma.gguf", "none")
|
||
|
||
|
||
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 CleanupHere(DikteTest):
|
||
def setUp(self):
|
||
super().setUp()
|
||
self.server = FakeServer("http://127.0.0.1:8888/v1")
|
||
self.patch_attr(ggml, "llm", self.server)
|
||
|
||
def test_it_goes_to_the_server_it_starts(self):
|
||
with fake_urlopen(chat_reply("Hello.")) as calls:
|
||
result = api.cleanup(LOCAL_LLM, "uh, hello", "clean it up")
|
||
self.assertEqual(result, "Hello.")
|
||
self.assertEqual(calls[0].full_url,
|
||
"http://127.0.0.1:8888/v1/chat/completions")
|
||
|
||
def test_no_key_is_wanted_and_none_is_sent(self):
|
||
with fake_urlopen(chat_reply("Hello.")) as calls:
|
||
api.cleanup(LOCAL_LLM, "hello", "prompt")
|
||
self.assertNotIn("Authorization", calls[0].headers)
|
||
|
||
def test_thinking_is_turned_off_in_the_words_llama_cpp_uses(self):
|
||
with fake_urlopen(chat_reply("Hello.")) as calls:
|
||
api.cleanup(LOCAL_LLM, "hello", "prompt")
|
||
self.assertEqual(sent_json(calls[0])["chat_template_kwargs"],
|
||
{"enable_thinking": False})
|
||
|
||
def test_the_models_own_default_asks_for_nothing(self):
|
||
with fake_urlopen(chat_reply("Hello.")) as calls:
|
||
api.cleanup(LOCAL_LLM._replace(reasoning=""), "hello", "prompt")
|
||
self.assertNotIn("chat_template_kwargs", sent_json(calls[0]))
|
||
|
||
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:
|
||
api.cleanup(LOCAL_LLM, "hello", "prompt")
|
||
self.assertIn("Thinking", str(caught.exception))
|
||
|
||
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("Hello.")) as calls:
|
||
api.cleanup(LOCAL_LLM, "x" * 4000, "prompt")
|
||
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("Hello.")) as calls:
|
||
api.cleanup(LOCAL_LLM, "uh, hi", "prompt")
|
||
self.assertEqual(sent_json(calls[0])["max_tokens"], 512)
|
||
|
||
def test_a_hosted_model_is_left_to_answer_at_length(self):
|
||
with fake_urlopen(chat_reply("Hello.")) as calls:
|
||
api.cleanup(openrouter(), "uh, hi", "prompt")
|
||
self.assertNotIn("max_tokens", sent_json(calls[0]))
|
||
|
||
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:
|
||
api.cleanup(LOCAL_LLM, "hello", "prompt")
|
||
self.assertIn("llama.cpp", str(caught.exception))
|