diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..e8f0798 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,40 @@ +name: tests + +on: + push: + branches: [master] + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python: ["3.11", "3.12", "3.13"] + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + + # PyQt6 ships Qt itself, but Qt still loads these from the system, even + # for the offscreen platform the tests run on. QtNetwork wants the Kerberos + # library, and the widgets want fontconfig, whether or not anything is + # ever drawn. + - name: Install the Qt runtime libraries + run: | + sudo apt-get update + sudo apt-get install --no-install-recommends -y \ + libegl1 libgl1 libxkbcommon0 libdbus-1-3 libglib2.0-0 \ + libfontconfig1 libfreetype6 libgssapi-krb5-2 + + - name: Install PyQt6 + run: python -m pip install --quiet PyQt6 + + # Nothing else is needed: the code is standard library and PyQt6, and the + # tests reach neither the network nor a sound device. + - name: Run the tests + run: python -m unittest discover --verbose diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..5894a68 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,90 @@ +# Contributing + +## Running the tests + +```sh +python -m unittest discover # all of them, about a second +python -m unittest tests.test_api # one file +python -m unittest tests.test_api.Transcribe.test_no_key_at_all +``` + +Nothing to install: the tests use the standard library's `unittest`, and the +only dependency is the PyQt6 the application already needs. They reach neither +the network, the microphone, nor your real `~/.config/dikte`, so they are safe +to run anywhere and they run on a machine with no display. + +CI runs the same command on Python 3.11 through 3.13. A pull request that turns +it red will not be merged. + +## Writing one + +Put it in `tests/`, named after the module it covers. Inherit from +`tests.support.DikteTest` whenever the code under test touches a file, a +setting or the interface language: it hands the test its own config and data +directories, resets the language, and puts them back afterwards. + +`tests/support.py` has the rest of what you need: + +| For | Use | +| --- | --- | +| An HTTP call | `fake_urlopen(reply, …)`, then read the recorded requests | +| A reply that fails | `http_error(429)`, `url_error()`, `raw_body("not json")` | +| Reading what was sent | `sent_json(request)`, `multipart_fields(request)` | +| A program on the PATH | `only_these_tools("pactl", "wl-copy")` | +| Audio | `silence()`, `tone()`, `speech()`, `stereo()`, `make_wav()` | +| A settings object | `self.config(cleanup_enabled=False)` | + +Three things about this codebase trip up a new test: + +**Signals from a worker thread are never delivered.** `Pipeline`, `MeetingPipeline` +and `FileTranscriber` emit from the thread `start()` spawned, which Qt queues +until an event loop runs one. Call `_work()` directly instead: it is the same +code one frame down, and the signals arrive at once. + +**A level that never moves is not speech.** The silence check is relative, so a +steady tone reads as its own noise floor however loud it is. Use `speech()` +rather than `tone()` when a recording is meant to have somebody talking in it. + +**`cli.launch_gui` replaces the process.** With no instance running, some verbs +`os.execv` into the application, which would take the test run with it. Patch +`cli.launch_gui`. `DikteTest` blocks `os.execv` as a backstop, so a test that +forgets fails rather than hangs. + +## Another platform + +Most of what Dikte does is not desktop-specific, and the tests are split along +that line. 511 of them pass anywhere: transcription, cleanup, the config file, +the history, the agent, the command line, the timeline of a meeting. The +remaining 59 cover what Dikte *is* on this desktop, and carry `@linux_only` +from `tests.support`: PipeWire capture and the pactl device list, wl-clipboard +and ydotool, KDE's shortcut file and the `/dev/input` listener. + +Mark a test `@linux_only` when it would fail on a machine that never had those +programs. Do not mark one because it happens to be convenient: a test that +quietly stops running on the platform you are porting to protects nothing. + +The other half of a port is where the branch goes. Keep `sys.platform` out of +the middle of a function; make the public name a chooser and give each platform +its own function underneath: + +```python +def copy(text): + return _copy_macos(text) if sys.platform == "darwin" else _copy_wayland(text) +``` + +Then each platform's test calls its own function directly and passes everywhere, +and adding a third one leaves the first two's tests alone. An `if` buried inside +`copy()` forces every existing test to patch `sys.platform` instead, and the +next port breaks all of them. + +## What a pull request should carry + +A change to behaviour comes with a test for it. Adding a provider means a test +that the request goes to the right URL with the right fields; adding a platform +means a test for whatever the parsing of its device list, clipboard or shortcuts +looks like. Adding a setting means both halves of `settings_ui.py`: the round +trip in `tests/test_ui.py` is what catches only one of them being written. + +Match the surrounding code: it is plain Python with no framework, comments +explain why rather than what, and neither the code nor the commit messages use +an em dash. diff --git a/cli.py b/cli.py index 85a63dc..b584458 100644 --- a/cli.py +++ b/cli.py @@ -361,9 +361,11 @@ def _find_meeting(which): return None if which in ("", "last"): return rows[-1] - if which.isdigit(): - index = int(which) - return rows[-index] if 0 < index <= len(rows) else None + # A stem is all digits too, so a number is only a position while there are + # that many meetings to count back through. Anything larger is a date + # somebody typed: nobody is looking for the twenty-millionth meeting. + if which.isdigit() and 0 < int(which) <= len(rows): + return rows[-int(which)] exact = [row for row in rows if row["base"] == which] if exact: return exact[0] diff --git a/i18n.py b/i18n.py index 13ec3d2..f2b1a70 100644 --- a/i18n.py +++ b/i18n.py @@ -27,7 +27,10 @@ def language(): return _lang -def t(text, **kwargs): +def t(text, /, **kwargs): + # The string is positional-only so that every name is free to be a + # placeholder: t("Discarded: {text}", text=…) would otherwise be two values + # for one argument, and fail at the moment the message is shown. out = TR.get(text, text) if _lang == "tr" else text return out.format(**kwargs) if kwargs else out @@ -42,7 +45,7 @@ _TR_CASES = { } -def name(text, case=""): +def name(text, /, case=""): if _lang != "tr" or not case: return text return _TR_CASES.get(case, {}).get(text, text) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..d422cec --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,31 @@ +"""Test-wide safety net, applied before anything under test is imported. + +Every module resolves its paths at import time from the XDG variables, so those +are redirected here: a test that forgets to ask for a throwaway directory writes +into a temporary one instead of into the real ~/.config/dikte. The Qt platform +is pinned for the same reason, so that a machine with no display and a CI runner +behave the way a desktop does. +""" + +import atexit +import os +import shutil +import tempfile + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +_SANDBOX = tempfile.mkdtemp(prefix="dikte-tests-") +os.environ["XDG_CONFIG_HOME"] = os.path.join(_SANDBOX, "config") +os.environ["XDG_DATA_HOME"] = os.path.join(_SANDBOX, "data") +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"): + os.environ.pop(_var, None) + +# The interface language leaks through module-level state, so the tests fix it +# rather than inherit whatever the developer's locale says. +for _var in ("LC_ALL", "LC_MESSAGES", "LANG"): + os.environ.pop(_var, None) diff --git a/tests/support.py b/tests/support.py new file mode 100644 index 0000000..b199940 --- /dev/null +++ b/tests/support.py @@ -0,0 +1,266 @@ +"""What the tests share: a throwaway home, a fake network, small WAV files. + +Two things about this codebase shape all of it. Paths are module-level constants +resolved at import time, so they are replaced object by object rather than +re-derived by reloading the module, which would hand every other module a second +copy of it. And the only way out to the network is urllib, so faking one function +is enough to run the whole chain offline. +""" + +import array +import contextlib +import io +import json +import math +import os +import shutil +import sys +import tempfile +import unittest +import urllib.error +import wave +from unittest import mock + +import assistant +import config as cfg +import i18n + +# What the application is, rather than what it does: PipeWire, wl-clipboard, +# ydotool, KDE's shortcut file, /dev/input. A port to another desktop replaces +# all of it, and the tests that pin this half say so rather than failing on a +# machine that never had any of it. +# +# Everything else is expected to pass everywhere, and that is the line worth +# holding: transcription, cleanup, the config file, the history, the agent, the +# command line and the timeline of a meeting are not desktop-specific and must +# not become so. +linux_only = unittest.skipUnless( + sys.platform.startswith("linux"), + "covers the Linux desktop stack (PipeWire, wl-clipboard, ydotool, KDE)", +) + + +def _no_exec(*args, **kwargs): + raise AssertionError( + "a test reached os.execv, which would replace the test process with the " + "application; patch cli.launch_gui instead" + ) + + +class DikteTest(unittest.TestCase): + """A test that owns its config, its data directory and its language.""" + + def setUp(self): + super().setUp() + self.root = tempfile.mkdtemp(prefix="dikte-test-") + self.addCleanup(shutil.rmtree, self.root, True) + + config_dir = self.path("config", "dikte") + data_dir = self.path("data", "dikte") + self.patch_paths( + CONFIG_DIR=config_dir, + CONFIG_FILE=config_dir / "config.json", + DATA_DIR=data_dir, + HISTORY_FILE=data_dir / "history.jsonl", + RECORDINGS_DIR=data_dir / "recordings", + MEETINGS_DIR=data_dir / "meetings", + MEETINGS_FILE=data_dir / "meetings.jsonl", + ) + # Resolved from cfg.DATA_DIR when assistant was imported, so it needs + # moving on its own. + self.patch_attr(assistant, "SESSION_FILE", data_dir / "assistant.json") + + i18n.set_language("en") + self.addCleanup(i18n.set_language, "en") + + # cli.launch_gui replaces this process with the application when no + # instance is running. A test that reaches it would take the whole run + # with it and hang, so it fails loudly here instead. + self.patch_attr(os, "execv", _no_exec) + + # ---- helpers --------------------------------------------------------- + + def path(self, *parts): + """A path inside this test's directory, as a pathlib.Path.""" + import pathlib + return pathlib.Path(self.root, *parts) + + def patch_paths(self, **paths): + patcher = mock.patch.multiple(cfg, **paths) + patcher.start() + self.addCleanup(patcher.stop) + + def patch_attr(self, target, name, value): + patcher = mock.patch.object(target, name, value) + patcher.start() + self.addCleanup(patcher.stop) + return value + + def config(self, **values): + """A Config with nothing stored, then the given settings applied.""" + conf = cfg.Config() + for key, value in values.items(): + conf[key] = value + return conf + + def write_config(self, payload): + """Put a config.json on disk, the way an older version would have.""" + cfg.CONFIG_DIR.mkdir(parents=True, exist_ok=True) + cfg.CONFIG_FILE.write_text(json.dumps(payload), encoding="utf-8") + + def read_config_file(self): + return json.loads(cfg.CONFIG_FILE.read_text(encoding="utf-8")) + + +# --- the network ---------------------------------------------------------- + + +def json_body(payload): + """A stand-in for what urlopen hands back: a context manager that reads.""" + body = json.dumps(payload).encode("utf-8") + resp = mock.MagicMock() + resp.read.return_value = body + resp.__enter__.return_value = resp + resp.__exit__.return_value = False + return resp + + +def raw_body(text): + """The same, for a reply that is not valid JSON.""" + resp = mock.MagicMock() + resp.read.return_value = text.encode("utf-8") + resp.__enter__.return_value = resp + resp.__exit__.return_value = False + return resp + + +def http_error(code, body=""): + return urllib.error.HTTPError( + "https://example.invalid/v1", code, "boom", {}, + io.BytesIO(body.encode("utf-8")), + ) + + +def url_error(reason="no route to host"): + return urllib.error.URLError(reason) + + +@contextlib.contextmanager +def fake_urlopen(*replies): + """Answer each call with the next reply; the last one repeats. + + A reply is a payload to encode as JSON, an exception to raise, or an object + already shaped like a response. The requests are collected so a test can + check what was actually sent. + """ + calls = [] + + def opener(req, timeout=None): + calls.append(req) + reply = replies[min(len(calls) - 1, len(replies) - 1)] if replies else {} + if isinstance(reply, Exception): + raise reply + if isinstance(reply, (dict, list)): + return json_body(reply) + return reply + + try: + with mock.patch("urllib.request.urlopen", side_effect=opener): + yield calls + finally: + # An HTTPError holds a file object and complains when it is collected + # without one; the tests raise the same one more than once, so closing + # it is the caller's job rather than the code's. + for reply in replies: + if isinstance(reply, urllib.error.HTTPError): + reply.close() + + +def sent_json(request): + """The JSON body of a recorded request.""" + return json.loads(request.data.decode("utf-8")) + + +def multipart_fields(request): + """{name: value} for the plain fields of a recorded multipart request.""" + body = request.data.decode("utf-8", "replace") + fields = {} + for part in body.split("\r\n--"): + if 'name="' not in part or "filename=" in part: + continue + name = part.split('name="', 1)[1].split('"', 1)[0] + _, _, value = part.partition("\r\n\r\n") + fields[name] = value.rstrip("\r\n") + return fields + + +# --- audio ---------------------------------------------------------------- + + +def pcm(samples): + return array.array("h", samples).tobytes() + + +def tone(seconds, rate=16000, amplitude=8000, channels=1, freq=440.0): + """Interleaved s16 samples for a sine wave, the same on every channel.""" + frames = int(seconds * rate) + out = array.array("h") + for index in range(frames): + value = int(amplitude * math.sin(2 * math.pi * freq * index / rate)) + out.extend([value] * channels) + return out.tobytes() + + +def silence(seconds, rate=16000, channels=1): + return b"\x00\x00" * int(seconds * rate) * channels + + +def speech(seconds, rate=16000, amplitude=16000, freq=440.0): + """A buffer the silence check reads as somebody talking. + + A steady tone does not, however loud it is: the check is relative, and a + level that never moves is its own noise floor. Speech is quiet, then loud, + which is what the pauses between words make it. + """ + half = seconds / 2 + return silence(half, rate) + tone(half, rate, amplitude, freq=freq) + + +def make_wav(path, data, rate=16000, channels=1, width=2): + os.makedirs(os.path.dirname(str(path)) or ".", exist_ok=True) + with contextlib.closing(wave.open(str(path), "wb")) as wav: + wav.setnchannels(channels) + wav.setsampwidth(width) + wav.setframerate(rate) + wav.writeframes(data) + return str(path) + + +def stereo(left, right): + """Interleave two equal-length mono buffers into one stereo buffer.""" + a, b = array.array("h"), array.array("h") + a.frombytes(left) + b.frombytes(right) + out = array.array("h") + for first, second in zip(a, b): + out.extend((first, second)) + return out.tobytes() + + +# --- processes ------------------------------------------------------------ + + +class FakeCompleted: + """What subprocess.run hands back, as much of it as the code reads.""" + + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +def only_these_tools(*names): + """shutil.which answers for the named tools and nothing else.""" + wanted = set(names) + return mock.patch("shutil.which", side_effect=lambda tool: ( + f"/usr/bin/{tool}" if tool in wanted else None)) diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..30e7e3e --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,437 @@ +"""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("502"), "502") + + 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("bad gateway")), \ + 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("", 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() diff --git a/tests/test_assistant.py b/tests/test_assistant.py new file mode 100644 index 0000000..8fee724 --- /dev/null +++ b/tests/test_assistant.py @@ -0,0 +1,535 @@ +"""Handing a dictation to an agent. + +Three providers behind one setting, so most of this is about the command line +each of them is given and about the conversation carried between dictations. The +CLIs are faked at subprocess.Popen: what the tests read is the argument list and +what the stream of JSON events is turned into. +""" + +import io +import json +import os +import subprocess +import time +import unittest +from unittest import mock + +import assistant +from tests.support import DikteTest, fake_urlopen, only_these_tools + + +class FakeCli: + """A CLI that prints the given events and exits.""" + + def __init__(self, events=(), code=0, stderr="", noise=()): + lines = list(noise) + [json.dumps(event) for event in events] + self.stdout = io.StringIO("\n".join(lines) + "\n") + self.stderr = io.StringIO(stderr) + self.returncode = code + self.killed = False + + def poll(self): + return self.returncode + + def wait(self, timeout=None): + return self.returncode + + def terminate(self): + self.killed = True + + def kill(self): + self.killed = True + + +class Provider(DikteTest): + def test_the_default(self): + self.assertEqual(assistant.provider(self.config()), "claude") + + def test_a_provider_this_version_does_not_have(self): + self.assertEqual( + assistant.provider(self.config(assistant_provider="ollama")), "claude") + + def test_each_one_is_recognised(self): + for name in assistant.PROVIDERS: + with self.subTest(name=name): + self.assertEqual( + assistant.provider(self.config(assistant_provider=name)), name) + + def test_what_each_one_runs(self): + self.assertEqual(assistant.executable("claude"), "claude") + self.assertEqual(assistant.executable("codex"), "codex") + self.assertEqual(assistant.executable("openrouter"), "") + + def test_what_each_one_is_called(self): + self.assertEqual(assistant.display_name(self.config()), "Claude") + self.assertEqual( + assistant.display_name(self.config(assistant_provider="codex")), "Codex") + self.assertEqual( + assistant.display_name(self.config(assistant_provider="openrouter")), + "OpenRouter") + + +class Effort(unittest.TestCase): + """One scale, offered once; a rung a provider lacks lands on its nearest.""" + + def test_the_scales_cover_the_same_settings(self): + self.assertEqual(set(assistant.CLAUDE_EFFORT), set(assistant.CODEX_EFFORT)) + + def test_codex_has_no_rung_above_high(self): + 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_an_empty_setting_asks_for_nothing(self): + self.assertEqual(assistant.CLAUDE_EFFORT.get("", ""), "") + self.assertEqual(assistant.CODEX_EFFORT.get("", ""), "") + + +class Session(DikteTest): + def test_nothing_stored_yet(self): + self.assertEqual(assistant.read_session("claude", 1800), "") + self.assertEqual(assistant.read_messages("openrouter", 1800), []) + self.assertEqual(assistant.stored_provider(), "") + self.assertIsNone(assistant.session_age()) + + def test_an_id_is_written_and_read_back(self): + assistant.write_session("claude", "abc-123") + self.assertEqual(assistant.read_session("claude", 1800), "abc-123") + self.assertEqual(assistant.stored_provider(), "claude") + + def test_nobody_picks_up_another_provider_s_thread(self): + assistant.write_session("claude", "abc-123") + self.assertEqual(assistant.read_session("codex", 1800), "") + + def test_a_conversation_that_has_sat_unused_is_dropped(self): + assistant.write_session("claude", "abc-123") + with mock.patch.object(time, "time", return_value=time.time() + 3600): + self.assertEqual(assistant.read_session("claude", 1800), "") + + def test_a_session_that_never_expires(self): + assistant.write_session("claude", "abc-123") + with mock.patch.object(time, "time", return_value=time.time() + 10 ** 6): + self.assertEqual(assistant.read_session("claude", 0), "abc-123") + + def test_the_messages_of_the_provider_that_keeps_none(self): + messages = [{"role": "user", "content": "hi"}] + assistant.write_session("openrouter", messages=messages) + self.assertEqual(assistant.read_messages("openrouter", 1800), messages) + + def test_the_history_window_ends_somewhere(self): + messages = [{"role": "user", "content": str(index)} for index in range(50)] + assistant.write_session("openrouter", messages=messages) + stored = assistant.read_messages("openrouter", 1800) + self.assertEqual(len(stored), assistant.MAX_HISTORY) + self.assertEqual(stored[-1]["content"], "49") + + def test_the_age_of_the_conversation(self): + assistant.write_session("claude", "abc-123") + self.assertLess(assistant.session_age(), 5) + + def test_a_row_with_neither_an_id_nor_messages_has_no_age(self): + assistant.write_session("claude", "") + self.assertIsNone(assistant.session_age()) + + def test_clearing(self): + assistant.write_session("claude", "abc-123") + assistant.clear_session() + self.assertEqual(assistant.read_session("claude", 1800), "") + + def test_clearing_one_that_is_not_there(self): + assistant.clear_session() # must not raise + + def test_a_session_file_that_is_not_json(self): + assistant.SESSION_FILE.parent.mkdir(parents=True, exist_ok=True) + assistant.SESSION_FILE.write_text("{oh dear", encoding="utf-8") + self.assertEqual(assistant.read_session("claude", 1800), "") + self.assertEqual(assistant.stored_provider(), "") + self.assertIsNone(assistant.session_age()) + + +class WorkingDir(DikteTest): + def test_the_home_directory_by_default(self): + self.assertEqual(assistant.working_dir(self.config()), + os.path.expanduser("~")) + + def test_a_directory_of_your_own(self): + conf = self.config(assistant_dir=self.root) + self.assertEqual(assistant.working_dir(conf), self.root) + + def test_a_tilde_is_expanded(self): + conf = self.config(assistant_dir="~") + self.assertEqual(assistant.working_dir(conf), os.path.expanduser("~")) + + def test_a_directory_that_is_not_there_falls_back(self): + conf = self.config(assistant_dir="/no/such/place") + self.assertEqual(assistant.working_dir(conf), os.path.expanduser("~")) + + +class Labels(DikteTest): + def test_a_tool_the_table_knows(self): + self.assertEqual(assistant._claude_label({"name": "Bash"}), + "Running a command…") + + def test_a_tool_arriving_from_an_mcp_server_is_named_by_its_server(self): + self.assertIn("gmail", assistant._claude_label({"name": "mcp__gmail__send"})) + + def test_a_skill_is_named_by_the_skill(self): + label = assistant._claude_label( + {"name": "Skill", "input": {"skill": "calendar"}}) + self.assertIn("calendar", label) + + def test_a_tool_nobody_wrote_a_line_for(self): + self.assertIn("SomeNewTool", + assistant._claude_label({"name": "SomeNewTool"})) + + def test_a_tool_with_no_name_at_all(self): + self.assertTrue(assistant._claude_label({})) + + def test_the_codex_table(self): + self.assertEqual(assistant._codex_label({"type": "command_execution"}), + "Running a command…") + + def test_a_codex_mcp_call(self): + self.assertIn("gmail", assistant._codex_label( + {"type": "mcp_tool_call", "server": "gmail"})) + + def test_a_codex_item_nobody_listed(self): + self.assertIn("something_new", + assistant._codex_label({"type": "something_new"})) + + +class Denials(DikteTest): + def test_nothing_was_denied(self): + self.assertEqual(assistant._denial_warning({}), "") + self.assertEqual(assistant._denial_warning({"permission_denials": []}), "") + + def test_a_denied_tool_is_named(self): + warning = assistant._denial_warning( + {"permission_denials": [{"tool_name": "Bash"}]}) + self.assertIn("Bash", warning) + + def test_the_same_tool_denied_twice_is_named_once(self): + warning = assistant._denial_warning({"permission_denials": [ + {"tool_name": "Bash"}, {"tool_name": "Bash"}, {"tool_name": "Write"}]}) + self.assertEqual(warning.count("Bash"), 1) + self.assertIn("Write", warning) + + +class SessionMissing(unittest.TestCase): + def test_a_session_that_is_gone(self): + for text in ("Error: session abc not found", + "No conversation with that id", + "unknown thread: abc"): + with self.subTest(text=text): + self.assertTrue(assistant._session_missing(text)) + + def test_an_unrelated_failure(self): + for text in ("", "network unreachable", "session limit exceeded"): + with self.subTest(text=text): + 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"), + "real error") + self.assertEqual(assistant._last_line(""), "") + self.assertEqual(assistant._last_line(None), "") + + +class Conclude(DikteTest): + def found(self, **changes): + row = {"answer": "", "warning": "", "session": "", "failure": ""} + row.update(changes) + return row + + def test_an_answer_and_its_session(self): + answer, warning = assistant._conclude( + self.found(answer="done", session="abc"), 0, "", "", "Claude") + self.assertEqual(answer, "done") + self.assertEqual(warning, "") + self.assertEqual(assistant.read_session("claude", 1800), "abc") + + def test_codex_stores_under_its_own_name(self): + assistant._conclude(self.found(answer="done", session="t-1"), 0, "", + "", "Codex") + self.assertEqual(assistant.read_session("codex", 1800), "t-1") + + def test_a_non_zero_exit_with_nothing_to_show_for_it(self): + with self.assertRaises(assistant.AssistantError) as caught: + assistant._conclude(self.found(), 1, "it all went wrong\n", "", "Claude") + self.assertIn("it all went wrong", str(caught.exception)) + + def test_a_session_that_is_gone_is_raised_apart(self): + with self.assertRaises(assistant._SessionGone): + assistant._conclude(self.found(), 1, "session abc not found", + "abc", "Claude") + + def test_a_session_that_is_gone_only_matters_when_one_was_resumed(self): + with self.assertRaises(assistant.AssistantError): + assistant._conclude(self.found(), 1, "session abc not found", + "", "Claude") + + def test_an_answer_survives_a_non_zero_exit(self): + answer, _ = assistant._conclude(self.found(answer="done"), 1, "noise", + "", "Claude") + self.assertEqual(answer, "done") + + def test_a_reported_failure_with_no_answer(self): + with self.assertRaises(assistant.AssistantError) as caught: + assistant._conclude(self.found(failure="the model refused"), 0, "", + "", "Claude") + self.assertIn("refused", str(caught.exception)) + + def test_a_run_that_said_nothing_at_all(self): + with self.assertRaises(assistant.AssistantError) as caught: + assistant._conclude(self.found(), 0, "", "", "Codex") + self.assertIn("Codex", str(caught.exception)) + + +class AskClaude(DikteTest): + def run_ask(self, conf=None, events=None, code=0, stderr="", noise=(), + session=""): + conf = conf or self.config() + proc = FakeCli(events or [ + {"type": "system", "subtype": "init", "session_id": "abc"}, + {"type": "result", "session_id": "abc", "result": " done "}, + ], code=code, stderr=stderr, noise=noise) + stages = [] + with only_these_tools("claude", "codex"), \ + mock.patch.object(subprocess, "Popen", return_value=proc) as popen: + result = assistant._ask_claude( + "book it", conf, session, stages.append, None) + return result, popen.call_args.args[0], stages + + def test_the_answer_comes_back_stripped(self): + (answer, warning), _, _ = self.run_ask() + self.assertEqual(answer, "done") + self.assertEqual(warning, "") + + def test_the_prompt_goes_in_as_one_argument(self): + _, cmd, _ = self.run_ask() + self.assertEqual(cmd[:3], ["claude", "-p", "book it"]) + + def test_the_stream_is_asked_for_so_progress_can_be_shown(self): + _, cmd, _ = self.run_ask() + self.assertIn("--output-format", cmd) + self.assertIn("stream-json", cmd) + self.assertIn("--verbose", cmd) + + def test_the_model_and_the_permission_mode_are_passed_on(self): + conf = self.config(assistant_model="opus", + assistant_permission_mode="plan") + _, cmd, _ = self.run_ask(conf) + self.assertEqual(cmd[cmd.index("--model") + 1], "opus") + self.assertEqual(cmd[cmd.index("--permission-mode") + 1], "plan") + + def test_the_instruction_rides_along_as_a_system_prompt(self): + conf = self.config() + _, cmd, _ = self.run_ask(conf) + self.assertEqual(cmd[cmd.index("--append-system-prompt") + 1], + conf.assistant_prompt()) + + def test_no_effort_asked_for_means_no_flag(self): + _, cmd, _ = self.run_ask() + self.assertNotIn("--effort", cmd) + + def test_an_effort_is_translated_to_the_provider_s_vocabulary(self): + _, cmd, _ = self.run_ask(self.config(assistant_reasoning="minimal")) + self.assertEqual(cmd[cmd.index("--effort") + 1], "low") + + def test_a_conversation_is_resumed(self): + _, cmd, _ = self.run_ask(session="abc-123") + self.assertEqual(cmd[cmd.index("--resume") + 1], "abc-123") + + def test_a_fresh_conversation_resumes_nothing(self): + _, cmd, _ = self.run_ask() + self.assertNotIn("--resume", cmd) + + def test_every_tool_it_picks_up_is_named_in_the_corner(self): + _, _, stages = self.run_ask(events=[ + {"type": "assistant", "message": {"content": [ + {"type": "tool_use", "name": "WebSearch"}]}}, + {"type": "assistant", "message": {"content": [ + {"type": "tool_use", "name": "Bash"}]}}, + {"type": "result", "result": "done"}, + ]) + self.assertEqual(stages, ["Searching the web…", "Running a command…"]) + + def test_a_denied_tool_comes_back_as_a_warning_beside_the_answer(self): + (answer, warning), _, _ = self.run_ask(events=[ + {"type": "result", "result": "I could not do that.", + "permission_denials": [{"tool_name": "Bash"}]}, + ]) + self.assertEqual(answer, "I could not do that.") + self.assertIn("Bash", warning) + + def test_a_run_that_ended_in_an_error(self): + with self.assertRaises(assistant.AssistantError): + self.run_ask(events=[{"type": "result", "is_error": True, + "result": "rate limited"}]) + + def test_the_odd_unstructured_line_among_the_json(self): + (answer, _), _, _ = self.run_ask(noise=["Loading…", "not json at all"]) + self.assertEqual(answer, "done") + + def test_a_json_line_that_is_not_an_object(self): + proc = FakeCli(code=0) + proc.stdout = io.StringIO('{"type": "result", "result": "done"}\n[1,2]\n') + with only_these_tools("claude"), \ + mock.patch.object(subprocess, "Popen", return_value=proc): + answer, _ = assistant._ask_claude("hi", self.config(), "", None, None) + self.assertEqual(answer, "done") + + +class AskCodex(DikteTest): + def run_ask(self, conf=None, events=None, session=""): + conf = conf or self.config(assistant_provider="codex") + proc = FakeCli(events or [ + {"type": "thread.started", "thread_id": "t-1"}, + {"type": "item.completed", + "item": {"type": "agent_message", "text": "done"}}, + ]) + stages = [] + with only_these_tools("codex"), \ + mock.patch.object(subprocess, "Popen", return_value=proc) as popen: + result = assistant._ask_codex("book it", conf, session, + stages.append, None) + return result, popen.call_args.args[0], stages + + def test_the_answer(self): + (answer, _), _, _ = self.run_ask() + self.assertEqual(answer, "done") + + def test_the_instruction_is_kept_apart_from_the_command(self): + """Codex takes no system prompt, so the two must not read as one.""" + conf = self.config(assistant_provider="codex") + _, cmd, _ = self.run_ask(conf) + body = cmd[-1] + self.assertTrue(body.startswith(conf.assistant_prompt())) + self.assertIn("\n\n---\n\n", body) + self.assertTrue(body.endswith("book it")) + + def test_there_is_nobody_here_to_approve_anything(self): + _, cmd, _ = self.run_ask() + self.assertIn('approval_policy="never"', cmd) + self.assertIn("--skip-git-repo-check", cmd) + self.assertIn("--json", cmd) + + def test_the_sandbox_setting_is_passed_on(self): + _, cmd, _ = self.run_ask( + self.config(assistant_provider="codex", + assistant_codex_sandbox="read-only")) + self.assertIn('sandbox_mode="read-only"', cmd) + + def test_no_model_named_means_whatever_codex_is_set_to(self): + _, cmd, _ = self.run_ask() + self.assertNotIn("-m", cmd) + + def test_a_model_of_your_own(self): + _, cmd, _ = self.run_ask( + self.config(assistant_provider="codex", assistant_codex_model=" gpt-5 ")) + self.assertEqual(cmd[cmd.index("-m") + 1], "gpt-5") + + def test_the_effort_lands_on_the_nearest_rung_codex_has(self): + _, cmd, _ = self.run_ask( + self.config(assistant_provider="codex", assistant_reasoning="max")) + self.assertIn('model_reasoning_effort="high"', cmd) + + def test_a_conversation_is_resumed(self): + _, cmd, _ = self.run_ask(session="t-1") + self.assertEqual(cmd[:4], ["codex", "exec", "resume", "t-1"]) + + def test_a_fresh_conversation(self): + _, cmd, _ = self.run_ask() + self.assertEqual(cmd[:2], ["codex", "exec"]) + + def test_the_closing_message_is_the_answer(self): + (answer, _), _, _ = self.run_ask(events=[ + {"type": "item.completed", + "item": {"type": "agent_message", "text": "let me look"}}, + {"type": "item.completed", + "item": {"type": "agent_message", "text": "it is on Thursday"}}, + ]) + self.assertEqual(answer, "it is on Thursday") + + def test_the_work_is_narrated_as_it_goes(self): + _, _, stages = self.run_ask(events=[ + {"type": "item.started", "item": {"type": "command_execution"}}, + {"type": "item.completed", + "item": {"type": "agent_message", "text": "done"}}, + ]) + self.assertEqual(stages, ["Running a command…"]) + + def test_a_turn_that_failed(self): + with self.assertRaises(assistant.AssistantError) as caught: + self.run_ask(events=[{"type": "turn.failed", + "error": {"message": "quota exhausted"}}]) + self.assertIn("quota", str(caught.exception)) + + +class AskOpenRouter(DikteTest): + def test_a_question_and_an_answer(self): + conf = self.config(assistant_provider="openrouter", + openrouter_api_key="sk-or-test") + with fake_urlopen({"choices": [{"message": {"content": "on Thursday"}}]}): + answer, warning = assistant.ask("when is it", conf) + self.assertEqual(answer, "on Thursday") + self.assertEqual(warning, "") + + def test_the_conversation_is_ours_to_keep(self): + conf = self.config(assistant_provider="openrouter", + openrouter_api_key="sk-or-test") + with fake_urlopen({"choices": [{"message": {"content": "on Thursday"}}]}): + assistant.ask("when is it", conf) + stored = assistant.read_messages("openrouter", 1800) + self.assertEqual([row["content"] for row in stored], + ["when is it", "on Thursday"]) + + def test_the_next_command_knows_what_that_means(self): + conf = self.config(assistant_provider="openrouter", + openrouter_api_key="sk-or-test") + assistant.write_session("openrouter", messages=[ + {"role": "user", "content": "when is it"}, + {"role": "assistant", "content": "on Thursday"}]) + with fake_urlopen({"choices": [{"message": {"content": "moved"}}]}) as calls: + assistant.ask("move it to Friday", conf) + sent = json.loads(calls[0].data.decode("utf-8"))["messages"] + self.assertEqual(len(sent), 4) # system, the two stored, the new one + + def test_an_api_failure_reads_as_an_assistant_failure(self): + conf = self.config(assistant_provider="openrouter") + with self.assertRaises(assistant.AssistantError): + assistant.ask("when is it", conf) + + +class Ask(DikteTest): + def test_a_cli_that_is_not_installed_says_where_to_change_it(self): + with only_these_tools(), \ + self.assertRaises(assistant.AssistantError) as caught: + assistant.ask("hi", self.config()) + self.assertIn("claude", str(caught.exception)) + self.assertIn("Settings", str(caught.exception)) + + def test_a_session_that_is_gone_is_started_over_without_a_word(self): + conf = self.config() + assistant.write_session("claude", "stale-id") + attempts = [] + + def run(prompt, conf, session, on_stage, should_stop): + attempts.append(session) + if session: + raise assistant._SessionGone() + return "done", "" + + with only_these_tools("claude"), \ + mock.patch.object(assistant, "_ask_claude", side_effect=run): + answer, _ = assistant.ask("hi", conf) + self.assertEqual(answer, "done") + self.assertEqual(attempts, ["stale-id", ""]) + self.assertEqual(assistant.stored_provider(), "") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_audio.py b/tests/test_audio.py new file mode 100644 index 0000000..9a8dbaa --- /dev/null +++ b/tests/test_audio.py @@ -0,0 +1,310 @@ +"""Level metering, the WAV writer, and what pactl is asked for. + +The device list is where a platform port lands first, so the parsing is pinned +here: a source that is not a monitor is an input, one that is belongs to the +speakers, and neither list may go missing when pactl is absent. +""" + +import array +import contextlib +import json +import os +import subprocess +import unittest +import wave +from unittest import mock + +import audio +from tests.support import ( + DikteTest, + FakeCompleted, + linux_only, + only_these_tools, + pcm, + silence, + stereo, + tone, +) + + +class ChunkLevels(unittest.TestCase): + def test_silence(self): + self.assertEqual(audio.chunk_levels(silence(0.1)), (0.0, 0.0)) + + def test_nothing_at_all(self): + self.assertEqual(audio.chunk_levels(b""), (0.0, 0.0)) + + def test_half_a_sample_is_not_a_sample(self): + self.assertEqual(audio.chunk_levels(b"\x00"), (0.0, 0.0)) + + def test_an_odd_trailing_byte_is_ignored_rather_than_fatal(self): + peak, _ = audio.chunk_levels(pcm([16384, 16384]) + b"\x7f") + self.assertAlmostEqual(peak, 0.5, places=3) + + def test_the_peak_is_the_loudest_sample_either_way(self): + peak, _ = audio.chunk_levels(pcm([0, 0, -32768, 100])) + self.assertEqual(peak, 1.0) + + def test_the_rms_of_a_constant_signal_is_that_constant(self): + _, rms = audio.chunk_levels(pcm([16384] * 100)) + self.assertAlmostEqual(rms, 0.5, places=3) + + def test_the_rms_sits_below_the_peak_for_a_tone(self): + peak, rms = audio.chunk_levels(tone(0.1, amplitude=16384)) + self.assertLess(rms, peak) + self.assertGreater(rms, 0.0) + + def test_neither_number_ever_passes_one(self): + peak, rms = audio.chunk_levels(pcm([-32768] * 100)) + self.assertEqual(peak, 1.0) + self.assertEqual(rms, 1.0) + + +class StereoLevels(unittest.TestCase): + def test_the_channels_are_read_apart(self): + left, right = audio.stereo_levels(stereo(pcm([16384] * 50), + pcm([0] * 50))) + self.assertAlmostEqual(left, 0.5, places=3) + self.assertEqual(right, 0.0) + + def test_nothing_at_all(self): + self.assertEqual(audio.stereo_levels(b""), (0.0, 0.0)) + + def test_a_partial_frame_is_ignored(self): + self.assertEqual(audio.stereo_levels(b"\x00\x01\x00"), (0.0, 0.0)) + + def test_a_meeting_with_both_sides_talking(self): + left, right = audio.stereo_levels(stereo(pcm([8192] * 50), + pcm([-16384] * 50))) + self.assertAlmostEqual(left, 0.25, places=3) + self.assertAlmostEqual(right, 0.5, places=3) + + +class WriteWav(DikteTest): + def test_the_header_says_what_the_recorder_captured(self): + path = audio.write_wav(silence(0.5)) + self.addCleanup(os.unlink, path) + with contextlib.closing(wave.open(path, "rb")) as wav: + self.assertEqual(wav.getnchannels(), audio.CHANNELS) + self.assertEqual(wav.getsampwidth(), audio.SAMPLE_WIDTH) + self.assertEqual(wav.getframerate(), audio.RATE) + self.assertEqual(wav.getnframes(), int(audio.RATE * 0.5)) + + def test_the_samples_survive(self): + path = audio.write_wav(pcm([1000, -1000, 2000])) + self.addCleanup(os.unlink, path) + with contextlib.closing(wave.open(path, "rb")) as wav: + samples = array.array("h") + samples.frombytes(wav.readframes(3)) + self.assertEqual(list(samples), [1000, -1000, 2000]) + + def test_a_meeting_is_written_at_two_channels(self): + path = audio.write_wav(stereo(silence(0.1), silence(0.1)), channels=2) + self.addCleanup(os.unlink, path) + with contextlib.closing(wave.open(path, "rb")) as wav: + self.assertEqual(wav.getnchannels(), 2) + + +SOURCES = [ + {"name": "alsa_input.pci-0000_00_1f.3.analog-stereo", + "description": "Built-in Audio Analog Stereo"}, + {"name": "alsa_output.pci-0000_00_1f.3.analog-stereo.monitor", + "description": "Monitor of Built-in Audio"}, + {"name": "bluez_input.AA_BB.headset", "description": ""}, +] + + +@linux_only +class Devices(DikteTest): + @contextlib.contextmanager + def pactl(self, sources=None, sink=None, tools=("pactl",)): + payloads = { + "list": FakeCompleted(stdout=json.dumps( + SOURCES if sources is None else sources)), + "get-default-sink": FakeCompleted(stdout=(sink or "") + "\n"), + } + + def run(cmd, **kwargs): + return payloads["get-default-sink" if "get-default-sink" in cmd + else "list"] + + with only_these_tools(*tools), \ + mock.patch.object(subprocess, "run", side_effect=run): + yield + + def test_no_pactl_installed(self): + with only_these_tools(): + self.assertEqual(audio.list_sources(), []) + self.assertEqual(audio.list_monitors(), []) + self.assertEqual(audio.default_monitor(), "") + + def test_inputs_leave_the_monitors_out(self): + with self.pactl(): + names = [name for name, _ in audio.list_sources()] + self.assertEqual(names, [SOURCES[0]["name"], SOURCES[2]["name"]]) + + def test_monitors_are_the_other_half(self): + with self.pactl(): + self.assertEqual([name for name, _ in audio.list_monitors()], + [SOURCES[1]["name"]]) + + def test_a_device_with_no_description_is_shown_by_its_name(self): + with self.pactl(): + sources = dict(audio.list_sources()) + self.assertEqual(sources[SOURCES[2]["name"]], SOURCES[2]["name"]) + + def test_pactl_output_that_is_not_json(self): + with only_these_tools("pactl"), \ + mock.patch.object(subprocess, "run", + return_value=FakeCompleted(stdout="not json")): + self.assertEqual(audio.list_sources(), []) + + def test_pactl_that_will_not_run(self): + with only_these_tools("pactl"), \ + mock.patch.object(subprocess, "run", side_effect=OSError("nope")): + self.assertEqual(audio.list_sources(), []) + + def test_pactl_that_exits_non_zero(self): + with only_these_tools("pactl"), \ + mock.patch.object(subprocess, "run", + side_effect=subprocess.CalledProcessError(1, "pactl")): + self.assertEqual(audio.list_sources(), []) + + def test_the_default_output_is_found_by_its_monitor(self): + with self.pactl(sink="alsa_output.pci-0000_00_1f.3.analog-stereo"): + self.assertEqual(audio.default_monitor(), + SOURCES[1]["name"]) + + def test_a_default_sink_with_no_monitor_of_its_own(self): + with self.pactl(sink="alsa_output.usb-something"): + self.assertEqual(audio.default_monitor(), "") + + def test_no_default_sink_at_all(self): + with self.pactl(sink=""): + self.assertEqual(audio.default_monitor(), "") + + def test_a_monitor_is_trusted_when_the_list_is_empty(self): + """pactl answered about the sink but not about the sources.""" + with self.pactl(sources=[], sink="alsa_output.usb-something"): + self.assertEqual(audio.default_monitor(), + "alsa_output.usb-something.monitor") + + +class FakeProcess: + """A pw-record that hands over a fixed buffer and then ends.""" + + def __init__(self, data): + import io + self.stdout = io.BytesIO(data) + self.stderr = io.BytesIO(b"") + self.signals = [] + self._alive = True + + def poll(self): + return None if self._alive else 0 + + def send_signal(self, sig): + self.signals.append(sig) + self._alive = False + + def wait(self, timeout=None): + self._alive = False + return 0 + + def kill(self): + self._alive = False + + +@linux_only +class RecorderChain(DikteTest): + """Start to WAV, with pw-record faked out.""" + + def record(self, data, target="", max_seconds=300): + recorder = audio.Recorder() + results = [] + failures = [] + recorder.stopped.connect(lambda *args: results.append(args)) + recorder.failed.connect(failures.append) + proc = FakeProcess(data) + with only_these_tools("pw-record"), \ + mock.patch.object(subprocess, "Popen", return_value=proc) as popen: + recorder.start(target=target, max_seconds=max_seconds) + recorder._thread.join(timeout=5) + recorder.stop() + return recorder, results, failures, popen + + def test_pw_record_is_not_installed(self): + recorder = audio.Recorder() + failures = [] + recorder.failed.connect(failures.append) + with only_these_tools(): + recorder.start() + self.assertEqual(len(failures), 1) + self.assertIn("pipewire", failures[0]) + + def test_the_capture_format_is_what_the_rest_of_the_code_expects(self): + _, _, _, popen = self.record(silence(1.0)) + cmd = popen.call_args.args[0] + self.assertEqual(cmd[0], "pw-record") + self.assertIn(f"--rate={audio.RATE}", cmd) + self.assertIn(f"--channels={audio.CHANNELS}", cmd) + self.assertIn("--format=s16", cmd) + self.assertEqual(cmd[-1], "-") + + def test_no_target_means_no_target_flag(self): + _, _, _, popen = self.record(silence(0.5)) + self.assertFalse([arg for arg in popen.call_args.args[0] + if arg.startswith("--target=")]) + + def test_a_chosen_microphone_is_passed_on(self): + _, _, _, popen = self.record(silence(0.5), target="alsa_input.usb") + self.assertIn("--target=alsa_input.usb", popen.call_args.args[0]) + + def test_a_recording_ends_as_a_wav_with_its_duration_and_levels(self): + _, results, failures, _ = self.record(tone(1.0)) + self.assertEqual(failures, []) + path, duration, rms = results[0] + self.addCleanup(os.unlink, path) + self.assertAlmostEqual(duration, 1.0, places=2) + self.assertTrue(rms) + self.assertGreater(max(rms), 0.0) + with contextlib.closing(wave.open(path, "rb")) as wav: + self.assertEqual(wav.getnframes(), audio.RATE) + + def test_a_stray_keypress_is_not_a_recording(self): + _, results, failures, _ = self.record(silence(0.1)) + self.assertEqual(results, []) + self.assertIn("0.3", failures[0]) + + def test_a_cancelled_recording_produces_nothing(self): + recorder = audio.Recorder() + results = [] + recorder.stopped.connect(lambda *args: results.append(args)) + proc = FakeProcess(tone(1.0)) + with only_these_tools("pw-record"), \ + mock.patch.object(subprocess, "Popen", return_value=proc): + recorder.start() + recorder._thread.join(timeout=5) + recorder.cancel() + recorder.stop() + self.assertEqual(results, []) + + def test_a_recording_that_runs_past_the_limit_is_cut_off(self): + _, results, _, _ = self.record(tone(3.0), max_seconds=1) + path, duration, _ = results[0] + self.addCleanup(os.unlink, path) + self.assertLessEqual(duration, 1.1) + + def test_a_recorder_that_could_not_start(self): + recorder = audio.Recorder() + failures = [] + recorder.failed.connect(failures.append) + with only_these_tools("pw-record"), \ + mock.patch.object(subprocess, "Popen", side_effect=OSError("nope")): + recorder.start() + self.assertEqual(len(failures), 1) + self.assertFalse(recorder.active) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..98145c3 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,490 @@ +"""The terminal interface, which is the part a script depends on. + +Output is a contract as much as an interface: --json prints one object on +stdout, progress goes to stderr so it never lands in a pipe, and the exit code +says which of the four things happened. Nothing here starts an instance; the +socket is faked, and everything that runs locally runs for real. +""" + +import contextlib +import io +import json +import unittest +from unittest import mock + +import cli +import config as cfg +import ipc +from tests.support import DikteTest + + +class Options: + """The parsed command line, as much of it as the printers read.""" + + def __init__(self, **values): + self.json = False + self.quiet = False + for key, value in values.items(): + setattr(self, key, value) + + +@contextlib.contextmanager +def captured(): + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + yield out, err + + +class Printing(unittest.TestCase): + def test_plain_output_is_the_thing_a_person_wanted(self): + with captured() as (out, err): + code = cli.out(Options(), {"ok": True, "text": "hello"}, "hello") + self.assertEqual(code, 0) + self.assertEqual(out.getvalue().strip(), "hello") + self.assertEqual(err.getvalue(), "") + + def test_json_output_is_one_object(self): + with captured() as (out, _): + cli.out(Options(json=True), {"ok": True, "text": "hello"}, "hello") + self.assertEqual(json.loads(out.getvalue()), {"ok": True, "text": "hello"}) + + def test_nothing_to_say_prints_nothing(self): + with captured() as (out, _): + cli.out(Options(), {"ok": True}) + self.assertEqual(out.getvalue(), "") + + def test_progress_never_reaches_stdout(self): + with captured() as (out, err): + cli.note(Options(), "Transcribing…") + self.assertEqual(out.getvalue(), "") + self.assertIn("Transcribing…", err.getvalue()) + + def test_quiet_keeps_progress_off_stderr_too(self): + with captured() as (_, err): + cli.note(Options(quiet=True), "Transcribing…") + self.assertEqual(err.getvalue(), "") + + def test_a_failure_goes_to_stderr_and_returns_one(self): + with captured() as (out, err): + code = cli.fail(Options(), "no microphone") + self.assertEqual(code, 1) + self.assertEqual(out.getvalue(), "") + self.assertIn("no microphone", err.getvalue()) + + def test_a_failure_as_json_stays_on_stdout(self): + with captured() as (out, err): + code = cli.fail(Options(json=True), "no microphone", 3, running=False) + self.assertEqual(code, 3) + self.assertEqual(json.loads(out.getvalue()), + {"ok": False, "error": "no microphone", "running": False}) + self.assertEqual(err.getvalue(), "") + + +class Coerce(unittest.TestCase): + """A value off the command line, in the type the setting is stored as.""" + + def test_a_string_stays_a_string(self): + self.assertEqual(cli._coerce("cleanup_model", "some/model"), "some/model") + + def test_the_words_that_mean_true(self): + for raw in ("1", "true", "TRUE", "yes", "on", " True "): + with self.subTest(raw=raw): + self.assertIs(cli._coerce("cleanup_enabled", raw), True) + + def test_the_words_that_mean_false(self): + for raw in ("0", "false", "no", "off"): + with self.subTest(raw=raw): + self.assertIs(cli._coerce("cleanup_enabled", raw), False) + + def test_anything_else_is_not_a_boolean(self): + with self.assertRaises(ValueError): + cli._coerce("cleanup_enabled", "maybe") + + def test_a_whole_number(self): + self.assertEqual(cli._coerce("history_limit", "50"), 50) + self.assertEqual(cli._coerce("history_limit", "50.9"), 50) + + def test_a_number_with_a_fraction(self): + self.assertEqual(cli._coerce("silence_db", "-42.5"), -42.5) + + def test_something_that_is_not_a_number(self): + with self.assertRaises(ValueError): + cli._coerce("history_limit", "lots") + + def test_a_boolean_is_settled_before_it_is_read_as_a_number(self): + """bool is a subclass of int, so the order of the checks matters.""" + self.assertIs(cli._coerce("cleanup_enabled", "1"), True) + + +class Masking(unittest.TestCase): + def test_a_key_is_shown_by_its_last_four(self): + self.assertEqual(cli._mask("openai_api_key", "sk-abcdefgh1234"), "…1234") + + def test_an_empty_key_is_not_masked_into_something(self): + self.assertEqual(cli._mask("openai_api_key", ""), "") + + def test_anything_that_is_not_a_key_is_shown(self): + self.assertEqual(cli._mask("cleanup_model", "some/model"), "some/model") + + def test_both_keys_are_covered(self): + for key in cli.SECRET_KEYS: + with self.subTest(key=key): + self.assertTrue(cli._mask(key, "sk-secret").startswith("…")) + + +class Parser(unittest.TestCase): + """Every verb has to parse, and keep the flag that was typed before it.""" + + def parse(self, *argv): + return cli.build_parser().parse_args(list(argv)) + + def test_no_verb_at_all_is_the_settings_window(self): + # argparse leaves the dest as None; run() is what turns it into "". + opts = self.parse() + self.assertIsNone(opts.verb) + self.assertEqual(opts.func, cli.cmd_plain) + + def test_every_verb_is_wired_to_something(self): + for verb in ("record", "toggle", "start", "stop", "cancel", "ask", + "session", "transcribe", "meeting", "meetings", "history", + "config", "prompt", "devices", "models", "test-key", + "doctor", "shortcut", "status", "settings", "restart", + "quit", "help"): + with self.subTest(verb=verb): + argv = [verb] + if verb == "transcribe": + argv.append("clip.mp3") + opts = self.parse(*argv) + self.assertEqual(opts.verb, verb) + self.assertTrue(callable(opts.func)) + + def test_the_old_spellings_still_parse(self): + for verb in ("ask-cancel", "ask-reset", "meeting-cancel"): + with self.subTest(verb=verb): + self.assertEqual(self.parse(verb).verb, verb) + + def test_a_flag_typed_before_the_verb_survives(self): + self.assertTrue(self.parse("--json", "status").json) + + def test_a_flag_typed_after_the_verb_works_too(self): + self.assertTrue(self.parse("status", "--json").json) + + def test_a_flag_before_the_verb_is_not_overwritten_by_the_default(self): + opts = self.parse("--quiet", "record") + self.assertTrue(opts.quiet) + + def test_a_command_to_the_agent_is_taken_as_written(self): + opts = self.parse("ask", "book", "it", "for", "Thursday") + self.assertEqual(opts.text, ["book", "it", "for", "Thursday"]) + + def test_a_command_with_no_text_is_a_recording(self): + self.assertEqual(self.parse("ask").text, []) + + def test_the_subcommands_of_a_group(self): + self.assertEqual(self.parse("config", "get", "cleanup_model").key, + "cleanup_model") + self.assertEqual(self.parse("history", "list", "--limit", "5").limit, 5) + self.assertEqual(self.parse("meetings", "show", "3").which, "3") + + def test_a_group_with_no_subcommand_asks_for_one(self): + with captured(): + self.assertEqual(self.parse("config").func(Options()), 2) + + def test_the_three_way_flags_start_out_undecided(self): + """--cleanup and --no-cleanup both given as nothing means the setting.""" + opts = self.parse("transcribe", "clip.mp3") + self.assertIsNone(opts.cleanup) + self.assertIsNone(opts.timestamps) + self.assertFalse(self.parse("transcribe", "clip.mp3", "--no-cleanup").cleanup) + self.assertTrue(self.parse("transcribe", "clip.mp3", "--cleanup").cleanup) + + def test_a_setting_that_is_not_a_choice_is_refused(self): + with self.assertRaises(SystemExit), captured(): + self.parse("models", "--provider", "ollama") + + def test_a_flag_that_falls_back_to_the_setting(self): + self.assertIs(cli._pick(None, True), True) + self.assertIs(cli._pick(False, True), False) + + +class ConfigCommands(DikteTest): + def run_cmd(self, func, **values): + with captured() as (out, err): + code = func(Options(**values)) + return code, out.getvalue(), err.getvalue() + + def test_reading_a_setting(self): + code, out, _ = self.run_cmd(cli.cmd_config_get, key="cleanup_model") + self.assertEqual(code, 0) + self.assertEqual(out.strip(), cfg.DEFAULTS["cleanup_model"]) + + def test_reading_a_setting_that_is_not_a_string(self): + _, out, _ = self.run_cmd(cli.cmd_config_get, key="history_limit") + self.assertEqual(out.strip(), str(cfg.DEFAULTS["history_limit"])) + + def test_a_setting_nobody_has(self): + code, _, err = self.run_cmd(cli.cmd_config_get, key="no_such_setting") + self.assertEqual(code, 2) + self.assertIn("unknown setting", err) + + def test_writing_a_setting_reaches_the_file(self): + with mock.patch.object(ipc, "send"): + code, _, _ = self.run_cmd(cli.cmd_config_set, key="cleanup_model", + value="some/model") + self.assertEqual(code, 0) + self.assertEqual(cfg.Config()["cleanup_model"], "some/model") + + def test_a_running_instance_is_told_to_read_it_back(self): + """It would otherwise write its own copy back over the change.""" + with mock.patch.object(ipc, "send") as send: + self.run_cmd(cli.cmd_config_set, key="cleanup_model", value="some/model") + send.assert_called_once_with("reload") + + def test_writing_a_boolean(self): + with mock.patch.object(ipc, "send"): + self.run_cmd(cli.cmd_config_set, key="cleanup_enabled", value="off") + self.assertIs(cfg.Config()["cleanup_enabled"], False) + + def test_a_value_of_the_wrong_type(self): + with mock.patch.object(ipc, "send"): + code, _, err = self.run_cmd(cli.cmd_config_set, + key="history_limit", value="lots") + self.assertEqual(code, 2) + # A number says only what could not be converted; a boolean names the + # setting as well, because "true or false" needs the context. + self.assertIn("lots", err) + + def test_a_key_is_masked_when_it_is_written_back(self): + with mock.patch.object(ipc, "send"): + _, out, _ = self.run_cmd(cli.cmd_config_set, key="openai_api_key", + value="sk-abcdefgh1234") + self.assertNotIn("sk-abcdefgh", out) + self.assertIn("1234", out) + + def test_the_listing_masks_the_keys(self): + with mock.patch.object(ipc, "send"): + self.run_cmd(cli.cmd_config_set, key="openai_api_key", + value="sk-abcdefgh1234") + _, out, _ = self.run_cmd(cli.cmd_config_list, reveal=False) + self.assertNotIn("sk-abcdefgh", out) + + def test_a_key_belongs_to_whoever_asked_for_it_by_name(self): + with mock.patch.object(ipc, "send"): + self.run_cmd(cli.cmd_config_set, key="openai_api_key", + value="sk-abcdefgh1234") + _, out, _ = self.run_cmd(cli.cmd_config_list, reveal=True) + self.assertIn("sk-abcdefgh1234", out) + + def test_the_listing_shortens_a_long_value(self): + with mock.patch.object(ipc, "send"): + self.run_cmd(cli.cmd_config_set, key="cleanup_prompt", value="x" * 200) + _, out, _ = self.run_cmd(cli.cmd_config_list, reveal=False) + self.assertNotIn("x" * 100, out) + + def test_resetting_one_setting(self): + with mock.patch.object(ipc, "send"): + self.run_cmd(cli.cmd_config_set, key="cleanup_model", value="some/model") + code, _, _ = self.run_cmd(cli.cmd_config_reset, key=["cleanup_model"], + all=False) + self.assertEqual(code, 0) + self.assertEqual(cfg.Config()["cleanup_model"], cfg.DEFAULTS["cleanup_model"]) + + def test_resetting_nothing_asks_what_to_reset(self): + code, _, err = self.run_cmd(cli.cmd_config_reset, key=[], all=False) + self.assertEqual(code, 2) + self.assertIn("--all", err) + + def test_resetting_everything(self): + with mock.patch.object(ipc, "send"): + self.run_cmd(cli.cmd_config_set, key="cleanup_model", value="some/model") + self.run_cmd(cli.cmd_config_reset, key=[], all=True) + self.assertEqual(cfg.Config()["cleanup_model"], cfg.DEFAULTS["cleanup_model"]) + + def test_where_things_are_stored(self): + _, out, _ = self.run_cmd(cli.cmd_config_path, json=True) + paths = json.loads(out) + self.assertEqual(paths["config"], str(cfg.CONFIG_FILE)) + self.assertEqual(paths["history"], str(cfg.HISTORY_FILE)) + + def test_the_prompt_a_run_would_really_send(self): + _, out, _ = self.run_cmd(cli.cmd_prompt, which="cleanup") + self.assertEqual(out.strip(), cfg.CLEANUP_PROMPT_EN.strip()) + + def test_all_four_prompts_at_once(self): + _, out, _ = self.run_cmd(cli.cmd_prompt, which=None, json=True) + self.assertEqual(set(json.loads(out)["prompts"]), + {"cleanup", "subtitles", "meeting", "agent"}) + + +class Finding(DikteTest): + def test_no_history_at_all(self): + self.assertIsNone(cli._find_history("last")) + + def test_the_newest_entry(self): + for text in ("first", "second"): + cfg.append_history({"ts": "now", "text": text}) + self.assertEqual(cli._find_history("last")["text"], "second") + self.assertEqual(cli._find_history("1")["text"], "second") + self.assertEqual(cli._find_history("2")["text"], "first") + + def test_counting_past_the_end(self): + cfg.append_history({"ts": "now", "text": "only one"}) + self.assertIsNone(cli._find_history("2")) + self.assertIsNone(cli._find_history("0")) + + def test_something_that_is_not_a_number(self): + cfg.append_history({"ts": "now", "text": "only one"}) + self.assertIsNone(cli._find_history("yesterday")) + + def test_a_meeting_by_its_stem(self): + for base in ("20260801-100000", "20260802-110000"): + cfg.save_meeting({"base": base, "status": "done"}) + self.assertEqual(cli._find_meeting("20260801-100000")["base"], + "20260801-100000") + + def test_a_meeting_by_the_start_of_its_stem(self): + for base in ("20260801-100000", "20260801-110000"): + cfg.save_meeting({"base": base, "status": "done"}) + self.assertEqual(cli._find_meeting("20260801-1")["base"], "20260801-110000") + + def test_a_meeting_by_the_date_it_was_recorded(self): + """A stem is all digits too, so a date must not be read as a position.""" + for base in ("20260801-100000", "20260801-140000", "20260802-110000"): + cfg.save_meeting({"base": base, "status": "done"}) + self.assertEqual(cli._find_meeting("20260801")["base"], "20260801-140000") + + def test_a_meeting_by_position(self): + for base in ("20260801-100000", "20260802-110000"): + cfg.save_meeting({"base": base, "status": "done"}) + self.assertEqual(cli._find_meeting("1")["base"], "20260802-110000") + self.assertEqual(cli._find_meeting("2")["base"], "20260801-100000") + self.assertEqual(cli._find_meeting("last")["base"], "20260802-110000") + + def test_a_position_wins_while_there_are_that_many_meetings(self): + """Counting back is what a small number has always meant, and a stem + never starts with one: it starts with the year.""" + for base in ("20260801-100000", "20260802-110000"): + cfg.save_meeting({"base": base, "status": "done"}) + self.assertEqual(cli._find_meeting("2")["base"], "20260801-100000") + + def test_counting_past_the_end_finds_nothing_rather_than_the_wrong_one(self): + cfg.save_meeting({"base": "20260801-100000", "status": "done"}) + self.assertIsNone(cli._find_meeting("9")) + self.assertIsNone(cli._find_meeting("0")) + + def test_a_meeting_nobody_recorded(self): + cfg.save_meeting({"base": "20260801-100000", "status": "done"}) + self.assertIsNone(cli._find_meeting("20261231")) + + +class WithoutAnInstance(DikteTest): + """Nothing is listening on the socket, which is three different things. + + A verb that can start the application does; one that asks for a state the + application is already in succeeds; anything else fails with code 3. + """ + + def run_verb(self, argv): + # launch_gui replaces this process with the application, so it never + # comes back in real use and must not be allowed to here. + with mock.patch.object(ipc, "send", return_value=None), \ + mock.patch.object(cli, "launch_gui") as launch, \ + captured() as (out, err): + code = cli.run(argv) + return code, out.getvalue(), err.getvalue(), launch + + def test_pressing_the_key_on_a_fresh_login_starts_it_recording(self): + """What the KDE shortcut has always relied on.""" + _, _, _, launch = self.run_verb(["toggle"]) + launch.assert_called_once_with("toggle") + + def test_every_verb_that_opens_a_window_can_start_it(self): + for verb in ("settings", "toggle", "ask", "meeting"): + with self.subTest(verb=verb): + self.assertTrue(self.run_verb([verb])[3].called) + + def test_a_verb_asked_to_wait_starts_nothing(self): + """There would be no run to wait for; the process would just be replaced.""" + _, _, _, launch = self.run_verb(["toggle", "--wait"]) + launch.assert_not_called() + + def test_asking_it_to_stop_when_it_is_not_going_is_not_a_failure(self): + for verb in ("cancel", "quit", "restart", "ask-reset"): + with self.subTest(verb=verb): + code, _, _, _ = self.run_verb([verb]) + self.assertEqual(code, 0) + + def test_anything_else_says_nothing_is_running(self): + for argv in (["record"], ["start"]): + with self.subTest(argv=argv): + code, _, err, _ = self.run_verb(argv) + self.assertEqual(code, cli.NOT_RUNNING) + self.assertIn("not running", err) + + def test_status_answers_the_question_rather_than_failing_it(self): + """"Is it running" has an answer when it is not, and it goes to stdout.""" + code, out, _, _ = self.run_verb(["status"]) + self.assertEqual(code, cli.NOT_RUNNING) + self.assertIn("not running", out) + + def test_status_as_json_says_so_in_a_field(self): + _, out, _, _ = self.run_verb(["--json", "status"]) + self.assertFalse(json.loads(out)["running"]) + + def test_the_answer_says_so_in_json_too(self): + code, out, _, _ = self.run_verb(["--json", "record"]) + self.assertEqual(code, cli.NOT_RUNNING) + payload = json.loads(out) + self.assertFalse(payload["ok"]) + self.assertFalse(payload["running"]) + + def test_a_verb_that_needs_nothing_running_still_works(self): + with captured() as (out, _): + code = cli.run(["config", "get", "cleanup_model"]) + self.assertEqual(code, 0) + self.assertEqual(out.getvalue().strip(), cfg.DEFAULTS["cleanup_model"]) + + +class Replies(DikteTest): + """What the instance said, turned into output and an exit code.""" + + def run_verb(self, argv, reply): + with mock.patch.object(ipc, "send", return_value=reply), \ + captured() as (out, err): + code = cli.run(argv) + return code, out.getvalue(), err.getvalue() + + def test_a_dictation_prints_its_transcript(self): + code, out, _ = self.run_verb(["stop", "--wait"], + {"ok": True, "text": "Book it for Thursday."}) + self.assertEqual(code, 0) + self.assertEqual(out.strip(), "Book it for Thursday.") + + def test_a_dictation_that_failed(self): + code, out, err = self.run_verb(["stop", "--wait"], + {"ok": False, "error": "No speech detected"}) + self.assertEqual(code, 1) + self.assertEqual(out, "") + self.assertIn("No speech", err) + + def test_a_warning_goes_to_stderr_beside_the_answer(self): + code, out, err = self.run_verb( + ["stop", "--wait"], + {"ok": True, "text": "hello", "warning": "cleanup failed"}) + self.assertEqual(code, 0) + self.assertEqual(out.strip(), "hello") + self.assertIn("cleanup failed", err) + + def test_a_verb_with_nothing_to_say_prints_nothing(self): + code, out, _ = self.run_verb(["restart"], {"ok": True}) + self.assertEqual(code, 0) + self.assertEqual(out, "") + + def test_cancelling_something_that_is_not_running_is_not_a_failure(self): + with mock.patch.object(ipc, "send", return_value={"ok": True, "legacy": True}), \ + captured(): + self.assertEqual(cli.run(["cancel"]), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..7120a0c --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,427 @@ +"""Settings, the history file and the meeting index. + +Every one of these lives on disk and outlives an update, so the tests care most +about what happens to a file written by an older version: an unknown key, a +setting stored under its Turkish name, a prompt that used to be copied into the +config and now shadows the default. +""" + +import json +import os +import unittest +from unittest import mock + +import api +import config as cfg +import i18n +from tests.support import DikteTest + + +class Loading(DikteTest): + def test_nothing_stored_yet(self): + conf = cfg.Config() + self.assertEqual(conf["cleanup_model"], cfg.DEFAULTS["cleanup_model"]) + + def test_a_stored_value_wins(self): + self.write_config({"cleanup_model": "some/other-model"}) + self.assertEqual(cfg.Config()["cleanup_model"], "some/other-model") + + def test_a_key_this_version_does_not_have_is_dropped(self): + """A setting from a fork, or from a version that removed it.""" + self.write_config({"cleanup_model": "kept", "invented_by_a_fork": True}) + conf = cfg.Config() + self.assertEqual(conf["cleanup_model"], "kept") + self.assertNotIn("invented_by_a_fork", conf.data) + + def test_a_config_that_is_not_json_falls_back_to_the_defaults(self): + cfg.CONFIG_DIR.mkdir(parents=True, exist_ok=True) + cfg.CONFIG_FILE.write_text("{not json", encoding="utf-8") + with mock.patch("builtins.print"): + conf = cfg.Config() + self.assertEqual(conf["cleanup_model"], cfg.DEFAULTS["cleanup_model"]) + + def test_a_config_that_is_json_but_not_an_object(self): + cfg.CONFIG_DIR.mkdir(parents=True, exist_ok=True) + cfg.CONFIG_FILE.write_text("[1, 2]", encoding="utf-8") + self.assertEqual(cfg.Config()["cleanup_model"], cfg.DEFAULTS["cleanup_model"]) + + def test_a_corner_stored_under_its_old_turkish_name(self): + self.write_config({"overlay_corner": "sağ-üst"}) + self.assertEqual(cfg.Config()["overlay_corner"], "top-right") + + def test_a_corner_that_needs_no_migrating(self): + self.write_config({"overlay_corner": "bottom-right"}) + self.assertEqual(cfg.Config()["overlay_corner"], "bottom-right") + + def test_a_default_prompt_an_old_version_copied_in_is_dropped(self): + """Otherwise it shadows every later improvement to that default.""" + old = "the 1.2 default prompt, whatever it said" + with mock.patch.object(cfg, "LEGACY_PROMPTS", {cfg._fingerprint(old)}): + self.write_config({"cleanup_prompt": old}) + self.assertEqual(cfg.Config()["cleanup_prompt"], "") + + def test_a_prompt_the_user_wrote_is_left_alone(self): + self.write_config({"cleanup_prompt": "Only fix the punctuation."}) + self.assertEqual(cfg.Config()["cleanup_prompt"], "Only fix the punctuation.") + + def test_loading_sets_the_interface_language(self): + self.write_config({"ui_language": "tr"}) + cfg.Config() + self.assertEqual(i18n.language(), "tr") + + def test_an_unknown_key_reads_as_its_default(self): + self.assertIsNone(cfg.Config()["no_such_setting"]) + self.assertEqual(cfg.Config().get("no_such_setting", "fallback"), "fallback") + + +class Saving(DikteTest): + def test_a_saved_setting_comes_back(self): + conf = cfg.Config() + conf["cleanup_model"] = "some/model" + conf.save() + self.assertEqual(cfg.Config()["cleanup_model"], "some/model") + + def test_the_directory_is_created(self): + self.assertFalse(cfg.CONFIG_DIR.exists()) + cfg.Config().save() + self.assertTrue(cfg.CONFIG_FILE.exists()) + + def test_the_file_is_readable_by_nobody_else(self): + """It holds two API keys.""" + cfg.Config().save() + self.assertEqual(cfg.CONFIG_FILE.stat().st_mode & 0o777, 0o600) + + def test_nothing_is_left_behind_half_written(self): + cfg.Config().save() + self.assertEqual([p.name for p in cfg.CONFIG_DIR.iterdir()], ["config.json"]) + + def test_turkish_is_stored_as_turkish(self): + conf = cfg.Config() + conf["transcribe_prompt"] = "Paraşüt, öğle" + conf.save() + self.assertIn("Paraşüt", cfg.CONFIG_FILE.read_text(encoding="utf-8")) + + def test_saving_applies_the_interface_language(self): + conf = cfg.Config() + conf["ui_language"] = "tr" + conf.save() + self.assertEqual(i18n.language(), "tr") + + +class Keys(DikteTest): + def test_a_stored_key_is_used(self): + conf = self.config(openai_api_key=" sk-stored ") + self.assertEqual(conf.openai_key(), "sk-stored") + + def test_the_environment_is_the_fallback(self): + with mock.patch.dict(os.environ, {"OPENAI_API_KEY": "sk-env"}): + self.assertEqual(cfg.Config().openai_key(), "sk-env") + + def test_a_stored_key_beats_the_environment(self): + with mock.patch.dict(os.environ, {"OPENROUTER_API_KEY": "sk-env"}): + conf = self.config(openrouter_api_key="sk-stored") + self.assertEqual(conf.openrouter_key(), "sk-stored") + + def test_no_key_anywhere(self): + self.assertEqual(cfg.Config().openai_key(), "") + + +class TranscribeTarget(DikteTest): + def test_openai_by_default(self): + target = self.config(openai_api_key="sk-test").transcribe_target() + self.assertEqual(target.provider, "openai") + self.assertEqual(target.service, "OpenAI") + self.assertEqual(target.api_key, "sk-test") + self.assertEqual(target.base_url, api.OPENAI_URL) + self.assertEqual(target.model, cfg.DEFAULTS["transcribe_model"]) + + def test_openrouter_when_it_is_picked(self): + conf = self.config(transcribe_provider="openrouter", + openrouter_api_key="sk-or-test", + openrouter_transcribe_model="openai/whisper-1") + target = conf.transcribe_target() + self.assertEqual(target.provider, "openrouter") + self.assertEqual(target.service, "OpenRouter") + self.assertEqual(target.api_key, "sk-or-test") + self.assertEqual(target.model, "openai/whisper-1") + + def test_a_self_hosted_endpoint(self): + conf = self.config(openai_base_url="http://localhost:8080/v1") + self.assertEqual(conf.transcribe_target().base_url, "http://localhost:8080/v1") + + +class CleanupPrompt(DikteTest): + def test_the_default_follows_the_interface_language(self): + self.assertEqual(cfg.Config().cleanup_prompt(), cfg.CLEANUP_PROMPT_EN) + # Building a Config applies the stored language, so it is set there + # rather than around it. + self.write_config({"ui_language": "tr"}) + self.assertEqual(cfg.Config().cleanup_prompt(), cfg.CLEANUP_PROMPT_TR) + + def test_a_prompt_of_your_own(self): + conf = self.config(cleanup_prompt=" Only fix punctuation. ") + self.assertEqual(conf.cleanup_prompt(), "Only fix punctuation.") + + def test_the_glossary_is_appended(self): + conf = self.config(transcribe_prompt="Paraşüt, OpenFrame") + self.assertIn("Paraşüt, OpenFrame", conf.cleanup_prompt()) + + def test_no_glossary_means_no_rule_about_one(self): + self.assertEqual(cfg.Config().cleanup_prompt(), cfg.CLEANUP_PROMPT_EN) + + def test_subtitles_use_their_own_prompt(self): + conf = cfg.Config() + self.assertNotEqual(conf.cleanup_prompt(subtitles=True), conf.cleanup_prompt()) + self.assertEqual(conf.cleanup_prompt(subtitles=True), + cfg.FILE_CLEANUP_PROMPT_EN) + + def test_a_subtitle_prompt_of_your_own(self): + conf = self.config(file_cleanup_prompt="Keep the stamps.") + self.assertEqual(conf.cleanup_prompt(subtitles=True), "Keep the stamps.") + self.assertEqual(conf.cleanup_prompt(), cfg.CLEANUP_PROMPT_EN) + + def test_timestamps_add_a_rule_about_them(self): + conf = cfg.Config() + self.assertGreater(len(conf.cleanup_prompt(with_timestamps=True)), + len(conf.cleanup_prompt())) + + def test_speakers_bring_the_names_in_with_them(self): + conf = self.config(meeting_self_name="Yusuf", meeting_other_name="Ayşe") + prompt = conf.cleanup_prompt(with_speakers=True) + self.assertIn("Yusuf", prompt) + self.assertIn("Ayşe", prompt) + + +class Participants(DikteTest): + def test_nobody_named(self): + self.assertEqual(cfg.Config().participants(), "") + + def test_the_two_sides_come_first(self): + conf = self.config(meeting_self_name="Yusuf", meeting_other_name="Ayşe", + meeting_participants="Mehmet") + self.assertEqual(conf.participants(), "Yusuf\nAyşe\nMehmet") + + def test_commas_and_newlines_both_separate(self): + conf = self.config(meeting_participants="Ayşe, Mehmet\nZeynep") + self.assertEqual(conf.participants().splitlines(), + ["Ayşe", "Mehmet", "Zeynep"]) + + def test_a_name_listed_twice_appears_once(self): + conf = self.config(meeting_self_name="Yusuf", + meeting_participants="yusuf, Ayşe") + self.assertEqual(conf.participants(), "Yusuf\nAyşe") + + def test_blank_entries_are_dropped(self): + conf = self.config(meeting_participants="Ayşe,, ,\nMehmet") + self.assertEqual(conf.participants(), "Ayşe\nMehmet") + + +class MeetingSettings(DikteTest): + def test_the_hint_carries_the_glossary_and_the_names(self): + conf = self.config(transcribe_prompt="OpenFrame", meeting_self_name="Yusuf") + self.assertEqual(conf.meeting_hint(), "OpenFrame\nYusuf") + + def test_the_hint_with_neither(self): + self.assertEqual(cfg.Config().meeting_hint(), "") + + def test_the_speaker_labels_fall_back_to_the_language(self): + self.assertEqual(cfg.Config().speaker_names(), ("Me", "Other side")) + self.write_config({"ui_language": "tr"}) + self.assertEqual(cfg.Config().speaker_names(), ("Ben", "Karşı taraf")) + + def test_named_speakers_are_used_as_given(self): + conf = self.config(meeting_self_name="Yusuf", meeting_other_name="Ayşe") + self.assertEqual(conf.speaker_names(), ("Yusuf", "Ayşe")) + + def test_the_meeting_prompt_lists_who_was_there(self): + conf = self.config(meeting_self_name="Yusuf", meeting_other_name="Ayşe") + self.assertIn("Yusuf", conf.meeting_prompt()) + + def test_the_meeting_prompt_with_nobody_named(self): + self.assertEqual(cfg.Config().meeting_prompt(), cfg.MEETING_PROMPT_EN) + + +class History(DikteTest): + def entry(self, text): + return {"ts": "2026-08-01 10:00:00", "text": text, "raw": text} + + def test_nothing_written_yet(self): + self.assertEqual(cfg.read_history(), []) + + def test_what_goes_in_comes_out_newest_last(self): + for text in ("first", "second"): + cfg.append_history(self.entry(text)) + self.assertEqual([row["text"] for row in cfg.read_history()], + ["first", "second"]) + + def test_a_limit_reads_the_tail(self): + for index in range(5): + cfg.append_history(self.entry(str(index))) + self.assertEqual([row["text"] for row in cfg.read_history(2)], ["3", "4"]) + + def test_a_limit_of_zero_reads_everything(self): + for index in range(3): + cfg.append_history(self.entry(str(index))) + self.assertEqual(len(cfg.read_history(0)), 3) + + def test_a_line_that_is_not_json_is_skipped_rather_than_fatal(self): + cfg.append_history(self.entry("good")) + with open(cfg.HISTORY_FILE, "a", encoding="utf-8") as fh: + fh.write("half a line, no newline at the end of the world\n") + cfg.append_history(self.entry("also good")) + self.assertEqual([row["text"] for row in cfg.read_history()], + ["good", "also good"]) + + def test_turkish_survives_the_round_trip(self): + cfg.append_history(self.entry("Öğleden sonra görüşürüz.")) + self.assertEqual(cfg.read_history()[0]["text"], "Öğleden sonra görüşürüz.") + + def test_trimming_keeps_the_newest(self): + for index in range(10): + cfg.append_history(self.entry(str(index))) + cfg.trim_history(3) + self.assertEqual([row["text"] for row in cfg.read_history()], + ["7", "8", "9"]) + + def test_a_limit_of_zero_keeps_everything(self): + for index in range(4): + cfg.append_history(self.entry(str(index))) + cfg.trim_history(0) + self.assertEqual(len(cfg.read_history()), 4) + + def test_trimming_a_file_that_is_already_short_enough(self): + cfg.append_history(self.entry("only one")) + cfg.trim_history(200) + self.assertEqual(len(cfg.read_history()), 1) + + def test_trimming_before_anything_was_written(self): + cfg.trim_history(10) # must not raise + + def test_deleting_matches_on_content_not_on_position(self): + """The worker may have appended a row since the list was read.""" + rows = [self.entry("a"), self.entry("b"), self.entry("c")] + for row in rows: + cfg.append_history(row) + cfg.delete_history([rows[1]]) + self.assertEqual([row["text"] for row in cfg.read_history()], ["a", "c"]) + + def test_deleting_is_insensitive_to_key_order(self): + cfg.append_history({"ts": "now", "text": "hello"}) + cfg.delete_history([{"text": "hello", "ts": "now"}]) + self.assertEqual(cfg.read_history(), []) + + def test_deleting_nothing_touches_nothing(self): + cfg.append_history(self.entry("a")) + cfg.delete_history([]) + self.assertEqual(len(cfg.read_history()), 1) + + def test_clearing(self): + cfg.append_history(self.entry("a")) + cfg.clear_history() + self.assertEqual(cfg.read_history(), []) + + def test_clearing_a_history_that_is_not_there(self): + cfg.clear_history() # must not raise + + +class Meetings(DikteTest): + def entry(self, base, **changes): + row = {"base": base, "ts": "2026-08-01 10:00", "title": "", + "duration": 60.0, "status": "recorded", "error": "", "model": ""} + row.update(changes) + return row + + def test_nothing_recorded_yet(self): + self.assertEqual(cfg.read_meetings(), []) + + def test_the_document_and_the_recording_share_a_stem(self): + doc, wav = cfg.meeting_paths("20260801-100000") + self.assertEqual(doc.name, "20260801-100000.md") + self.assertEqual(wav.name, "20260801-100000.wav") + self.assertEqual(doc.parent, cfg.MEETINGS_DIR) + + def test_saving_and_reading_back(self): + cfg.save_meeting(self.entry("a")) + cfg.save_meeting(self.entry("b")) + self.assertEqual([row["base"] for row in cfg.read_meetings()], ["a", "b"]) + + def test_saving_the_same_base_replaces_rather_than_appends(self): + cfg.save_meeting(self.entry("a")) + cfg.save_meeting(self.entry("a", status="done")) + rows = cfg.read_meetings() + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["status"], "done") + + def test_a_row_with_no_base_is_ignored(self): + cfg.DATA_DIR.mkdir(parents=True, exist_ok=True) + cfg.MEETINGS_FILE.write_text( + json.dumps({"title": "orphan"}) + "\n" + json.dumps(self.entry("a")) + "\n", + encoding="utf-8") + self.assertEqual([row["base"] for row in cfg.read_meetings()], ["a"]) + + def test_a_broken_line_is_skipped(self): + cfg.save_meeting(self.entry("a")) + with open(cfg.MEETINGS_FILE, "a", encoding="utf-8") as fh: + fh.write("{oh dear\n") + self.assertEqual(len(cfg.read_meetings()), 1) + + def test_updating_patches_and_hands_the_row_back(self): + cfg.save_meeting(self.entry("a")) + row = cfg.update_meeting("a", status="done", title="Kickoff") + self.assertEqual(row["status"], "done") + self.assertEqual(cfg.read_meetings()[0]["title"], "Kickoff") + + def test_updating_one_that_is_gone(self): + self.assertIsNone(cfg.update_meeting("nope", status="done")) + + def test_deleting_takes_the_files_with_it(self): + cfg.save_meeting(self.entry("a")) + doc, wav = cfg.meeting_paths("a") + doc.parent.mkdir(parents=True, exist_ok=True) + doc.write_text("# minutes", encoding="utf-8") + wav.write_bytes(b"RIFF") + cfg.delete_meetings(["a"]) + self.assertEqual(cfg.read_meetings(), []) + self.assertFalse(doc.exists()) + self.assertFalse(wav.exists()) + + def test_deleting_a_row_whose_files_are_already_gone(self): + cfg.save_meeting(self.entry("a")) + cfg.delete_meetings(["a"]) # must not raise + self.assertEqual(cfg.read_meetings(), []) + + def test_deleting_nothing(self): + cfg.save_meeting(self.entry("a")) + cfg.delete_meetings([]) + self.assertEqual(len(cfg.read_meetings()), 1) + + +class Defaults(unittest.TestCase): + """The table itself, which every command line and settings tab reads.""" + + def test_no_setting_defaults_to_none(self): + """cli._coerce switches on the type of the default, so there has to be one.""" + for key, value in cfg.DEFAULTS.items(): + with self.subTest(key=key): + self.assertIsNotNone(value) + + def test_the_prompts_ship_empty_so_the_default_can_improve(self): + for key in ("cleanup_prompt", "file_cleanup_prompt", "meeting_prompt", + "assistant_prompt"): + with self.subTest(key=key): + self.assertEqual(cfg.DEFAULTS[key], "") + + def test_the_keys_ship_empty(self): + self.assertEqual(cfg.DEFAULTS["openai_api_key"], "") + self.assertEqual(cfg.DEFAULTS["openrouter_api_key"], "") + + def test_every_language_specific_prompt_has_both_languages(self): + for name in ("CLEANUP_PROMPT", "FILE_CLEANUP_PROMPT", "MEETING_PROMPT", + "ASSISTANT_PROMPT"): + for suffix in ("EN", "TR"): + with self.subTest(prompt=f"{name}_{suffix}"): + self.assertTrue(getattr(cfg, f"{name}_{suffix}").strip()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_filetranscribe.py b/tests/test_filetranscribe.py new file mode 100644 index 0000000..def4f5c --- /dev/null +++ b/tests/test_filetranscribe.py @@ -0,0 +1,237 @@ +"""Transcribing a file: splitting it, stamping it, and writing subtitles. + +to_srt is the awkward one. The text is the authority on wording and the segments +on timing, and they meet at a whole-second stamp that a cleanup model was asked +to leave alone. It has to survive a model that wrapped a line, dropped one, or +made up a stamp nobody recorded. +""" + +import contextlib +import unittest +import wave +from unittest import mock + +import api +import filetranscribe as ft +from tests.support import DikteTest, make_wav, silence, tone + + +class Timestamps(unittest.TestCase): + def test_under_an_hour(self): + self.assertEqual(ft.format_timestamp(0), "00:00") + self.assertEqual(ft.format_timestamp(65.9), "01:05") + self.assertEqual(ft.format_timestamp(599), "09:59") + + def test_past_an_hour_the_hours_show(self): + self.assertEqual(ft.format_timestamp(3600), "1:00:00") + self.assertEqual(ft.format_timestamp(3725), "1:02:05") + + def test_srt_wants_milliseconds_and_a_comma(self): + self.assertEqual(ft.srt_timestamp(0), "00:00:00,000") + self.assertEqual(ft.srt_timestamp(1.5), "00:00:01,500") + self.assertEqual(ft.srt_timestamp(3725.25), "01:02:05,250") + + def test_a_negative_start_is_pulled_up_to_zero(self): + self.assertEqual(ft.srt_timestamp(-3), "00:00:00,000") + + +class ToSrt(unittest.TestCase): + def test_nothing_to_do(self): + self.assertEqual(ft.to_srt("", []), "") + self.assertEqual(ft.to_srt("no stamps here", []), "") + + def test_a_cue_takes_its_timing_from_the_segment_it_came_from(self): + srt = ft.to_srt("[00:01] Hello there.", [(1.25, 2.75, "hello there")]) + self.assertIn("00:00:01,250 --> 00:00:02,750", srt) + self.assertIn("Hello there.", srt) + + def test_the_text_wins_on_wording(self): + """Cleanup edits survive; only the timing comes from the segments.""" + srt = ft.to_srt("[00:01] Hello there.", [(1.0, 2.0, "uh hello uh there")]) + self.assertIn("Hello there.", srt) + self.assertNotIn("uh", srt) + + def test_cues_are_numbered_from_one(self): + srt = ft.to_srt("[00:00] One\n[00:02] Two", + [(0.0, 1.0, "One"), (2.0, 3.0, "Two")]) + self.assertTrue(srt.startswith("1\n")) + self.assertIn("\n2\n", srt) + + def test_a_wrapped_line_joins_the_cue_above_it(self): + srt = ft.to_srt("[00:01] Hello there,\nand welcome.", + [(1.0, 4.0, "hello there and welcome")]) + self.assertIn("Hello there, and welcome.", srt) + self.assertEqual(srt.count(" --> "), 1) + + def test_a_stamp_nobody_recorded_still_gets_timing(self): + srt = ft.to_srt("[00:05] Invented.", []) + self.assertIn("00:00:05,000 --> ", srt) + + def test_a_cue_with_no_end_runs_until_the_next_one(self): + srt = ft.to_srt("[00:00] One\n[00:04] Two", []) + self.assertIn("00:00:00,000 --> 00:00:04,000", srt) + + def test_the_last_cue_gets_a_minimum_length(self): + srt = ft.to_srt("[00:10] Last words.", []) + self.assertIn("00:00:10,000 --> 00:00:11,500", srt) + + def test_a_cue_is_cut_short_when_the_next_one_starts_first(self): + """Whisper's end times overlap now and then; subtitles must not.""" + srt = ft.to_srt("[00:00] One\n[00:02] Two", + [(0.0, 9.0, "One"), (2.0, 3.0, "Two")]) + self.assertIn("00:00:00,000 --> 00:00:02,000", srt) + + def test_blank_lines_and_empty_cues_are_dropped(self): + srt = ft.to_srt("[00:00] One\n\n[00:02]\n[00:03] Three", []) + self.assertEqual(srt.count(" --> "), 2) + + def test_the_hour_form_of_a_stamp_is_understood(self): + srt = ft.to_srt("[1:02:05] Late.", []) + self.assertIn("01:02:05,000", srt) + + def test_the_file_ends_with_a_newline(self): + self.assertTrue(ft.to_srt("[00:00] One", []).endswith("\n")) + + +class SplitText(unittest.TestCase): + def test_short_text_stays_whole(self): + self.assertEqual(ft.split_text("hello", False), ["hello"]) + + def test_a_long_transcript_is_broken_up(self): + text = " ".join(["word"] * 8000) + blocks = ft.split_text(text, False) + self.assertGreater(len(blocks), 1) + for block in blocks: + self.assertLessEqual(len(block), ft.CLEANUP_CHUNK_CHARS) + + def test_nothing_is_lost_in_the_splitting(self): + text = " ".join(f"word{index}" for index in range(4000)) + self.assertEqual(" ".join(ft.split_text(text, False)), text) + + def test_a_timestamped_transcript_is_never_broken_mid_line(self): + text = "\n".join(f"[00:{index:02d}] a line of some length here" + for index in range(600)) + blocks = ft.split_text(text, True) + self.assertGreater(len(blocks), 1) + for block in blocks: + for line in block.splitlines(): + self.assertTrue(line.startswith("[")) + + def test_a_single_line_longer_than_the_limit_is_kept_whole(self): + text = "x" * (ft.CLEANUP_CHUNK_CHARS + 100) + self.assertEqual(ft.split_text(text, True), [text]) + + +class SplitWav(DikteTest): + def wav(self, seconds, name="in.wav"): + return make_wav(self.path(name), silence(seconds)) + + def test_a_short_file_is_handed_back_as_it_is(self): + path = self.wav(2) + self.assertEqual(ft.split_wav(path, self.root), [(path, 0.0)]) + + def test_a_long_file_is_cut_at_the_chunk_length(self): + path = self.wav(5) + with mock.patch.object(ft, "CHUNK_SECONDS", 2): + chunks = ft.split_wav(path, self.root) + self.assertEqual([offset for _, offset in chunks], [0, 2, 4]) + + def test_the_chunks_add_up_to_the_original(self): + path = self.wav(5) + with mock.patch.object(ft, "CHUNK_SECONDS", 2): + chunks = ft.split_wav(path, self.root) + total = 0 + for chunk_path, _ in chunks: + with contextlib.closing(wave.open(chunk_path, "rb")) as wav: + total += wav.getnframes() + self.assertEqual(wav.getframerate(), 16000) + self.assertEqual(total, 5 * 16000) + + def test_the_chunks_do_not_write_over_each_other(self): + path = self.wav(5) + with mock.patch.object(ft, "CHUNK_SECONDS", 2): + chunks = ft.split_wav(path, self.root) + self.assertEqual(len({chunk for chunk, _ in chunks}), len(chunks)) + + +class Transcriber(DikteTest): + """The chain, with ffmpeg and both API calls faked.""" + + def setUp(self): + super().setUp() + self.source = make_wav(self.path("input.wav"), tone(1.0)) + self.conf = self.config(openrouter_api_key="sk-or-test") + + def run_chain(self, timestamps=False, cleanup=False, transcript="raw text", + segments=None, cleaned="clean text", fail=None): + worker = ft.FileTranscriber(self.conf) + done, failures, progress = [], [], [] + worker.finished.connect(lambda *args: done.append(args)) + worker.failed.connect(failures.append) + worker.progress.connect(progress.append) + + def to_wav(path, workdir): + return make_wav(self.path("converted.wav"), tone(1.0)) + + with mock.patch.object(ft, "_to_wav", side_effect=to_wav), \ + mock.patch.object(ft.shutil, "which", return_value="/usr/bin/ffmpeg"), \ + mock.patch.object(api, "transcribe", + side_effect=fail or (lambda *a, **k: transcript)), \ + mock.patch.object(api, "transcribe_segments", + return_value=segments or [(0.0, 1.0, "raw text")]), \ + mock.patch.object(api, "cleanup", return_value=cleaned) as cleanup_call: + # The chain is run here rather than through start(): its signals are + # emitted from the worker thread, and a queued connection would need + # an event loop to deliver them. This is the same code, one frame down. + worker._work(self.source, timestamps, cleanup) + return done, failures, progress, cleanup_call + + def test_plain_text_out(self): + done, failures, _, _ = self.run_chain() + self.assertEqual(failures, []) + self.assertEqual(done[0][0], "raw text") + + def test_cleanup_replaces_the_text(self): + done, _, _, _ = self.run_chain(cleanup=True) + self.assertEqual(done[0][0], "clean text") + + def test_cleanup_is_told_it_is_writing_subtitles(self): + _, _, _, cleanup_call = self.run_chain(cleanup=True) + prompt = cleanup_call.call_args.args[3] + self.assertEqual(prompt, self.conf.cleanup_prompt(subtitles=True)) + + def test_timestamps_come_back_as_segments_and_as_stamped_lines(self): + done, _, _, _ = self.run_chain( + timestamps=True, segments=[(0.0, 1.0, "one"), (2.0, 3.0, "two")]) + text, segments = done[0] + self.assertEqual(text, "[00:00] one\n[00:02] two") + self.assertEqual(len(segments), 2) + + def test_no_ffmpeg_installed(self): + worker = ft.FileTranscriber(self.conf) + failures = [] + worker.failed.connect(failures.append) + with mock.patch.object(ft.shutil, "which", return_value=None): + worker._work(self.source, False, False) + self.assertIn("ffmpeg", failures[0]) + + def test_an_api_failure_is_reported_rather_than_raised(self): + def boom(*args, **kwargs): + raise api.ApiError("OpenAI rejected the API key") + _, failures, _, _ = self.run_chain(fail=boom) + self.assertIn("rejected", failures[0]) + + def test_empty_text_is_not_sent_to_cleanup(self): + _, _, _, cleanup_call = self.run_chain(cleanup=True, transcript="") + cleanup_call.assert_not_called() + + 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) + self.assertTrue(worker.busy) + worker.start(self.source, False, False) + self.assertTrue(worker.busy) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_hotkey.py b/tests/test_hotkey.py new file mode 100644 index 0000000..b3e0d70 --- /dev/null +++ b/tests/test_hotkey.py @@ -0,0 +1,222 @@ +"""Parsing a shortcut, and the KDE entry it is installed as.""" + +import subprocess +import unittest +from unittest import mock + +import hotkey +from tests.support import DikteTest, FakeCompleted, linux_only + +SHORTCUTS_RC = """[services][dikte-toggle.desktop] +_launch=Ctrl+Space + +[services][org.kde.spectacle.desktop] +RectangularRegionScreenShot=Meta+Shift+Print\tMeta+Shift+Print\t + +[kwin] +Overview=Meta+W,Meta+W,Toggle Overview +Switch Window Down=Meta+Alt+Down,Meta+Alt+Down,Switch to Window Below +""" + + +class ParseShortcut(unittest.TestCase): + def test_the_default(self): + self.assertEqual(hotkey.parse_shortcut("Ctrl+Space"), ({"ctrl"}, 57)) + + def test_case_and_spacing_do_not_matter(self): + self.assertEqual(hotkey.parse_shortcut(" ctrl + SPACE "), ({"ctrl"}, 57)) + + def test_several_modifiers(self): + mods, key = hotkey.parse_shortcut("Ctrl+Alt+Shift+D") + self.assertEqual(mods, {"ctrl", "alt", "shift"}) + self.assertEqual(key, 32) + + def test_the_synonyms_land_on_one_name(self): + self.assertEqual(hotkey.parse_shortcut("Control+Space"), + hotkey.parse_shortcut("Ctrl+Space")) + self.assertEqual(hotkey.parse_shortcut("Meta+Space"), + hotkey.parse_shortcut("Super+Space")) + + def test_a_key_on_its_own(self): + self.assertEqual(hotkey.parse_shortcut("F9"), (set(), 67)) + + def test_modifiers_with_no_key(self): + self.assertEqual(hotkey.parse_shortcut("Ctrl+Alt"), (None, None)) + + def test_a_key_nobody_mapped(self): + self.assertEqual(hotkey.parse_shortcut("Ctrl+F13"), (None, None)) + + def test_nothing(self): + self.assertEqual(hotkey.parse_shortcut(""), (None, None)) + self.assertEqual(hotkey.parse_shortcut("+++"), (None, None)) + + def test_something_that_is_not_even_a_string(self): + self.assertEqual(hotkey.parse_shortcut(None), (None, None)) + + +class ModsMatch(unittest.TestCase): + """The combination has to be exact, or Ctrl+Space fires on Ctrl+Shift+Space.""" + + def match(self, held, wanted): + return hotkey.EvdevHotkey._mods_match(set(held), set(wanted)) + + def test_the_wanted_modifier_is_down(self): + self.assertTrue(self.match({29}, {"ctrl"})) + + def test_either_side_of_the_keyboard_counts(self): + self.assertTrue(self.match({97}, {"ctrl"})) + + def test_nothing_held_and_nothing_wanted(self): + self.assertTrue(self.match(set(), set())) + + def test_a_modifier_too_many(self): + self.assertFalse(self.match({29, 42}, {"ctrl"})) + + def test_a_modifier_missing(self): + self.assertFalse(self.match(set(), {"ctrl"})) + + def test_the_wrong_modifier(self): + self.assertFalse(self.match({56}, {"ctrl"})) + + +@linux_only +class Bindings(DikteTest): + """start() before it reaches /dev/input, which a test may not read.""" + + def test_a_binding_with_no_shortcut_is_skipped(self): + listener = hotkey.EvdevHotkey() + with mock.patch.object(listener, "_open_devices", return_value=[]): + self.assertFalse(listener.start({"toggle": "", "ask": ""})) + + def test_an_unparsable_shortcut_is_reported_and_the_rest_go_on(self): + listener = hotkey.EvdevHotkey() + self.addCleanup(listener.stop) + failures = [] + listener.failed.connect(failures.append) + with mock.patch.object(listener, "_open_devices", return_value=[99]), \ + mock.patch.object(hotkey.threading, "Thread"): + self.assertTrue(listener.start({"toggle": "Ctrl+F13", + "ask": "Ctrl+Space"})) + self.assertEqual(len(failures), 1) + self.assertIn("Ctrl+F13", failures[0]) + self.assertEqual(list(listener._bindings), [57]) + + def test_no_readable_devices_says_what_to_do_about_it(self): + listener = hotkey.EvdevHotkey() + failures = [] + listener.failed.connect(failures.append) + with mock.patch.object(listener, "_open_devices", return_value=[]): + self.assertFalse(listener.start({"toggle": "Ctrl+Space"})) + self.assertIn("input", failures[0]) + + def test_two_shortcuts_on_one_key_are_both_kept(self): + listener = hotkey.EvdevHotkey() + self.addCleanup(listener.stop) + with mock.patch.object(listener, "_open_devices", return_value=[]), \ + mock.patch.object(hotkey.threading, "Thread") as thread: + listener._open_devices.return_value = [99] + self.assertTrue(listener.start({"toggle": "Ctrl+Space", + "ask": "Ctrl+Alt+Space"})) + thread.assert_called_once() + self.assertEqual(len(listener._bindings[57]), 2) + + +@linux_only +class KdeShortcut(DikteTest): + def setUp(self): + super().setUp() + self.apps = self.path("applications") + self.apps.mkdir(parents=True) + self.rc = self.path("kglobalshortcutsrc") + self.patch_attr(hotkey, "APPLICATIONS_DIR", self.apps) + self.patch_attr(hotkey, "SHORTCUTS_FILE", self.rc) + + def test_installing_writes_a_desktop_file_kwin_will_launch(self): + with mock.patch.object(subprocess, "run", return_value=FakeCompleted()): + ok, message = hotkey.install_kde_shortcut("Ctrl+Space", "dikte toggle") + self.assertTrue(ok) + text = (self.apps / hotkey.DESKTOP_ID).read_text(encoding="utf-8") + self.assertIn("Exec=dikte toggle", text) + self.assertIn("X-KDE-GlobalAccel-CommandShortcut=true", text) + self.assertIn("log out", message) + + def test_the_shortcut_is_registered_under_the_desktop_id(self): + with mock.patch.object(subprocess, "run", + return_value=FakeCompleted()) as run: + hotkey.install_kde_shortcut("Meta+D", "dikte ask", + desktop_id=hotkey.ASK_DESKTOP_ID) + cmd = run.call_args.args[0] + self.assertEqual(cmd[0], "kwriteconfig6") + self.assertIn(hotkey.ASK_DESKTOP_ID, cmd) + self.assertEqual(cmd[-1], "Meta+D") + + def test_each_verb_gets_its_own_entry(self): + with mock.patch.object(subprocess, "run", return_value=FakeCompleted()): + hotkey.install_kde_shortcut("Ctrl+Space", "dikte toggle") + hotkey.install_kde_shortcut("Meta+M", "dikte meeting", + desktop_id=hotkey.MEETING_DESKTOP_ID) + self.assertTrue((self.apps / hotkey.DESKTOP_ID).exists()) + self.assertTrue((self.apps / hotkey.MEETING_DESKTOP_ID).exists()) + + def test_no_kwriteconfig_installed(self): + with mock.patch.object(subprocess, "run", side_effect=OSError("nope")): + ok, message = hotkey.install_kde_shortcut("Ctrl+Space", "dikte toggle") + self.assertFalse(ok) + self.assertIn("kglobalshortcutsrc", message) + + def test_removing_takes_the_desktop_file_with_it(self): + (self.apps / hotkey.DESKTOP_ID).write_text("[Desktop Entry]", encoding="utf-8") + with mock.patch.object(subprocess, "run", return_value=FakeCompleted()) as run: + hotkey.remove_kde_shortcut() + self.assertFalse((self.apps / hotkey.DESKTOP_ID).exists()) + self.assertIn("--delete", run.call_args.args[0]) + + def test_removing_one_that_was_never_installed(self): + with mock.patch.object(subprocess, "run", side_effect=OSError("nope")): + hotkey.remove_kde_shortcut() # must not raise + + def test_the_registered_shortcut_is_read_back(self): + (self.apps / hotkey.DESKTOP_ID).write_text("[Desktop Entry]", encoding="utf-8") + self.rc.write_text(SHORTCUTS_RC, encoding="utf-8") + self.assertEqual(hotkey.kde_shortcut_status(), "Ctrl+Space") + + def test_no_desktop_file_means_nothing_is_installed(self): + self.rc.write_text(SHORTCUTS_RC, encoding="utf-8") + self.assertIsNone(hotkey.kde_shortcut_status()) + + def test_a_desktop_file_with_no_entry_beside_it(self): + (self.apps / hotkey.DESKTOP_ID).write_text("[Desktop Entry]", encoding="utf-8") + self.rc.write_text("[kwin]\nOverview=Meta+W\n", encoding="utf-8") + self.assertIsNone(hotkey.kde_shortcut_status()) + + def test_no_shortcuts_file_at_all(self): + (self.apps / hotkey.DESKTOP_ID).write_text("[Desktop Entry]", encoding="utf-8") + self.assertIsNone(hotkey.kde_shortcut_status()) + + def test_a_combination_somebody_else_already_took(self): + self.rc.write_text(SHORTCUTS_RC, encoding="utf-8") + hits = hotkey.conflicting_shortcuts("Meta+W") + self.assertEqual(len(hits), 1) + self.assertIn("kwin", hits[0]) + self.assertIn("Overview", hits[0]) + + def test_our_own_entry_is_not_a_conflict(self): + self.rc.write_text(SHORTCUTS_RC, encoding="utf-8") + self.assertEqual(hotkey.conflicting_shortcuts("Ctrl+Space"), []) + + def test_a_free_combination(self): + self.rc.write_text(SHORTCUTS_RC, encoding="utf-8") + self.assertEqual(hotkey.conflicting_shortcuts("Ctrl+Alt+J"), []) + + def test_a_tab_separated_entry_is_read_too(self): + self.rc.write_text(SHORTCUTS_RC, encoding="utf-8") + hits = hotkey.conflicting_shortcuts("Meta+Shift+Print") + self.assertEqual(len(hits), 1) + self.assertIn("spectacle", hits[0]) + + def test_no_shortcuts_file_means_no_conflicts(self): + self.assertEqual(hotkey.conflicting_shortcuts("Ctrl+Space"), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_i18n.py b/tests/test_i18n.py new file mode 100644 index 0000000..0b1f9d5 --- /dev/null +++ b/tests/test_i18n.py @@ -0,0 +1,122 @@ +"""Translation lookup, and the table itself. + +The table test is the one that matters for a pull request: a Turkish string +whose placeholder was renamed raises KeyError at the moment the message is +shown, which is exactly when nobody is watching a terminal. +""" + +import string +import unittest +from unittest import mock + +import i18n +from tests.support import DikteTest + + +def placeholders(text): + return {name for _, name, _, _ in string.Formatter().parse(text) if name} + + +class Resolve(unittest.TestCase): + def test_an_explicit_language_wins(self): + with mock.patch.dict("os.environ", {"LANG": "tr_TR.UTF-8"}): + self.assertEqual(i18n.resolve("en"), "en") + self.assertEqual(i18n.resolve("tr"), "tr") + + def test_auto_reads_the_locale(self): + with mock.patch.dict("os.environ", {"LANG": "tr_TR.UTF-8"}, clear=True): + self.assertEqual(i18n.resolve("auto"), "tr") + with mock.patch.dict("os.environ", {"LANG": "en_GB.UTF-8"}, clear=True): + self.assertEqual(i18n.resolve("auto"), "en") + + def test_lc_all_outranks_lang(self): + with mock.patch.dict("os.environ", + {"LC_ALL": "tr_TR.UTF-8", "LANG": "en_GB.UTF-8"}, + clear=True): + self.assertEqual(i18n.resolve("auto"), "tr") + + def test_no_locale_at_all_falls_back_to_english(self): + with mock.patch.dict("os.environ", {}, clear=True): + self.assertEqual(i18n.resolve("auto"), "en") + + def test_an_unknown_code_is_not_taken_at_its_word(self): + with mock.patch.dict("os.environ", {}, clear=True): + self.assertEqual(i18n.resolve("de"), "en") + + +class Translate(DikteTest): + def test_english_returns_the_source_string(self): + self.assertEqual(i18n.t("Quit"), "Quit") + + def test_turkish_looks_the_string_up(self): + i18n.set_language("tr") + self.assertEqual(i18n.t("Quit"), "Çık") + + def test_an_untranslated_string_falls_through(self): + i18n.set_language("tr") + self.assertEqual(i18n.t("Nobody translated this"), "Nobody translated this") + + def test_placeholders_are_filled_in_both_languages(self): + self.assertEqual(i18n.t("Unknown key: {key}", key="f13"), "Unknown key: f13") + i18n.set_language("tr") + self.assertIn("f13", i18n.t("Unknown key: {key}", key="f13")) + + def test_a_string_with_no_arguments_is_not_formatted(self): + # Braces in the text itself must survive when nothing is passed in. + self.assertEqual(i18n.t("{not a placeholder}"), "{not a placeholder}") + + def test_a_placeholder_may_be_called_anything(self): + """Including the names of t()'s own parameters, which is why they are + positional-only: worker.py says {text}, and that has to work.""" + self.assertEqual(i18n.t("Discarded: {text}", text="hello"), "Discarded: hello") + self.assertEqual(i18n.name("Claude", case="dative"), "Claude") + + +class Names(DikteTest): + def test_english_leaves_the_name_alone(self): + self.assertEqual(i18n.name("Claude", "dative"), "Claude") + + def test_turkish_inflects_by_the_vowels_of_the_name(self): + i18n.set_language("tr") + self.assertEqual(i18n.name("Claude", "dative"), "Claude'a") + self.assertEqual(i18n.name("Codex", "dative"), "Codex'e") + self.assertEqual(i18n.name("OpenRouter", "accusative"), "OpenRouter'ı") + + def test_no_case_asked_for(self): + i18n.set_language("tr") + self.assertEqual(i18n.name("Claude"), "Claude") + + def test_an_unlisted_name_or_case_comes_back_unchanged(self): + i18n.set_language("tr") + self.assertEqual(i18n.name("Ollama", "dative"), "Ollama") + self.assertEqual(i18n.name("Claude", "ablative"), "Claude") + + +class Table(unittest.TestCase): + """The Turkish table against the English strings it stands in for.""" + + def test_every_translation_keeps_the_placeholders_of_its_source(self): + for source, translated in i18n.TR.items(): + with self.subTest(source=source[:50]): + self.assertEqual( + placeholders(source), placeholders(translated), + "the Turkish string does not take the same arguments", + ) + + def test_nothing_is_translated_to_an_empty_string(self): + for source, translated in i18n.TR.items(): + with self.subTest(source=source[:50]): + self.assertTrue(translated.strip()) + + def test_every_translation_is_formattable(self): + """Whatever the table holds, .format() must not blow up on it.""" + for source, translated in i18n.TR.items(): + names = placeholders(translated) + if not names: + continue + with self.subTest(source=source[:50]): + translated.format(**{key: "x" for key in names}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_ipc.py b/tests/test_ipc.py new file mode 100644 index 0000000..0409cad --- /dev/null +++ b/tests/test_ipc.py @@ -0,0 +1,159 @@ +"""The request a terminal sends to the running instance, and the reply it reads. + +The wire format has to stay backwards compatible in both directions: a stale KDE +shortcut still sends a bare verb, and an instance from before replies existed +answers by saying nothing at all. +""" + +import json +import os +import sys +import unittest +from unittest import mock + +import ipc + + +class FakeSocket: + """QLocalSocket, as much of it as ipc.send() touches.""" + + def __init__(self, connected=True, reply=b""): + self.connected = connected + self.reply = reply + self.written = b"" + self.server = "" + self.disconnected = False + self.read_limits = [] + self._served = False + + def connectToServer(self, name): + self.server = name + + def waitForConnected(self, ms): + return self.connected + + def write(self, data): + self.written += bytes(data) + + def flush(self): + pass + + def waitForBytesWritten(self, ms): + return True + + def waitForReadyRead(self, ms): + self.read_limits.append(ms) + if self._served or not self.reply: + return False + self._served = True + return True + + def readAll(self): + return self.reply + + def disconnectFromServer(self): + self.disconnected = True + + +class Paths(unittest.TestCase): + def test_script_path_points_at_dikte(self): + self.assertTrue(ipc.script_path().endswith("dikte.py")) + self.assertTrue(os.path.exists(ipc.script_path())) + + def test_the_shortcut_command_runs_it_with_this_interpreter(self): + command = ipc.command_for("toggle") + self.assertTrue(command.startswith(sys.executable)) + self.assertTrue(command.endswith(" toggle")) + + @unittest.skipUnless(hasattr(os, "getuid"), + "the socket is named after a user id, which Windows " + "has no equivalent of") + def test_the_socket_is_per_user(self): + self.assertEqual(ipc.SERVER_NAME, f"dikte-{os.getuid()}") + + +class Send(unittest.TestCase): + def send(self, socket, *args, **kwargs): + with mock.patch.object(ipc, "QLocalSocket", return_value=socket): + return ipc.send(*args, **kwargs) + + def written_line(self, socket): + return socket.written.decode("utf-8").strip() + + def test_nothing_running(self): + sock = FakeSocket(connected=False) + self.assertIsNone(self.send(sock, "toggle")) + self.assertEqual(sock.written, b"") + + def test_a_verb_on_its_own_goes_as_the_bare_word(self): + """An older instance only understands this, and it is how updates land.""" + sock = FakeSocket(reply=b'{"ok": true}\n') + self.send(sock, "restart") + self.assertEqual(self.written_line(sock), "restart") + + def test_a_verb_with_arguments_goes_as_json(self): + sock = FakeSocket(reply=b'{"ok": true}\n') + self.send(sock, "ask", text="what time is it") + self.assertEqual(json.loads(self.written_line(sock)), + {"cmd": "ask", "text": "what time is it"}) + + def test_arguments_that_are_none_are_left_out(self): + sock = FakeSocket(reply=b'{"ok": true}\n') + self.send(sock, "record", seconds=None, paste=False) + self.assertEqual(json.loads(self.written_line(sock)), + {"cmd": "record", "paste": False}) + + def test_asking_to_be_waited_for_says_so(self): + sock = FakeSocket(reply=b'{"ok": true, "text": "hello"}\n') + reply = self.send(sock, "toggle", wait=True) + self.assertTrue(json.loads(self.written_line(sock))["wait"]) + self.assertEqual(reply["text"], "hello") + + def test_a_wait_with_no_timeout_reads_without_a_deadline(self): + sock = FakeSocket(reply=b'{"ok": true}\n') + self.send(sock, "toggle", wait=True) + self.assertEqual(sock.read_limits[0], -1) + + def test_a_timeout_is_passed_on_in_milliseconds(self): + sock = FakeSocket(reply=b'{"ok": true}\n') + self.send(sock, "toggle", wait=True, timeout=2.5) + self.assertEqual(sock.read_limits[0], 2500) + + def test_a_fire_and_forget_verb_does_not_wait_around(self): + sock = FakeSocket(reply=b'{"ok": true}\n') + self.send(sock, "cancel") + self.assertEqual(sock.read_limits[0], ipc.CONNECT_MS) + + def test_the_reply_comes_back_as_it_was_sent(self): + sock = FakeSocket(reply=b'{"ok": false, "error": "no microphone"}\n') + self.assertEqual(self.send(sock, "toggle"), + {"ok": False, "error": "no microphone"}) + + def test_silence_from_an_old_instance_means_the_verb_went_through(self): + sock = FakeSocket(reply=b"") + reply = self.send(sock, "cancel") + self.assertTrue(reply["ok"]) + self.assertTrue(reply["legacy"]) + + def test_silence_during_a_wait_is_a_failure_with_a_way_out(self): + sock = FakeSocket(reply=b"") + reply = self.send(sock, "toggle", wait=True) + self.assertFalse(reply["ok"]) + self.assertIn("dikte restart", reply["error"]) + + def test_a_reply_that_is_not_json(self): + sock = FakeSocket(reply=b"ok\n") + self.assertEqual(self.send(sock, "toggle"), {"ok": True, "legacy": True}) + + def test_a_reply_that_is_json_but_not_an_object(self): + sock = FakeSocket(reply=b"[1, 2, 3]\n") + self.assertEqual(self.send(sock, "toggle"), {"ok": True, "legacy": True}) + + def test_the_socket_is_always_let_go_of(self): + sock = FakeSocket(reply=b'{"ok": true}\n') + self.send(sock, "toggle") + self.assertTrue(sock.disconnected) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_meeting.py b/tests/test_meeting.py new file mode 100644 index 0000000..471b77c --- /dev/null +++ b/tests/test_meeting.py @@ -0,0 +1,328 @@ +"""A two-channel recording turned into minutes. + +Attribution is settled by the channels rather than guessed, so the tests care +about what happens at the seams: the microphone picking the other side up +through the speakers, two people talking over each other, and a run that died +after the transcription and must not pay for it twice. +""" + +import contextlib +import unittest +import wave +from unittest import mock + +import api +import config as cfg +import meeting +from tests.support import DikteTest, make_wav, silence, speech, stereo, tone + + +def seg(start, end, text, speaker): + return (start, end, text, speaker) + + +class SplitChannels(DikteTest): + def stereo_file(self, left, right, name="meeting.wav"): + return make_wav(self.path(name), stereo(left, right), channels=2) + + def test_the_two_sides_come_out_as_separate_files(self): + path = self.stereo_file(tone(1.0, amplitude=16000), silence(1.0)) + mine, theirs = meeting.split_channels(path, self.root) + for side in (mine, theirs): + with contextlib.closing(wave.open(side, "rb")) as wav: + self.assertEqual(wav.getnchannels(), 1) + self.assertEqual(wav.getnframes(), 16000) + + def test_the_left_channel_is_mine(self): + path = self.stereo_file(tone(1.0, amplitude=16000), silence(1.0)) + mine, theirs = meeting.split_channels(path, self.root) + self.assertGreater(max(meeting.rms_series(mine)), 0.1) + self.assertEqual(max(meeting.rms_series(theirs)), 0.0) + + def test_a_recording_longer_than_the_read_block(self): + path = self.stereo_file(tone(3.0), silence(3.0)) + mine, _ = meeting.split_channels(path, self.root) + with contextlib.closing(wave.open(mine, "rb")) as wav: + self.assertEqual(wav.getnframes(), 3 * 16000) + + def test_a_dictation_is_not_a_meeting(self): + path = make_wav(self.path("mono.wav"), silence(1.0)) + with self.assertRaises(api.ApiError): + meeting.split_channels(path, self.root) + + +class RmsSeries(DikteTest): + def test_silence_reads_as_nothing(self): + path = make_wav(self.path("quiet.wav"), silence(1.0)) + self.assertEqual(set(meeting.rms_series(path)), {0.0}) + + def test_a_block_per_level_frame(self): + path = make_wav(self.path("clip.wav"), silence(1.0)) + self.assertEqual(len(meeting.rms_series(path)), + -(-16000 // meeting.LEVEL_FRAMES)) + + def test_a_loud_recording_reads_above_zero(self): + path = make_wav(self.path("loud.wav"), tone(1.0, amplitude=16000)) + self.assertGreater(max(meeting.rms_series(path)), 0.1) + + def test_the_rate_is_read_off_the_file(self): + path = make_wav(self.path("clip.wav"), silence(0.1), rate=8000) + self.assertEqual(meeting.wav_rate(path), 8000) + + +class MergeTurns(unittest.TestCase): + def test_one_timeline_out_of_two_channels(self): + turns = meeting.merge_turns([ + seg(5.0, 6.0, "and you?", "theirs"), + seg(0.0, 1.0, "hello", "mine"), + ]) + self.assertEqual([(start, speaker) for start, speaker, _ in turns], + [(0.0, "mine"), (5.0, "theirs")]) + + def test_one_person_carrying_on_stays_one_turn(self): + turns = meeting.merge_turns([ + seg(0.0, 1.0, "hello", "mine"), + seg(1.2, 2.0, "how are you", "mine"), + ]) + self.assertEqual(len(turns), 1) + self.assertEqual(turns[0][2], "hello how are you") + + def test_a_long_pause_starts_a_new_line(self): + turns = meeting.merge_turns([ + seg(0.0, 1.0, "hello", "mine"), + seg(20.0, 21.0, "still there?", "mine"), + ]) + self.assertEqual(len(turns), 2) + + def test_the_gap_is_measured_from_the_end_of_the_last_words(self): + turns = meeting.merge_turns([ + seg(0.0, 10.0, "a long sentence", "mine"), + seg(15.0, 16.0, "and another", "mine"), + ]) + self.assertEqual(len(turns), 1) + + def test_the_speaker_changing_always_starts_a_new_turn(self): + turns = meeting.merge_turns([ + seg(0.0, 1.0, "hello", "mine"), + seg(1.1, 2.0, "hi", "theirs"), + ]) + self.assertEqual(len(turns), 2) + + def test_my_microphone_hearing_them_through_the_speakers_is_dropped(self): + turns = meeting.merge_turns([ + seg(0.0, 2.0, "we should ship it on Friday", "theirs"), + seg(0.1, 2.0, "we should ship it on friday", "mine"), + ]) + self.assertEqual(len(turns), 1) + self.assertEqual(turns[0][1], "theirs") + + def test_talking_over_each_other_is_not_an_echo(self): + turns = meeting.merge_turns([ + seg(0.0, 2.0, "we should ship it on Friday", "theirs"), + seg(0.1, 2.0, "no, next week is better", "mine"), + ]) + self.assertEqual(len(turns), 2) + + def test_the_same_sentence_said_later_is_not_an_echo(self): + turns = meeting.merge_turns([ + seg(0.0, 2.0, "ship it on Friday", "theirs"), + seg(30.0, 32.0, "ship it on Friday", "mine"), + ]) + self.assertEqual(len(turns), 2) + + def test_a_side_that_transcribed_to_nothing_is_dropped(self): + turns = meeting.merge_turns([ + seg(0.0, 1.0, "...", "mine"), + seg(2.0, 3.0, "hello", "theirs"), + ]) + self.assertEqual(len(turns), 1) + + def test_nothing_was_said_at_all(self): + self.assertEqual(meeting.merge_turns([]), []) + + +class RenderTurns(unittest.TestCase): + def test_a_stamp_and_a_name_per_line(self): + text = meeting.render_turns( + [(0.0, "mine", "hello"), (65.0, "theirs", " hi ")], "Yusuf", "Ayşe") + self.assertEqual(text, "[00:00] Yusuf: hello\n[01:05] Ayşe: hi") + + def test_nothing_to_render(self): + self.assertEqual(meeting.render_turns([], "Me", "Them"), "") + + +class Document(DikteTest): + def test_a_heading_becomes_the_title(self): + self.assertEqual(meeting.split_title("# Kickoff\n\nWe agreed."), + ("Kickoff", "We agreed.")) + + def test_minutes_that_open_with_prose_have_no_title(self): + self.assertEqual(meeting.split_title("We agreed to ship."), + ("", "We agreed to ship.")) + + def test_nothing_written(self): + self.assertEqual(meeting.split_title(""), ("", "")) + self.assertEqual(meeting.split_title(None), ("", "")) + + def test_the_document_carries_the_title_the_date_and_the_length(self): + text = meeting.build_document("Kickoff", "2026-08-01 10:00", 3900, + "We agreed.", "[00:00] Me: hello") + self.assertTrue(text.startswith("# Kickoff")) + self.assertIn("2026-08-01 10:00", text) + self.assertIn("1 h 5 min", text) + + def test_the_transcript_can_be_read_back_out(self): + transcript = "[00:00] Me: hello\n[00:05] Other side: hi" + text = meeting.build_document("Kickoff", "now", 60, "We agreed.", transcript) + self.assertEqual(meeting.read_transcript(text), transcript) + + def test_a_document_with_no_minutes_yet_still_gives_its_transcript_back(self): + transcript = "[00:00] Me: hello" + text = meeting.build_document("Kickoff", "now", 60, "", transcript) + self.assertEqual(meeting.read_transcript(text), transcript) + + def test_a_document_written_by_something_else(self): + self.assertEqual(meeting.read_transcript("# Notes\n\nJust prose."), "") + + def test_the_marker_is_a_comment_so_it_never_renders(self): + self.assertTrue(meeting.TRANSCRIPT_MARKER.startswith("