Files
voculinux/tests/test_custom_dictionary.py
soubarnak ea7be5f111 feat(identity): rename user-visible identity to voculinux with one-time legacy import
The fork takes its own name everywhere a user can see it: binaries
(voculinux, voculinux-gui), desktop entry, autostart, tray, D-Bus
namespace (com.voculinux.Voculinux, no alias for the old name),
config/data/cache dirs, single-instance lock, IBus engine identity,
tray icons, resource prefixes, VOCULINUX_* environment names, and the
update-checker user agent. The internal Python module stays vocalinux
and the distribution metadata keeps the upstream URLs as provenance.

On first run a previous Vocalinux install's config.json, custom
dictionary, and portal restore token are copied once into the new
locations (new config always wins, no merge, no write-back); models
re-resolve through the pinned-checksum download path. An offline
naming-sweep test proves no old-name strings remain on user-visible
surfaces outside the allowlist.

Refs #18
2026-11-06 01:29:39 +05:30

1107 lines
46 KiB
Python

"""Tests for the file-backed custom dictionary contract."""
import json
import re
import stat
import threading
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock, Mock, patch
from vocalinux.custom_dictionary import (
CORRECTIONS_FILENAME,
DEFAULT_TERMS_PATH,
LEGACY_DEFAULT_TERMS_PATH,
MAX_CORRECTION_CHARACTERS,
MAX_TERMS_FILE_BYTES,
MAX_TERMS_YIELDED,
TERMS_FILENAME,
CustomDictionaryManager,
apply_corrections,
normalize_corrections,
)
from vocalinux.main import parse_arguments
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
class FakeConfig:
"""Minimal ConfigManager substitute with controllable persistence."""
def __init__(self, values: dict[str, dict[str, Any]] | None = None, save_result: bool = True):
self.values = values or {
"dictionary": {
"enabled": False,
"file_path": DEFAULT_TERMS_PATH,
"max_words": 200,
}
}
self.save_result = save_result
def get(self, section: str, key: str, default: Any = None) -> Any:
"""Return one configured value."""
return self.values.get(section, {}).get(key, default)
def set(self, section: str, key: str, value: Any) -> bool:
"""Set one configured value."""
self.values.setdefault(section, {})[key] = value
return True
def save_config(self) -> bool:
"""Return the requested persistence result."""
return self.save_result
def manager_at(
tmp_path: Path, monkeypatch, config: FakeConfig | None = None
) -> CustomDictionaryManager:
"""Create a manager whose configured files live in pytest's temp directory."""
monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path))
config = config or FakeConfig()
dictionary = config.values.setdefault("dictionary", {})
if dictionary.get("file_path") in (None, DEFAULT_TERMS_PATH):
dictionary["file_path"] = str(tmp_path / TERMS_FILENAME)
return CustomDictionaryManager(config)
def recognition_manager(
engine: str, dictionary: CustomDictionaryManager, advanced_prompt: str = ""
) -> SpeechRecognitionManager:
"""Build a recognition manager without loading a real speech model."""
with patch.object(SpeechRecognitionManager, "_init_vosk"):
with patch.object(SpeechRecognitionManager, "_init_whisper"):
with patch.object(SpeechRecognitionManager, "_init_whispercpp"):
return SpeechRecognitionManager(
engine=engine,
dictionary_manager=dictionary,
whispercpp_initial_prompt=advanced_prompt,
)
def mock_numpy() -> MagicMock:
"""Return the minimum NumPy surface used by transcription helpers."""
numpy = MagicMock()
numpy.frombuffer.return_value = MagicMock(__len__=lambda _: 16_000)
numpy.frombuffer.return_value.astype.return_value = numpy.frombuffer.return_value
numpy.int16 = "int16"
numpy.float32 = "float32"
return numpy
def test_terms_are_live_reloaded_and_scanner_friendly(tmp_path: Path, monkeypatch) -> None:
"""The UTF-8 line file accepts comments and applies an external replacement."""
manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
terms_file = tmp_path / TERMS_FILENAME
terms_file.write_text("# scanner comment\nVoculinux\nvoculinux\nSupabase\n", encoding="utf-8")
assert manager.get_terms() == ["Voculinux", "Supabase"]
assert manager.build_initial_prompt() == "Voculinux Supabase"
terms_file.write_text("PyGObject\n", encoding="utf-8")
assert manager.build_initial_prompt() == "PyGObject"
def test_terms_save_is_atomic_shape_and_transient_override_is_not_written(
tmp_path: Path, monkeypatch
) -> None:
"""UI saves standard terms safely while CLI terms remain session-only."""
manager = manager_at(tmp_path, monkeypatch)
assert manager.save_terms(["Voculinux", "voculinux", "# ignored", " PyGObject "])
assert (tmp_path / TERMS_FILENAME).read_text(encoding="utf-8") == "Voculinux\nPyGObject\n"
override = tmp_path / "override.txt"
override.write_text("Override\n", encoding="utf-8")
transient = CustomDictionaryManager(FakeConfig(), str(override))
assert transient.terms_enabled()
assert transient.get_terms() == ["Override"]
assert not transient.save_terms(["No write"])
def test_incremental_term_edits_preserve_comments_and_blank_lines(
tmp_path: Path, monkeypatch
) -> None:
"""GTK add/remove operations retain unrelated scanner-managed source lines."""
manager = manager_at(tmp_path, monkeypatch)
terms_file = tmp_path / TERMS_FILENAME
terms_file.write_text("# scanner note\n\nVoculinux\n# retain me\nPyGObject\n", encoding="utf-8")
assert manager.add_term("Supabase")
assert terms_file.read_text(encoding="utf-8") == (
"# scanner note\n\nVoculinux\n# retain me\nPyGObject\nSupabase\n"
)
assert manager.remove_term("PyGObject")
assert terms_file.read_text(encoding="utf-8") == (
"# scanner note\n\nVoculinux\n# retain me\nSupabase\n"
)
def test_enable_rollback_when_config_persistence_fails(tmp_path: Path, monkeypatch) -> None:
"""A failed config save preserves the previous in-memory setting."""
config = FakeConfig(save_result=False)
manager = manager_at(tmp_path, monkeypatch, config)
assert not manager.set_terms_enabled(True)
assert not manager.terms_enabled()
def test_prompt_read_stops_at_the_terms_limit(tmp_path: Path, monkeypatch) -> None:
"""Prompt assembly stops processing terms after the configured term cap."""
config = FakeConfig({"dictionary": {"enabled": True, "max_words": 2}})
manager = manager_at(tmp_path, monkeypatch, config)
(tmp_path / TERMS_FILENAME).write_text("one\ntwo\nthree\n", encoding="utf-8")
assert manager.build_initial_prompt() == "one two"
def test_partially_invalid_terms_file_fails_closed(tmp_path: Path, monkeypatch) -> None:
"""Valid lines before corrupt UTF-8 must not reach recognition."""
manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
payload = b"one\ntwo\n" + b"x" * 9000 + b"\xff\nthree\n"
(tmp_path / TERMS_FILENAME).write_bytes(payload)
assert manager.get_terms() == []
assert manager.build_initial_prompt() is None
assert manager.terms_status() == "Terms file is not valid UTF-8."
def test_oversized_terms_file_supplies_its_leading_lines(tmp_path: Path, monkeypatch) -> None:
"""A file beyond the read bound still supplies its complete leading lines."""
manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
line_count = MAX_TERMS_FILE_BYTES // 8 + 2 # 8-byte lines plus a partial tail
(tmp_path / TERMS_FILENAME).write_bytes(
b"".join(f"t{i:06d}\n".encode() for i in range(line_count))
)
terms = manager.get_terms()
assert terms
assert all(re.fullmatch(r"t\d{6}", term) for term in terms)
assert len(terms) == MAX_TERMS_YIELDED
assert manager.build_initial_prompt() is not None
assert "leading lines only" in manager.terms_status()
def test_prompt_on_oversized_file_streams_only_the_needed_prefix(
tmp_path: Path, monkeypatch
) -> None:
"""Prompt assembly stops after max_terms instead of decoding the bound."""
config = FakeConfig({"dictionary": {"enabled": True, "max_words": 3}})
manager = manager_at(tmp_path, monkeypatch, config)
line_count = MAX_TERMS_FILE_BYTES // 8 + 2
(tmp_path / TERMS_FILENAME).write_bytes(
b"".join(f"t{i:06d}\n".encode() for i in range(line_count))
)
assert manager.build_initial_prompt() == "t000000 t000001 t000002"
def test_prompt_on_oversized_file_with_corrupt_prefix_still_fails_closed(
tmp_path: Path, monkeypatch
) -> None:
"""Invalid UTF-8 inside the usable prefix supplies no prompt at all."""
manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
line_count = MAX_TERMS_FILE_BYTES // 8 + 2
(tmp_path / TERMS_FILENAME).write_bytes(
b"one\ntwo\n" + b"\xff\n" + b"".join(f"t{i:06d}\n".encode() for i in range(line_count))
)
assert manager.build_initial_prompt() is None
def test_prompt_on_oversized_file_validates_past_the_term_limit(
tmp_path: Path, monkeypatch
) -> None:
"""Invalid UTF-8 after the collected prompt terms still supplies none."""
config = FakeConfig({"dictionary": {"enabled": True, "max_words": 2}})
manager = manager_at(tmp_path, monkeypatch, config)
line_count = MAX_TERMS_FILE_BYTES // 8 + 2
(tmp_path / TERMS_FILENAME).write_bytes(
b"one\ntwo\nthree\n"
+ b"\xff\n"
+ b"".join(f"t{i:06d}\n".encode() for i in range(line_count))
)
assert manager.build_initial_prompt() is None
assert manager.get_terms() == []
def test_prompt_on_oversized_file_with_single_huge_line_stays_bounded(
tmp_path: Path, monkeypatch
) -> None:
"""A first line longer than the read bound supplies no prompt terms."""
manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
(tmp_path / TERMS_FILENAME).write_bytes(b"x" * (MAX_TERMS_FILE_BYTES + 10) + b"\nvalid\n")
assert manager.build_initial_prompt() is None
def test_add_term_on_oversized_file_stays_visible(tmp_path: Path, monkeypatch) -> None:
"""Prepending past the read window keeps the new term visible while
preserving every scanner-owned line, comment, and blank line."""
manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
line_count = MAX_TERMS_FILE_BYTES // 8 + 2
original = b"# scanner tail\n" + b"".join(f"t{i:06d}\n".encode() for i in range(line_count))
(tmp_path / TERMS_FILENAME).write_bytes(original)
assert manager.add_term("brandnew")
assert (tmp_path / TERMS_FILENAME).read_bytes() == b"brandnew\n" + original
terms = manager.get_terms()
assert terms[0] == "brandnew"
assert len(terms) == MAX_TERMS_YIELDED
def test_add_term_on_yield_capped_file_stays_visible(tmp_path: Path, monkeypatch) -> None:
"""A file at the yield cap but under the byte bound also needs the new
term prepended so Settings and the prompt can see it."""
manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
original = "".join(f"t{i:05d}\n" for i in range(MAX_TERMS_YIELDED))
terms_file = tmp_path / TERMS_FILENAME
terms_file.write_text(original, encoding="utf-8")
assert manager.add_term("brandnew")
assert terms_file.read_text(encoding="utf-8") == f"brandnew\n{original}"
terms = manager.get_terms()
assert terms[0] == "brandnew"
assert len(terms) == MAX_TERMS_YIELDED
def test_oversized_terms_file_without_complete_line_yields_nothing(
tmp_path: Path, monkeypatch
) -> None:
"""A huge file whose first line never ends supplies no terms."""
manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
(tmp_path / TERMS_FILENAME).write_bytes(b"x" * (MAX_TERMS_FILE_BYTES + 10))
assert manager.get_terms() == []
def test_pr_767_dictionary_configuration_keys_and_contract_are_preserved(
tmp_path: Path, monkeypatch
) -> None:
"""Existing enabled, file_path, and max_words settings remain effective."""
configured_path = tmp_path / TERMS_FILENAME
configured_path.write_text("Voculinux\nPyGObject\n", encoding="utf-8")
config = FakeConfig(
{
"dictionary": {
"enabled": True,
"file_path": str(configured_path),
"max_words": 1,
}
}
)
manager = manager_at(tmp_path, monkeypatch, config)
assert TERMS_FILENAME == "dictionary.txt"
assert Path(DEFAULT_TERMS_PATH).name == TERMS_FILENAME
assert manager.terms_enabled()
assert manager.terms_path() == configured_path
assert manager.build_initial_prompt() == "Voculinux"
def test_default_terms_path_follows_xdg_config_home(tmp_path: Path, monkeypatch) -> None:
"""A missing saved path uses the same XDG-aware root as corrections."""
monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path))
manager = CustomDictionaryManager(FakeConfig({"dictionary": {}}))
assert manager.terms_path() == tmp_path / TERMS_FILENAME
assert manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME
def test_legacy_voculinux_terms_root_keeps_corrections_at_config_dir(
tmp_path: Path, monkeypatch
) -> None:
"""Legacy ~/.config/vocalinux terms do not move the corrections file."""
monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path))
legacy_corrections = Path.home() / ".config" / "vocalinux" / CORRECTIONS_FILENAME
expanded_legacy = str(Path.home() / ".config" / "vocalinux" / TERMS_FILENAME)
assert legacy_corrections != tmp_path / CORRECTIONS_FILENAME
tilde_manager = CustomDictionaryManager(
FakeConfig({"dictionary": {"file_path": LEGACY_DEFAULT_TERMS_PATH}})
)
assert tilde_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME
expanded_manager = CustomDictionaryManager(
FakeConfig({"dictionary": {"file_path": expanded_legacy}})
)
assert expanded_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME
def test_explicit_legacy_terms_path_keeps_corrections_at_config_dir(
tmp_path: Path, monkeypatch
) -> None:
"""An explicit leftover pre-XDG terms path cannot relocate corrections."""
monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path))
legacy_root = Path.home() / ".config" / "vocalinux"
expanded_legacy = str(legacy_root / TERMS_FILENAME)
assert legacy_root / CORRECTIONS_FILENAME != tmp_path / CORRECTIONS_FILENAME
tilde_manager = CustomDictionaryManager(
FakeConfig(
{
"dictionary": {
"file_path": LEGACY_DEFAULT_TERMS_PATH,
"file_path_explicit": True,
}
}
)
)
assert tilde_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME
stamped = FakeConfig({"dictionary": {"file_path": expanded_legacy}})
stamped_manager = CustomDictionaryManager(stamped)
stamped_manager.terms_path_text()
assert stamped.get("dictionary", "file_path_explicit") is True
assert stamped_manager.corrections_path() == tmp_path / CORRECTIONS_FILENAME
def test_corrections_fall_back_from_terms_dir_when_primary_missing(
tmp_path: Path, monkeypatch
) -> None:
"""Leftover terms-adjacent corrections are read and copied to the standard file."""
fake_home = tmp_path / "home"
xdg_dir = tmp_path / "xdg-config"
legacy_root = fake_home / ".config" / "vocalinux"
legacy_root.mkdir(parents=True)
xdg_dir.mkdir()
monkeypatch.setenv("HOME", str(fake_home))
monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir))
old_path = Path(LEGACY_DEFAULT_TERMS_PATH).expanduser().parent / CORRECTIONS_FILENAME
payload = {
"version": 1,
"corrections": [{"heard": "super base", "replacement": "Supabase"}],
}
old_path.write_text(json.dumps(payload) + "\n", encoding="utf-8")
manager = CustomDictionaryManager(
FakeConfig(
{
"dictionary": {
"file_path": LEGACY_DEFAULT_TERMS_PATH,
"file_path_explicit": True,
}
}
)
)
primary = xdg_dir / CORRECTIONS_FILENAME
assert manager.corrections_path() == primary
assert primary != old_path
assert not primary.exists()
assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}]
assert primary.is_file()
old_path.unlink()
assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}]
def test_corrections_fallback_survives_copy_failure(tmp_path: Path, monkeypatch) -> None:
"""A failed copy-once still returns leftover terms-adjacent corrections."""
terms_root = tmp_path / "legacy-root"
xdg_dir = tmp_path / "xdg-config"
terms_root.mkdir()
xdg_dir.mkdir()
monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir))
old_path = terms_root / CORRECTIONS_FILENAME
payload = {
"version": 1,
"corrections": [{"heard": "super base", "replacement": "Supabase"}],
}
old_path.write_text(json.dumps(payload) + "\n", encoding="utf-8")
manager = CustomDictionaryManager(
FakeConfig(
{
"dictionary": {
"file_path": str(terms_root / TERMS_FILENAME),
"file_path_explicit": True,
}
}
)
)
primary = xdg_dir / CORRECTIONS_FILENAME
def fail_replace(source: Path, destination: Path) -> None:
"""Simulate a filesystem failure during atomic replacement."""
raise OSError("disk full")
monkeypatch.setattr("vocalinux.custom_dictionary.os.replace", fail_replace)
assert manager.corrections_path() == primary
assert not primary.exists()
assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}]
assert not primary.exists()
def test_corrections_stay_at_config_dir_with_arbitrary_terms_parent(
tmp_path: Path, monkeypatch
) -> None:
"""Corrections ignore the terms file location outside a voculinux directory."""
xdg_dir = tmp_path / "xdg-config"
xdg_dir.mkdir()
monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir))
terms = tmp_path / "my-terms.txt"
manager = CustomDictionaryManager(
FakeConfig(
{
"dictionary": {
"file_path": str(terms),
"file_path_explicit": True,
}
}
)
)
assert manager.corrections_path() == xdg_dir / CORRECTIONS_FILENAME
assert manager.corrections_path() != tmp_path / CORRECTIONS_FILENAME
def test_transient_terms_override_keeps_corrections_at_config_dir(
tmp_path: Path, monkeypatch
) -> None:
"""A --dictionary-file session override must not move the corrections file."""
xdg_dir = tmp_path / "xdg-config"
xdg_dir.mkdir()
monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir))
override = tmp_path / "override.txt"
override.write_text("Override\n", encoding="utf-8")
manager = CustomDictionaryManager(FakeConfig(), str(override))
assert manager.corrections_path() == xdg_dir / CORRECTIONS_FILENAME
assert manager.corrections_path() != tmp_path / CORRECTIONS_FILENAME
def test_transient_terms_override_does_not_import_beside_corrections(
tmp_path: Path, monkeypatch
) -> None:
"""A --dictionary-file session must not import the override's neighbor file."""
xdg_dir = tmp_path / "xdg-config"
override_dir = tmp_path / "elsewhere"
xdg_dir.mkdir()
override_dir.mkdir()
monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(xdg_dir))
override = override_dir / "terms.txt"
override.write_text("Override\n", encoding="utf-8")
payload = {
"version": 1,
"corrections": [{"heard": "super base", "replacement": "Supabase"}],
}
(override_dir / CORRECTIONS_FILENAME).write_text(json.dumps(payload) + "\n", encoding="utf-8")
manager = CustomDictionaryManager(FakeConfig(), str(override))
assert manager.get_corrections() == []
assert manager.get_corrections_for_edit() == []
assert not (xdg_dir / CORRECTIONS_FILENAME).exists()
def test_historical_legacy_looking_path_without_explicit_marker_is_not_migrated(
tmp_path: Path, monkeypatch
) -> None:
"""A leftover pre-XDG path stays put when XDG_CONFIG_HOME points elsewhere."""
monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path))
xdg_path = str(tmp_path / TERMS_FILENAME)
tilde_path = LEGACY_DEFAULT_TERMS_PATH
expanded_legacy = str(Path.home() / ".config" / "vocalinux" / TERMS_FILENAME)
assert expanded_legacy != xdg_path
config = FakeConfig({"dictionary": {"file_path": tilde_path}})
manager = CustomDictionaryManager(config)
assert manager.terms_path_text() == tilde_path
assert manager.terms_path() == Path(tilde_path).expanduser()
assert config.get("dictionary", "file_path") == tilde_path
assert config.get("dictionary", "file_path") != xdg_path
assert config.get("dictionary", "file_path_explicit") is True
failing = FakeConfig(
{"dictionary": {"file_path": tilde_path}},
save_result=False,
)
failing_manager = CustomDictionaryManager(failing)
assert failing_manager.terms_path_text() == tilde_path
assert failing.get("dictionary", "file_path") == tilde_path
assert failing.get("dictionary", "file_path") != xdg_path
assert not failing.get("dictionary", "file_path_explicit", False)
expanded_config = FakeConfig({"dictionary": {"file_path": expanded_legacy}})
expanded_manager = CustomDictionaryManager(expanded_config)
assert expanded_manager.terms_path_text() == expanded_legacy
assert expanded_config.get("dictionary", "file_path") == expanded_legacy
assert expanded_config.get("dictionary", "file_path") != xdg_path
assert expanded_config.get("dictionary", "file_path_explicit") is True
custom = str(tmp_path / "custom-terms.txt")
custom_config = FakeConfig({"dictionary": {"file_path": custom}})
custom_manager = CustomDictionaryManager(custom_config)
assert custom_manager.terms_path_text() == custom
assert custom_config.get("dictionary", "file_path") == custom
def test_explicit_legacy_terms_path_is_not_migrated(tmp_path: Path, monkeypatch) -> None:
"""set_terms_path of the pre-XDG string is kept when XDG points elsewhere."""
monkeypatch.setattr("vocalinux.custom_dictionary.config_dir", lambda: str(tmp_path))
xdg_path = str(tmp_path / TERMS_FILENAME)
expanded_legacy = str(Path.home() / ".config" / "vocalinux" / TERMS_FILENAME)
assert expanded_legacy != xdg_path
tilde_config = FakeConfig({"dictionary": {"file_path": xdg_path}})
tilde_manager = CustomDictionaryManager(tilde_config)
assert tilde_manager.set_terms_path(LEGACY_DEFAULT_TERMS_PATH)
assert tilde_config.get("dictionary", "file_path") == LEGACY_DEFAULT_TERMS_PATH
assert tilde_config.get("dictionary", "file_path_explicit") is True
assert tilde_manager.terms_path_text() == LEGACY_DEFAULT_TERMS_PATH
assert tilde_config.get("dictionary", "file_path") == LEGACY_DEFAULT_TERMS_PATH
expanded_config = FakeConfig({"dictionary": {"file_path": xdg_path}})
expanded_manager = CustomDictionaryManager(expanded_config)
assert expanded_manager.set_terms_path(expanded_legacy)
assert expanded_config.get("dictionary", "file_path") == expanded_legacy
assert expanded_config.get("dictionary", "file_path_explicit") is True
assert expanded_manager.terms_path_text() == expanded_legacy
assert expanded_config.get("dictionary", "file_path") == expanded_legacy
def test_invalid_or_unreadable_configured_terms_path_is_not_persisted(
tmp_path: Path, monkeypatch
) -> None:
"""Invalid paths are harmless and leave the previous configured path intact."""
previous_path = str(tmp_path / TERMS_FILENAME)
config = FakeConfig({"dictionary": {"file_path": previous_path}})
manager = manager_at(tmp_path, monkeypatch, config)
assert not manager.set_terms_path("~voculinux-user-does-not-exist/dictionary.txt")
assert config.get("dictionary", "file_path") == previous_path
assert not manager.set_terms_path(str(tmp_path))
assert config.get("dictionary", "file_path") == previous_path
def test_invalid_saved_terms_path_is_safe_to_read_and_report(tmp_path: Path, monkeypatch) -> None:
"""A pre-existing unresolved path cannot crash recognition or Settings."""
config = FakeConfig(
{
"dictionary": {
"enabled": True,
"file_path": "~voculinux-user-does-not-exist/dictionary.txt",
}
}
)
manager = manager_at(tmp_path, monkeypatch, config)
assert manager.terms_path() is None
assert manager.get_terms() == []
assert manager.build_initial_prompt() is None
assert manager.terms_status() == "Configured terms path is invalid or cannot be expanded."
def test_terms_path_save_failure_preserves_the_previous_configured_path(
tmp_path: Path, monkeypatch
) -> None:
"""A failed config save never claims a new path persisted."""
previous_path = str(tmp_path / TERMS_FILENAME)
config = FakeConfig({"dictionary": {"file_path": previous_path}}, save_result=False)
manager = manager_at(tmp_path, monkeypatch, config)
assert not manager.set_terms_path(str(tmp_path / "new-terms.txt"))
assert config.get("dictionary", "file_path") == previous_path
assert not config.get("dictionary", "file_path_explicit", False)
def test_invalid_transient_terms_path_does_not_crash(tmp_path: Path, monkeypatch) -> None:
"""An unresolved CLI tilde path disables only its session's terms safely."""
manager_at(tmp_path, monkeypatch)
manager = CustomDictionaryManager(FakeConfig(), "~voculinux-user-does-not-exist/dictionary.txt")
assert manager.terms_enabled()
assert manager.terms_path() is None
assert manager.get_terms() == []
assert manager.terms_status() == "Configured terms path is invalid or cannot be expanded."
def test_atomic_save_failure_preserves_existing_terms_file(tmp_path: Path, monkeypatch) -> None:
"""A failed replacement leaves the scanner-visible file intact."""
manager = manager_at(tmp_path, monkeypatch)
terms_file = tmp_path / TERMS_FILENAME
terms_file.write_text("Existing\n", encoding="utf-8")
def fail_replace(source: Path, destination: Path) -> None:
"""Simulate a filesystem failure during atomic replacement."""
raise OSError("disk full")
monkeypatch.setattr("vocalinux.custom_dictionary.os.replace", fail_replace)
assert not manager.save_terms(["New"])
assert terms_file.read_text(encoding="utf-8") == "Existing\n"
def test_file_edits_preserve_existing_permissions(tmp_path: Path, monkeypatch) -> None:
"""Replacing either dictionary file keeps the mode the owner set on it."""
manager = manager_at(tmp_path, monkeypatch)
terms_file = tmp_path / TERMS_FILENAME
terms_file.write_text("Voculinux\n", encoding="utf-8")
terms_file.chmod(0o600)
assert manager.save_terms(["Supabase"])
assert stat.S_IMODE(terms_file.stat().st_mode) == 0o600
corrections_file = tmp_path / CORRECTIONS_FILENAME
corrections_file.write_text('{"version": 1, "corrections": []}\n', encoding="utf-8")
corrections_file.chmod(0o640)
assert manager.save_corrections([{"heard": "a", "replacement": "b"}])
assert stat.S_IMODE(corrections_file.stat().st_mode) == 0o640
def test_dictionary_file_argument_is_available_for_a_session_override(monkeypatch) -> None:
"""The retained CLI option accepts a terms file without changing config."""
monkeypatch.setattr("sys.argv", ["voculinux", "--dictionary-file", "/tmp/terms.txt"])
assert parse_arguments().dictionary_file == "/tmp/terms.txt"
def test_corrections_use_versioned_json_and_live_reload(tmp_path: Path, monkeypatch) -> None:
"""Structured corrections use exact replacement spelling on every read."""
manager = manager_at(tmp_path, monkeypatch)
assert manager.save_corrections([{"heard": "super base", "replacement": "Supabase"}])
stored = json.loads((tmp_path / CORRECTIONS_FILENAME).read_text(encoding="utf-8"))
assert stored == {
"version": 1,
"corrections": [{"heard": "super base", "replacement": "Supabase"}],
}
assert manager.get_corrections_for_edit() == [
{"heard": "super base", "replacement": "Supabase"}
]
assert manager.apply_corrections("super base is useful") == "Supabase is useful"
assert manager.save_corrections([{"heard": "pie object", "replacement": "PyGObject"}])
assert manager.apply_corrections("pie object") == "PyGObject"
def test_corrections_are_literal_longest_first_and_unicode_safe() -> None:
"""Matching is whole-phrase, case-insensitive, and never maps through lower()."""
entries = normalize_corrections(
[
{"heard": "super", "replacement": "short"},
{"heard": "super base", "replacement": "Supabase"},
{"heard": "C++", "replacement": "C plus plus"},
{"heard": "i", "replacement": "letter"},
]
)
assert apply_corrections("SUPER BASE and super", entries) == "Supabase and short"
assert apply_corrections("C++ is not C++foo", entries) == "C plus plus is not C++foo"
assert apply_corrections("İ", entries) == "letter"
assert apply_corrections("e\u0301clair", [{"heard": "e", "replacement": "E"}]) == "éclair"
def test_invalid_corrections_file_fails_closed(tmp_path: Path, monkeypatch) -> None:
"""Malformed JSON never aborts a segment or invents replacements."""
manager = manager_at(tmp_path, monkeypatch)
(tmp_path / CORRECTIONS_FILENAME).write_text("{not json", encoding="utf-8")
assert manager.get_corrections() == []
assert manager.get_corrections_for_edit() is None
assert manager.apply_corrections("super base") == "super base"
def test_partially_invalid_corrections_cannot_be_destructively_edited(
tmp_path: Path, monkeypatch
) -> None:
"""UI write-back is refused when runtime reads had to filter source entries."""
manager = manager_at(tmp_path, monkeypatch)
path = tmp_path / CORRECTIONS_FILENAME
original = {
"version": 1,
"corrections": [
{"heard": "super base", "replacement": "Supabase"},
{"heard": "missing replacement"},
],
}
path.write_text(json.dumps(original), encoding="utf-8")
assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}]
assert manager.get_corrections_for_edit() is None
assert json.loads(path.read_text(encoding="utf-8")) == original
def test_corrections_with_extra_fields_cannot_be_destructively_edited(
tmp_path: Path, monkeypatch
) -> None:
"""Unknown fields still apply at runtime but block a UI rewrite that would drop them."""
manager = manager_at(tmp_path, monkeypatch)
path = tmp_path / CORRECTIONS_FILENAME
original = {
"version": 1,
"corrections": [
{
"heard": "super base",
"replacement": "Supabase",
"note": "scanner annotation",
}
],
}
path.write_text(json.dumps(original), encoding="utf-8")
assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}]
assert manager.apply_corrections("super base is useful") == "Supabase is useful"
assert manager.get_corrections_for_edit() is None
assert json.loads(path.read_text(encoding="utf-8")) == original
def test_corrections_with_top_level_metadata_cannot_be_destructively_edited(
tmp_path: Path, monkeypatch
) -> None:
"""Unknown top-level keys apply at runtime but block a UI rewrite that would drop them."""
manager = manager_at(tmp_path, monkeypatch)
path = tmp_path / CORRECTIONS_FILENAME
original = {
"version": 1,
"metadata": {"author": "scanner"},
"corrections": [{"heard": "super base", "replacement": "Supabase"}],
}
path.write_text(json.dumps(original), encoding="utf-8")
assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}]
assert manager.apply_corrections("super base is useful") == "Supabase is useful"
assert manager.get_corrections_for_edit() is None
assert json.loads(path.read_text(encoding="utf-8")) == original
def test_corrections_with_whitespace_cannot_be_destructively_edited(
tmp_path: Path, monkeypatch
) -> None:
"""Leading or trailing whitespace still applies at runtime but blocks a UI rewrite."""
manager = manager_at(tmp_path, monkeypatch)
path = tmp_path / CORRECTIONS_FILENAME
original = {
"version": 1,
"corrections": [{"heard": " super base ", "replacement": " Supabase "}],
}
path.write_text(json.dumps(original), encoding="utf-8")
assert manager.get_corrections() == [{"heard": "super base", "replacement": "Supabase"}]
assert manager.apply_corrections("super base is useful") == "Supabase is useful"
assert manager.get_corrections_for_edit() is None
assert json.loads(path.read_text(encoding="utf-8")) == original
def test_corrections_with_decomposed_unicode_cannot_be_destructively_edited(
tmp_path: Path, monkeypatch
) -> None:
"""NFD heard values still apply at runtime but block a UI rewrite to NFC."""
manager = manager_at(tmp_path, monkeypatch)
path = tmp_path / CORRECTIONS_FILENAME
original = {
"version": 1,
"corrections": [{"heard": "e\u0301clair", "replacement": "éclair"}],
}
path.write_text(json.dumps(original, ensure_ascii=False), encoding="utf-8")
assert manager.get_corrections() == [{"heard": "éclair", "replacement": "éclair"}]
assert manager.apply_corrections("e\u0301clair") == "éclair"
assert manager.get_corrections_for_edit() is None
assert json.loads(path.read_text(encoding="utf-8")) == original
def test_corrections_with_duplicate_keys_cannot_be_destructively_edited(
tmp_path: Path, monkeypatch
) -> None:
"""Duplicate JSON keys still apply at runtime but block a UI rewrite that would drop them."""
manager = manager_at(tmp_path, monkeypatch)
path = tmp_path / CORRECTIONS_FILENAME
cases = [
(
"{\n"
' "version": 1,\n'
' "corrections": [{"heard": "pie object", "replacement": "PyGObject"}],\n'
' "corrections": [{"heard": "super base", "replacement": "Supabase"}]\n'
"}\n",
[{"heard": "super base", "replacement": "Supabase"}],
),
(
"{\n"
' "version": 1,\n'
' "corrections": ['
'{"heard": "super base", "replacement": "Wrong", "replacement": "Supabase"}'
"]\n}\n",
[{"heard": "super base", "replacement": "Supabase"}],
),
]
for original, expected_runtime in cases:
path.write_text(original, encoding="utf-8")
before = path.read_bytes()
assert manager.get_corrections() == expected_runtime
assert manager.apply_corrections("super base is useful") == "Supabase is useful"
assert manager.get_corrections_for_edit() is None
assert path.read_bytes() == before
def test_invalid_correction_update_does_not_wipe_existing_entry(
tmp_path: Path, monkeypatch
) -> None:
"""A too-long replacement must not delete the previous same-heard correction."""
import importlib
import sys
import vocalinux.ui as ui_pkg
manager = manager_at(tmp_path, monkeypatch)
original = [{"heard": "super base", "replacement": "Supabase"}]
assert manager.save_corrections(original)
corrections_path = tmp_path / CORRECTIONS_FILENAME
before = corrections_path.read_text(encoding="utf-8")
repository = sys.modules["gi.repository"]
bases = {name: type(name, (), {}) for name in ("Box", "ListBoxRow", "Dialog")}
saved = sys.modules.pop("vocalinux.ui.settings_dialog", None)
try:
with patch.object(repository, "Gtk", MagicMock(**bases)):
settings_dialog = importlib.import_module("vocalinux.ui.settings_dialog")
finally:
sys.modules.pop("vocalinux.ui.settings_dialog", None)
if saved is not None:
sys.modules["vocalinux.ui.settings_dialog"] = saved
ui_pkg.settings_dialog = saved
else:
ui_pkg.__dict__.pop("settings_dialog", None)
dialog = Mock()
dialog._initializing = False
dialog._applying_settings = False
dialog._dictionary_available.return_value = True
dialog.dictionary_manager = manager
dialog.dictionary_heard_entry.get_text.return_value = "super base"
dialog.dictionary_replacement_entry.get_text.return_value = "x" * (
MAX_CORRECTION_CHARACTERS + 1
)
settings_dialog.SettingsDialog._on_dictionary_add_correction(dialog, None)
assert corrections_path.read_text(encoding="utf-8") == before
assert manager.get_corrections() == original
dialog.dictionary_heard_entry.set_text.assert_not_called()
dialog.dictionary_replacement_entry.set_text.assert_not_called()
dialog.dictionary_feedback_label.set_text.assert_called_once_with(
"That correction is not valid for the file."
)
def test_nfd_correction_update_replaces_nfc_entry(tmp_path: Path, monkeypatch) -> None:
"""A decomposed heard phrase must update the existing NFC correction."""
import importlib
import sys
import vocalinux.ui as ui_pkg
manager = manager_at(tmp_path, monkeypatch)
original = [{"heard": "éclair", "replacement": "OLD"}]
assert manager.save_corrections(original)
corrections_path = tmp_path / CORRECTIONS_FILENAME
repository = sys.modules["gi.repository"]
bases = {name: type(name, (), {}) for name in ("Box", "ListBoxRow", "Dialog")}
saved = sys.modules.pop("vocalinux.ui.settings_dialog", None)
try:
with patch.object(repository, "Gtk", MagicMock(**bases)):
settings_dialog = importlib.import_module("vocalinux.ui.settings_dialog")
finally:
sys.modules.pop("vocalinux.ui.settings_dialog", None)
if saved is not None:
sys.modules["vocalinux.ui.settings_dialog"] = saved
ui_pkg.settings_dialog = saved
else:
ui_pkg.__dict__.pop("settings_dialog", None)
dialog = Mock()
dialog._initializing = False
dialog._applying_settings = False
dialog._dictionary_available.return_value = True
dialog.dictionary_manager = manager
dialog.dictionary_heard_entry.get_text.return_value = "e\u0301clair"
dialog.dictionary_replacement_entry.get_text.return_value = "NEW"
settings_dialog.SettingsDialog._on_dictionary_add_correction(dialog, None)
stored = json.loads(corrections_path.read_text(encoding="utf-8"))
assert stored == {
"version": 1,
"corrections": [{"heard": "éclair", "replacement": "NEW"}],
}
assert manager.get_corrections() == [{"heard": "éclair", "replacement": "NEW"}]
dialog.dictionary_feedback_label.set_text.assert_called_once_with("Correction updated.")
def test_non_integer_corrections_version_fails_closed(tmp_path: Path, monkeypatch) -> None:
"""Boolean, float, and string versions must not be treated as schema 1."""
manager = manager_at(tmp_path, monkeypatch)
path = tmp_path / CORRECTIONS_FILENAME
cases = (
'{"version": true, "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n',
'{"version": 1.0, "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n',
'{"version": "1", "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n',
'{"version": 1e0, "corrections": [{"heard": "super base", "replacement": "Supabase"}]}\n',
)
for original in cases:
path.write_text(original, encoding="utf-8")
before = path.read_bytes()
assert manager.get_corrections() == []
assert manager.get_corrections_for_edit() is None
assert path.read_bytes() == before
def test_invalid_terms_file_is_ignored_and_explained(tmp_path: Path, monkeypatch) -> None:
"""An invalid scanner file leaves prompt bias empty with a clear status."""
manager = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
(tmp_path / TERMS_FILENAME).write_bytes(b"\xff\xfe")
assert manager.build_initial_prompt() is None
assert manager.terms_status() == "Terms file is not valid UTF-8."
def test_whisper_receives_the_live_custom_terms_prompt(tmp_path: Path, monkeypatch) -> None:
"""Whisper gets the current file contents as initial_prompt at transcription time."""
dictionary = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
(tmp_path / TERMS_FILENAME).write_text("Voculinux\n", encoding="utf-8")
manager = recognition_manager("whisper", dictionary)
manager.model = MagicMock()
manager.model.transcribe.return_value = {"text": "ok"}
manager.model.device = MagicMock()
with patch.dict("sys.modules", {"numpy": mock_numpy(), "torch": MagicMock()}):
assert manager._transcribe_with_whisper([b"\x00\x00"]) == "ok"
assert manager.model.transcribe.call_args.kwargs["initial_prompt"] == "Voculinux"
def test_faster_whisper_receives_the_live_custom_terms_prompt(tmp_path: Path, monkeypatch) -> None:
"""Faster Whisper receives the same live vocabulary bias as OpenAI Whisper."""
dictionary = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
(tmp_path / TERMS_FILENAME).write_text("Voculinux\n", encoding="utf-8")
manager = object.__new__(SpeechRecognitionManager)
manager.dictionary_manager = dictionary
manager._faster_whisper_engine = MagicMock()
manager._faster_whisper_engine.is_ready.return_value = True
manager._faster_whisper_engine.transcribe.return_value = "ok"
assert manager._transcribe_with_faster_whisper([b"\x00\x00"]) == "ok"
manager._faster_whisper_engine.transcribe.assert_called_once_with(
[b"\x00\x00"], language="auto", initial_prompt="Voculinux"
)
def test_whispercpp_composes_and_clears_live_terms_prompt(tmp_path: Path, monkeypatch) -> None:
"""whisper.cpp keeps Advanced context and explicitly clears a stale prompt."""
config = FakeConfig({"dictionary": {"enabled": True}})
dictionary = manager_at(tmp_path, monkeypatch, config)
terms_file = tmp_path / TERMS_FILENAME
terms_file.write_text("Voculinux\n", encoding="utf-8")
manager = recognition_manager("whisper_cpp", dictionary, advanced_prompt="Explicit context")
manager._model_lock = threading.Lock()
manager.model = MagicMock()
manager.model.transcribe.return_value = [MagicMock(text="ok")]
with patch.dict("sys.modules", {"numpy": mock_numpy()}):
assert manager._transcribe_with_whispercpp([b"\x00\x00"]) == "ok"
config.values["dictionary"]["enabled"] = False
assert manager._transcribe_with_whispercpp([b"\x00\x00"]) == "ok"
manager.whispercpp_initial_prompt = ""
assert manager._transcribe_with_whispercpp([b"\x00\x00"]) == "ok"
prompts = [call.kwargs["initial_prompt"] for call in manager.model.transcribe.call_args_list]
assert prompts == ["Explicit context Voculinux", "Explicit context", ""]
def test_vosk_warns_once_for_bias_while_corrections_remain_available(
tmp_path: Path, monkeypatch
) -> None:
"""VOSK rejects vocabulary bias only; its transcript can still be corrected."""
dictionary = manager_at(tmp_path, monkeypatch, FakeConfig({"dictionary": {"enabled": True}}))
(tmp_path / TERMS_FILENAME).write_text("Voculinux\n", encoding="utf-8")
manager = recognition_manager("vosk", dictionary)
manager._get_vosk_model_path = MagicMock(return_value="/models/vosk")
vosk = MagicMock()
with patch.dict("sys.modules", {"vosk": vosk}):
with patch(
"vocalinux.speech_recognition.recognition_manager.os.path.exists", return_value=True
):
with patch(
"vocalinux.speech_recognition.recognition_manager.logger.warning"
) as warning:
manager._init_vosk()
manager._init_vosk()
warning.assert_called_once_with(
"Custom terms are ignored by VOSK; transcript corrections still apply."
)
def test_whispercpp_prompt_composes_advanced_and_terms() -> None:
"""Advanced prompt remains a prefix and an emptied terms file clears the value."""
recognition_manager = object.__new__(SpeechRecognitionManager)
recognition_manager.whispercpp_initial_prompt = "Write product names accurately."
recognition_manager.dictionary_manager = type(
"Dictionary", (), {"build_initial_prompt": lambda self: "Voculinux"}
)()
assert (
recognition_manager._get_whispercpp_prompt() == "Write product names accurately. Voculinux"
)
recognition_manager.dictionary_manager = type(
"Dictionary", (), {"build_initial_prompt": lambda self: None}
)()
assert recognition_manager._get_whispercpp_prompt() == "Write product names accurately."
def test_corrections_run_before_voice_commands() -> None:
"""A correction changes raw text before the command processor can act on it."""
class Recognizer:
"""Minimal VOSK recognizer."""
def AcceptWaveform(self, data: bytes) -> None:
"""Accept a fake audio chunk."""
def FinalResult(self) -> str:
"""Return the raw model text."""
return '{"text": "delete that"}'
recognition_manager = object.__new__(SpeechRecognitionManager)
recognition_manager.engine = "vosk"
recognition_manager.recognizer = Recognizer()
recognition_manager._model_lock = threading.Lock()
recognition_manager._voice_commands_enabled = True
recognition_manager.dictionary_manager = type(
"Dictionary", (), {"apply_corrections": lambda self, text: "delete note"}
)()
received: list[str] = []
recognition_manager.command_processor = type(
"Commands", (), {"process_text": lambda self, text: (text, [])}
)()
recognition_manager.text_callbacks = [received.append]
recognition_manager.action_callbacks = []
recognition_manager.segment_callbacks = []
recognition_manager._process_audio_buffer([b"audio"])
assert received == ["delete note"]
def test_correction_stage_is_shared_by_every_engine() -> None:
"""The common post-transcription stage is available to local and remote engines."""
for engine in (
"vosk",
"whisper",
"whisper_cpp",
"faster_whisper",
"parakeet",
"remote_api",
):
manager = object.__new__(SpeechRecognitionManager)
manager.engine = engine
manager.dictionary_manager = type(
"Dictionary", (), {"apply_corrections": lambda self, text: "Supabase"}
)()
assert manager._apply_dictionary_corrections("super base") == "Supabase"