497 lines
19 KiB
Python
497 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / "install.sh"
|
|
SYSTEMD_ROOT = SCRIPT.parents[2] / "systemd/user"
|
|
LEGACY_SERVICE = """# Managed by pi-dictate install.sh
|
|
[Unit]
|
|
Description=Pi local Parakeet dictation service
|
|
Requires=pi-dictate.socket
|
|
After=pipewire.service
|
|
|
|
[Service]
|
|
Type=simple
|
|
ExecStart=%h/.local/share/pi-dictate/venv/bin/pi-dictate-service
|
|
Environment=PI_DICTATE_MODEL_PATH=%h/.local/share/pi-dictate/models/parakeet-tdt-0.6b-v3/parakeet-tdt-0.6b-v3.nemo
|
|
Environment=PI_DICTATE_TERMS_PATH=%h/.local/share/pi-dictate/config/coding-terms.txt
|
|
Environment=PI_DICTATE_BOOST_ALPHA=1.0
|
|
Environment=PI_DICTATE_IDLE_SECONDS=1800
|
|
Environment=HF_HUB_OFFLINE=1
|
|
Environment=TRANSFORMERS_OFFLINE=1
|
|
Environment=NEMO_HOME=/tmp/pi-dictate-nemo
|
|
Environment=HF_HOME=/tmp/pi-dictate-huggingface
|
|
Environment=TRITON_CACHE_DIR=/tmp/pi-dictate-triton
|
|
Environment=NUMBA_CACHE_DIR=/tmp/pi-dictate-numba
|
|
Environment=PYTHONDONTWRITEBYTECODE=1
|
|
UMask=0077
|
|
LimitCORE=0
|
|
MemorySwapMax=0
|
|
NoNewPrivileges=yes
|
|
PrivateTmp=yes
|
|
ProtectClock=yes
|
|
ProtectControlGroups=yes
|
|
ProtectKernelLogs=yes
|
|
ProtectKernelModules=yes
|
|
ProtectKernelTunables=yes
|
|
ProtectSystem=strict
|
|
ProtectHome=read-only
|
|
RestrictAddressFamilies=AF_UNIX
|
|
IPAddressDeny=any
|
|
LockPersonality=yes
|
|
|
|
# The socket unit owns %t/pi-dictate.sock. The service must not unlink it.
|
|
"""
|
|
|
|
|
|
class InstallFlowTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.directory.name)
|
|
self.home = self.root / "home"
|
|
self.home.mkdir()
|
|
self.bin = self.root / "bin"
|
|
self.bin.mkdir()
|
|
uv = self.bin / "uv"
|
|
# A real uv sync creates the requested environment. Keep the fake
|
|
# faithful to that contract so transactional environment swapping is
|
|
# exercised without installing dependencies.
|
|
uv.write_text("#!/bin/sh\nmkdir -p \"$UV_PROJECT_ENVIRONMENT\"\nexit 0\n", encoding="utf-8")
|
|
uv.chmod(0o755)
|
|
systemctl = self.bin / "systemctl"
|
|
systemctl.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
systemctl.chmod(0o755)
|
|
|
|
def tearDown(self) -> None:
|
|
self.directory.cleanup()
|
|
|
|
def run_installer(self, *arguments: str) -> subprocess.CompletedProcess[str]:
|
|
environment = {
|
|
**os.environ,
|
|
"HOME": str(self.home),
|
|
"XDG_CONFIG_HOME": str(self.home / ".config"),
|
|
"PATH": f"{self.bin}:{os.environ['PATH']}",
|
|
}
|
|
return subprocess.run(
|
|
[str(SCRIPT), *arguments],
|
|
env=environment,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
def test_real_staged_cloud_venv_python_runs_after_transactional_move(self) -> None:
|
|
real_uv = shutil.which("uv")
|
|
self.assertIsNotNone(real_uv)
|
|
(self.bin / "uv").unlink()
|
|
source = self.root / "openai-key"
|
|
source.write_text("test-key", encoding="utf-8")
|
|
|
|
result = self.run_installer(
|
|
"install", "--backend", "openai", "--api-key-file", str(source)
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
python = self.home / ".local/share/pi-dictate/venv/bin/python"
|
|
executed = subprocess.run(
|
|
[str(python), "-c", "import pi_dictate_service"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
self.assertEqual(executed.returncode, 0, executed.stderr)
|
|
|
|
def test_upgrades_the_previous_service_unit_and_preserves_its_idle_setting(self) -> None:
|
|
unit_dir = self.home / ".config/systemd/user"
|
|
unit_dir.mkdir(parents=True)
|
|
(unit_dir / "pi-dictate.service").write_text(LEGACY_SERVICE, encoding="utf-8")
|
|
(unit_dir / "pi-dictate.socket").write_bytes(
|
|
(SYSTEMD_ROOT / "pi-dictate.socket").read_bytes()
|
|
)
|
|
source = self.root / "custom.nemo"
|
|
source.write_bytes(b"custom archive")
|
|
digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
|
|
result = self.run_installer(
|
|
"install",
|
|
"--backend",
|
|
"local",
|
|
"--model-source",
|
|
str(source),
|
|
"--model-sha256",
|
|
digest,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(
|
|
(unit_dir / "pi-dictate.service").read_bytes(),
|
|
(SYSTEMD_ROOT / "pi-dictate.service").read_bytes(),
|
|
)
|
|
config = json.loads(
|
|
(self.home / ".config/pi-dictate/config.json").read_text(encoding="utf-8")
|
|
)
|
|
self.assertEqual(config["idleSeconds"], 1800)
|
|
|
|
def test_rejects_symlinked_install_ancestry(self) -> None:
|
|
outside = self.root / "outside"
|
|
outside.mkdir()
|
|
(self.home / ".local").symlink_to(outside, target_is_directory=True)
|
|
source = self.root / "custom.nemo"
|
|
source.write_bytes(b"custom archive")
|
|
digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
|
|
result = self.run_installer(
|
|
"install",
|
|
"--backend",
|
|
"local",
|
|
"--model-source",
|
|
str(source),
|
|
"--model-sha256",
|
|
digest,
|
|
)
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("symlinked path ancestry", result.stderr)
|
|
self.assertEqual(list(outside.iterdir()), [])
|
|
|
|
def test_cloud_install_uses_only_cloud_profile_and_managed_key(self) -> None:
|
|
source = self.root / "meta-key"
|
|
source.write_text("test-key\n", encoding="utf-8")
|
|
|
|
result = self.run_installer(
|
|
"install", "--backend", "meta", "--api-key-file", str(source)
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertNotIn("test-key", result.stdout + result.stderr)
|
|
self.assertNotIn("nvidia", result.stdout)
|
|
self.assertFalse(
|
|
(self.home / ".local/share/pi-dictate/models").exists()
|
|
)
|
|
config_path = self.home / ".config/pi-dictate/config.json"
|
|
config = json.loads(config_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(config["backend"], "meta")
|
|
self.assertEqual(config["meta"]["model"], "muse-voice-transcribe-1.0")
|
|
self.assertEqual(config["meta"]["languageBias"], [])
|
|
self.assertEqual(
|
|
(self.home / ".config/pi-dictate/credentials/meta-api-key").read_text(
|
|
encoding="utf-8"
|
|
),
|
|
"test-key",
|
|
)
|
|
self.assertEqual(stat.S_IMODE(config_path.stat().st_mode), 0o600)
|
|
self.assertEqual(
|
|
stat.S_IMODE(
|
|
(self.home / ".config/pi-dictate/credentials/meta-api-key").stat().st_mode
|
|
),
|
|
0o600,
|
|
)
|
|
self.assertEqual(
|
|
(self.home / ".local/share/pi-dictate/.dependency-profile").read_text(
|
|
encoding="utf-8"
|
|
),
|
|
"cloud\n",
|
|
)
|
|
|
|
def test_check_rejects_values_that_runtime_configuration_rejects(self) -> None:
|
|
source = self.root / "openai-key"
|
|
source.write_text("test-key", encoding="utf-8")
|
|
self.assertEqual(
|
|
self.run_installer(
|
|
"install", "--backend", "openai", "--api-key-file", str(source)
|
|
).returncode,
|
|
0,
|
|
)
|
|
config_path = self.home / ".config/pi-dictate/config.json"
|
|
valid = json.loads(config_path.read_text(encoding="utf-8"))
|
|
invalid_configs = []
|
|
boolean_version = {**valid, "version": True}
|
|
invalid_configs.append(boolean_version)
|
|
unhashable_backend = {**valid, "backend": []}
|
|
invalid_configs.append(unhashable_backend)
|
|
long_model = json.loads(json.dumps(valid))
|
|
long_model["openai"]["model"] = "m" * 129
|
|
invalid_configs.append(long_model)
|
|
unhashable_delay = json.loads(json.dumps(valid))
|
|
unhashable_delay["openai"]["delay"] = []
|
|
invalid_configs.append(unhashable_delay)
|
|
relative_terms = json.loads(json.dumps(valid))
|
|
relative_terms["openai"]["termsPath"] = "relative/terms.txt"
|
|
invalid_configs.append(relative_terms)
|
|
|
|
for invalid in invalid_configs:
|
|
with self.subTest(invalid=invalid):
|
|
config_path.write_text(json.dumps(invalid), encoding="utf-8")
|
|
config_path.chmod(0o600)
|
|
result = self.run_installer("check")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertNotIn("Traceback", result.stderr)
|
|
|
|
def test_check_rejects_extra_dependency_marker_newlines(self) -> None:
|
|
source = self.root / "openai-key"
|
|
source.write_text("test-key", encoding="utf-8")
|
|
self.assertEqual(
|
|
self.run_installer(
|
|
"install", "--backend", "openai", "--api-key-file", str(source)
|
|
).returncode,
|
|
0,
|
|
)
|
|
marker = self.home / ".local/share/pi-dictate/.dependency-profile"
|
|
marker.write_bytes(b"cloud\n\n")
|
|
marker.chmod(0o600)
|
|
|
|
result = self.run_installer("check")
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("dependency profile", result.stderr)
|
|
|
|
def test_uninstall_removes_an_exact_openai_dropin(self) -> None:
|
|
source = self.root / "openai-key"
|
|
source.write_text("test-key", encoding="utf-8")
|
|
self.assertEqual(
|
|
self.run_installer(
|
|
"install", "--backend", "openai", "--api-key-file", str(source)
|
|
).returncode,
|
|
0,
|
|
)
|
|
|
|
result = self.run_installer("uninstall")
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertFalse(
|
|
(self.home / ".config/systemd/user/pi-dictate.service.d/10-pi-dictate-network.conf").exists()
|
|
)
|
|
|
|
def test_openai_configure_preserves_live_verified_delay(self) -> None:
|
|
source = self.root / "openai-key"
|
|
source.write_text("test-key", encoding="utf-8")
|
|
self.assertEqual(
|
|
self.run_installer(
|
|
"install", "--backend", "openai", "--api-key-file", str(source)
|
|
).returncode,
|
|
0,
|
|
)
|
|
config_path = self.home / ".config/pi-dictate/config.json"
|
|
config = json.loads(config_path.read_text(encoding="utf-8"))
|
|
config["openai"]["delay"] = "low"
|
|
config_path.write_text(json.dumps(config), encoding="utf-8")
|
|
config_path.chmod(0o600)
|
|
|
|
result = self.run_installer("configure", "--backend", "openai")
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
configured = json.loads(config_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(configured["openai"]["delay"], "low")
|
|
|
|
def test_meta_configure_preserves_valid_language_bias(self) -> None:
|
|
source = self.root / "meta-key"
|
|
source.write_text("test-key", encoding="utf-8")
|
|
self.assertEqual(
|
|
self.run_installer(
|
|
"install", "--backend", "meta", "--api-key-file", str(source)
|
|
).returncode,
|
|
0,
|
|
)
|
|
config_path = self.home / ".config/pi-dictate/config.json"
|
|
config = json.loads(config_path.read_text(encoding="utf-8"))
|
|
config["meta"]["languageBias"] = ["en", "fr"]
|
|
config_path.write_text(json.dumps(config), encoding="utf-8")
|
|
config_path.chmod(0o600)
|
|
|
|
result = self.run_installer("configure", "--backend", "meta")
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
configured = json.loads(config_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(configured["meta"]["languageBias"], ["en", "fr"])
|
|
|
|
def test_local_custom_archive_requires_and_records_explicit_digest(self) -> None:
|
|
source = self.root / "custom.nemo"
|
|
source.write_bytes(b"custom archive")
|
|
digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
|
|
result = self.run_installer(
|
|
"install",
|
|
"--backend",
|
|
"local",
|
|
"--model-source",
|
|
str(source),
|
|
"--model-sha256",
|
|
digest,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
config = json.loads(
|
|
(self.home / ".config/pi-dictate/config.json").read_text(encoding="utf-8")
|
|
)
|
|
self.assertEqual(config["local"]["modelSha256"], digest)
|
|
installed = self.home / (
|
|
".local/share/pi-dictate/models/parakeet-tdt-0.6b-v3/"
|
|
"parakeet-tdt-0.6b-v3.nemo"
|
|
)
|
|
self.assertEqual(installed.read_bytes(), source.read_bytes())
|
|
|
|
def test_replacing_a_custom_local_model_preserves_its_configured_path(self) -> None:
|
|
first = self.root / "first.nemo"
|
|
first.write_bytes(b"first archive")
|
|
first_digest = hashlib.sha256(first.read_bytes()).hexdigest()
|
|
self.assertEqual(
|
|
self.run_installer(
|
|
"install", "--backend", "local", "--model-source", str(first),
|
|
"--model-sha256", first_digest,
|
|
).returncode,
|
|
0,
|
|
)
|
|
config_path = self.home / ".config/pi-dictate/config.json"
|
|
config = json.loads(config_path.read_text(encoding="utf-8"))
|
|
custom_model = self.root / "custom-models/model.nemo"
|
|
custom_model.parent.mkdir()
|
|
custom_model.write_bytes(first.read_bytes())
|
|
config["local"]["modelPath"] = str(custom_model)
|
|
config_path.write_text(json.dumps(config), encoding="utf-8")
|
|
config_path.chmod(0o600)
|
|
second = self.root / "second.nemo"
|
|
second.write_bytes(b"second archive")
|
|
second_digest = hashlib.sha256(second.read_bytes()).hexdigest()
|
|
|
|
result = self.run_installer(
|
|
"configure", "--backend", "local", "--model-source", str(second),
|
|
"--model-sha256", second_digest,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
configured = json.loads(config_path.read_text(encoding="utf-8"))
|
|
self.assertEqual(configured["local"]["modelPath"], str(custom_model))
|
|
self.assertEqual(custom_model.read_bytes(), second.read_bytes())
|
|
self.assertFalse(list(custom_model.parent.glob("*.pi-dictate-rollback.*")))
|
|
|
|
def test_switching_from_cloud_to_local_accepts_a_local_archive(self) -> None:
|
|
key = self.root / "meta-key"
|
|
key.write_text("test-key", encoding="utf-8")
|
|
self.assertEqual(
|
|
self.run_installer(
|
|
"install", "--backend", "meta", "--api-key-file", str(key)
|
|
).returncode,
|
|
0,
|
|
)
|
|
source = self.root / "custom.nemo"
|
|
source.write_bytes(b"custom archive")
|
|
digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
|
|
result = self.run_installer(
|
|
"configure",
|
|
"--backend",
|
|
"local",
|
|
"--model-source",
|
|
str(source),
|
|
"--model-sha256",
|
|
digest,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
config = json.loads(
|
|
(self.home / ".config/pi-dictate/config.json").read_text(encoding="utf-8")
|
|
)
|
|
self.assertEqual(config["backend"], "local")
|
|
self.assertEqual(config["local"]["modelSha256"], digest)
|
|
self.assertIn("meta", config)
|
|
|
|
def test_systemd_failure_rolls_back_staged_install(self) -> None:
|
|
old_venv = self.home / ".local/share/pi-dictate/venv"
|
|
old_venv.mkdir(parents=True)
|
|
(old_venv / "previous-runtime").write_text("old", encoding="utf-8")
|
|
source = self.root / "meta-key"
|
|
source.write_text("test-key", encoding="utf-8")
|
|
(self.bin / "systemctl").write_text(
|
|
"#!/bin/sh\necho \"$*\" >> \"$HOME/systemctl.log\"\n"
|
|
"case \"$*\" in *'is-enabled'*) exit 1;; "
|
|
"*'stop pi-dictate.service'*) exit 9;; esac\nexit 0\n",
|
|
encoding="utf-8",
|
|
)
|
|
(self.bin / "systemctl").chmod(0o755)
|
|
|
|
result = self.run_installer(
|
|
"install", "--backend", "meta", "--api-key-file", str(source)
|
|
)
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse((self.home / ".config/pi-dictate/config.json").exists())
|
|
self.assertFalse((self.home / ".local/share/pi-dictate/.dependency-profile").exists())
|
|
self.assertFalse((self.home / ".local/share/pi-dictate/config/coding-terms.txt").exists())
|
|
self.assertEqual((old_venv / "previous-runtime").read_text(encoding="utf-8"), "old")
|
|
systemctl_calls = (self.home / "systemctl.log").read_text(encoding="utf-8")
|
|
self.assertGreaterEqual(systemctl_calls.count("daemon-reload"), 1)
|
|
self.assertIn("disable pi-dictate.socket", systemctl_calls)
|
|
|
|
def test_invalid_cloud_key_does_not_commit_a_dependency_marker(self) -> None:
|
|
source = self.root / "invalid-key"
|
|
source.write_text("two\nlines\n", encoding="utf-8")
|
|
|
|
result = self.run_installer(
|
|
"install", "--backend", "meta", "--api-key-file", str(source)
|
|
)
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(
|
|
(self.home / ".local/share/pi-dictate/.dependency-profile").exists()
|
|
)
|
|
|
|
def test_dictionary_symlink_is_rejected(self) -> None:
|
|
dictionary = self.home / ".local/share/pi-dictate/config/coding-terms.txt"
|
|
dictionary.parent.mkdir(parents=True)
|
|
target = self.root / "unexpected-terms.txt"
|
|
target.write_text("unexpected", encoding="utf-8")
|
|
dictionary.symlink_to(target)
|
|
source = self.root / "custom.nemo"
|
|
source.write_bytes(b"custom archive")
|
|
digest = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
|
|
result = self.run_installer(
|
|
"install",
|
|
"--backend",
|
|
"local",
|
|
"--model-source",
|
|
str(source),
|
|
"--model-sha256",
|
|
digest,
|
|
)
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("non-regular coding dictionary", result.stderr)
|
|
|
|
def test_uninstall_preserves_config_unless_purge_is_requested(self) -> None:
|
|
source = self.root / "meta-key"
|
|
source.write_text("test-key", encoding="utf-8")
|
|
self.assertEqual(
|
|
self.run_installer(
|
|
"install", "--backend", "meta", "--api-key-file", str(source)
|
|
).returncode,
|
|
0,
|
|
)
|
|
|
|
result = self.run_installer("uninstall")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertTrue((self.home / ".config/pi-dictate/config.json").exists())
|
|
self.assertFalse((self.home / ".local/share/pi-dictate").exists())
|
|
|
|
# Reinstalling from the preserved key is possible without passing a
|
|
# literal secret or a second key source.
|
|
result = self.run_installer("install", "--backend", "meta")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(
|
|
self.run_installer("uninstall", "--purge-config").returncode, 0
|
|
)
|
|
self.assertFalse((self.home / ".config/pi-dictate").exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|