mirror of
https://github.com/yusufipk/dikte.git
synced 2026-09-11 10:56:10 +00:00
huseyin-emre-tigciandClaude Fable 5
6f601ab969
listen() was the whole of the single-instance check and it cannot be one: a Windows named pipe takes a second server on the same name rather than refusing it, and everywhere else removeServer() first takes the live socket away from the instance holding it. Starting Dikte over a running Dikte then left two whole copies up, two tray icons and all, and the newer one's sweep() killed the whisper the older one was answering dictations with. On a machine that sleeps instead of logging out, that is one Start Menu click away, and it cost a real dictation before it was understood. A QLockFile in the data directory closes the race on all three systems, taken before the QApplication is even built, and behind it the probe is the side-effect-free status verb: the Settings window opens as the sign of life only when a bare second start deliberately asks for it, not as a byproduct of a probe racing a forwarded toggle. The two Windows relaunch dances collapse into one ipc.respawn. The checkout installer and the packaged setup each kept an autostart the other could not see, so a machine that tried both started two copies at every sign-in: each autostart now removes the other's entry, the silent every-start repair backs off from a Run value whose target still exists, and each uninstaller deletes the shared dikte.cmd only when the shim names its own install. Co-Authored-By: Claude Fable 5 <[email protected]>
260 lines
10 KiB
Python
260 lines
10 KiB
Python
"""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 pathlib
|
|
import shlex
|
|
import sys
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from dikte import ipc
|
|
from tests.support import DikteTest
|
|
|
|
|
|
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):
|
|
# By its parts rather than as a string: the separator is a backslash on
|
|
# Windows, and the path is what a shortcut there runs too.
|
|
path = pathlib.Path(ipc.script_path())
|
|
self.assertEqual(path.parts[-2:], ("dikte", "__main__.py"))
|
|
self.assertTrue(os.path.exists(ipc.script_path()))
|
|
|
|
def test_the_shortcut_command_runs_it_with_this_interpreter(self):
|
|
# Read back through the same quoting it went out with: a Windows path
|
|
# is spelled with backslashes and comes out of the join quoted.
|
|
self.assertEqual(shlex.split(ipc.command_for("toggle")),
|
|
[sys.executable, ipc.script_path(), "toggle"])
|
|
|
|
def test_a_packaged_build_names_itself_and_no_interpreter(self):
|
|
"""There is no __main__.py on disk in one, and sys.executable is the
|
|
build's own binary rather than a Python anybody could run it with."""
|
|
with mock.patch.object(sys, "frozen", True, create=True), \
|
|
mock.patch.object(sys, "executable", "/Applications/Dikte.app/Contents/MacOS/Dikte"), \
|
|
mock.patch.dict(os.environ, {}, clear=True):
|
|
self.assertEqual(ipc.launcher(),
|
|
["/Applications/Dikte.app/Contents/MacOS/Dikte"])
|
|
|
|
def test_an_appimage_names_the_file_rather_than_this_run_s_mount(self):
|
|
"""A shortcut written to the mount works until the next login."""
|
|
with mock.patch.object(sys, "frozen", True, create=True), \
|
|
mock.patch.object(sys, "executable", "/tmp/.mount_ab12/usr/bin/dikte"), \
|
|
mock.patch.dict(os.environ, {"APPIMAGE": "/home/me/Dikte.AppImage"}):
|
|
self.assertEqual(ipc.command_for("toggle"),
|
|
"/home/me/Dikte.AppImage toggle")
|
|
|
|
def test_a_path_with_a_space_in_it_is_quoted(self):
|
|
"""Which is every Mac, and an AppImage kept anywhere with a name."""
|
|
with mock.patch.object(sys, "frozen", True, create=True), \
|
|
mock.patch.dict(os.environ, {"APPIMAGE": "/home/me/My Things/Dikte.AppImage"}):
|
|
self.assertEqual(ipc.command_for("cancel"),
|
|
"'/home/me/My Things/Dikte.AppImage' cancel")
|
|
|
|
@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)
|
|
|
|
|
|
class AlreadyServing(unittest.TestCase):
|
|
"""The single-instance check, which listen() cannot be: a Windows pipe
|
|
takes a second server on the same name rather than refusing it."""
|
|
|
|
def probe(self, socket):
|
|
with mock.patch.object(ipc, "QLocalSocket", return_value=socket):
|
|
return ipc.already_serving()
|
|
|
|
def test_nothing_running_means_go_ahead(self):
|
|
self.assertFalse(self.probe(FakeSocket(connected=False)))
|
|
|
|
def test_an_answer_means_yield(self):
|
|
self.assertTrue(self.probe(FakeSocket(reply=b'{"ok": true}\n')))
|
|
|
|
def test_the_probe_has_no_side_effect(self):
|
|
"""A probe that opened a window would open it during the relaunch a
|
|
slow instance provokes, on top of the verb being forwarded."""
|
|
sock = FakeSocket(reply=b'{"ok": true}\n')
|
|
self.probe(sock)
|
|
self.assertEqual(sock.written.decode("utf-8").strip(), "status")
|
|
|
|
def test_an_instance_too_old_to_answer_still_counts_as_running(self):
|
|
self.assertTrue(self.probe(FakeSocket(reply=b"")))
|
|
|
|
|
|
class InstanceLock(DikteTest):
|
|
def setUp(self):
|
|
super().setUp()
|
|
# The lock derives its home from paths, which DikteTest's cfg patches
|
|
# do not cover; without this the test would write into the real one.
|
|
from dikte import paths
|
|
self.patch_attr(paths, "DATA_DIR", self.path("data"))
|
|
|
|
def test_one_holder_at_a_time(self):
|
|
first = ipc.instance_lock()
|
|
self.assertIsNotNone(first)
|
|
self.assertTrue(first.tryLock(0))
|
|
second = ipc.instance_lock()
|
|
self.assertFalse(second.tryLock(0))
|
|
first.unlock()
|
|
self.assertTrue(second.tryLock(0))
|
|
second.unlock()
|
|
|
|
def test_the_lock_lives_in_the_data_directory(self):
|
|
from dikte import paths
|
|
lock = ipc.instance_lock()
|
|
self.assertTrue(lock.tryLock(0))
|
|
self.assertTrue((paths.DATA_DIR / "dikte.lock").exists())
|
|
lock.unlock()
|
|
|
|
|
|
class Respawn(unittest.TestCase):
|
|
def test_windows_starts_a_detached_process_and_returns(self):
|
|
with mock.patch.object(sys, "platform", "win32"), \
|
|
mock.patch.object(ipc, "launcher", return_value=["py", "x"]), \
|
|
mock.patch.object(ipc.subprocess, "Popen") as popen:
|
|
ipc.respawn(["--gui"])
|
|
self.assertEqual(popen.call_args.args[0], ["py", "x", "--gui"])
|
|
self.assertEqual(popen.call_args.kwargs["creationflags"],
|
|
0x00000008 | 0x00000200)
|
|
|
|
def test_everywhere_else_the_process_is_replaced(self):
|
|
with mock.patch.object(sys, "platform", "linux"), \
|
|
mock.patch.object(ipc, "launcher", return_value=["py", "x"]), \
|
|
mock.patch.object(ipc.os, "execv") as execv:
|
|
ipc.respawn(["toggle", "--gui"])
|
|
execv.assert_called_once_with("py", ["py", "x", "toggle", "--gui"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|