voculinux/tests/test_legacy_import.py

"""Ticket #18: one-time legacy Vocalinux import into the Voculinux locations.

Seams under test (all offline, no hardware):
- A fresh run with a legacy install present imports ``config.json``, the
  custom dictionary, and the portal restore token exactly once and logs it;
  a second run imports nothing.
- A fresh run with an existing new configuration never imports, merges, or
  writes to the legacy location.
- A fresh run with no legacy install starts from defaults (nothing imported).
- Models are not migrated.
"""

import json
import os
from pathlib import Path

import pytest

from vocalinux import legacy_import
from vocalinux.legacy_import import import_legacy_once


def _isolate(monkeypatch, tmp_path: Path) -> tuple[Path, Path]:
    """Point XDG roots at scratch dirs; return (config_root, data_root)."""
    config_root = tmp_path / "config"
    data_root = tmp_path / "data"
    monkeypatch.setenv("XDG_CONFIG_HOME", str(config_root))
    monkeypatch.setenv("XDG_DATA_HOME", str(data_root))
    return config_root, data_root


def _seed_legacy(config_root: Path, data_root: Path) -> dict[str, Path]:
    """Create a legacy install's payload; return the created files."""
    legacy_config = config_root / "vocalinux"
    legacy_data = data_root / "vocalinux"
    legacy_config.mkdir(parents=True)
    legacy_data.mkdir(parents=True)
    created = {}
    for name, content in (
        ("config.json", '{"general": {"first_run": false}}'),
        ("dictionary.txt", "Supabase\n"),
        ("custom-dictionary-corrections.json", json.dumps({"version": 1, "corrections": []})),
    ):
        target = legacy_config / name
        target.write_text(content, encoding="utf-8")
        created[name] = target
    token = legacy_data / "remote_desktop_restore_token"
    token.write_text("legacy-token\n", encoding="utf-8")
    created["remote_desktop_restore_token"] = token
    return created


def test_fresh_run_with_legacy_install_imports_once(monkeypatch, tmp_path, caplog) -> None:
    """Legacy config, dictionary, and restore token copy over exactly once."""
    config_root, data_root = _isolate(monkeypatch, tmp_path)
    _seed_legacy(config_root, data_root)

    with caplog.at_level("INFO", logger="vocalinux.legacy_import"):
        imported = import_legacy_once()

    assert len(imported) == 4
    assert (
        config_root / "voculinux" / "config.json"
    ).read_text() == '{"general": {"first_run": false}}'
    assert (config_root / "voculinux" / "dictionary.txt").read_text() == "Supabase\n"
    assert json.loads(
        (config_root / "voculinux" / "custom-dictionary-corrections.json").read_text()
    ) == {
        "version": 1,
        "corrections": [],
    }
    assert (
        data_root / "voculinux" / "remote_desktop_restore_token"
    ).read_text() == "legacy-token\n"
    assert any("Legacy Vocalinux import complete" in r.message for r in caplog.records)

    # A second run imports nothing.
    assert import_legacy_once() == []


def test_existing_new_config_never_imports_or_touches_legacy(monkeypatch, tmp_path) -> None:
    """An existing new configuration always wins: no merge, no write-back."""
    config_root, data_root = _isolate(monkeypatch, tmp_path)
    seeded = _seed_legacy(config_root, data_root)
    before = {name: path.read_bytes() for name, path in seeded.items()}

    new_config = config_root / "voculinux"
    new_config.mkdir(parents=True)
    (new_config / "config.json").write_text('{"general": {"first_run": true}}', encoding="utf-8")

    assert import_legacy_once() == []
    assert (new_config / "config.json").read_text() == '{"general": {"first_run": true}}'
    assert not (new_config / "dictionary.txt").exists()
    assert not (data_root / "voculinux").exists()
    for name, path in seeded.items():
        assert path.read_bytes() == before[name], name


def test_no_legacy_install_starts_from_defaults(monkeypatch, tmp_path) -> None:
    """No legacy directories means nothing is imported."""
    _isolate(monkeypatch, tmp_path)
    assert import_legacy_once() == []


def test_models_are_not_migrated(monkeypatch, tmp_path) -> None:
    """A legacy models directory is left alone; rung files re-download."""
    config_root, data_root = _isolate(monkeypatch, tmp_path)
    _seed_legacy(config_root, data_root)
    legacy_models = data_root / "vocalinux" / "models"
    legacy_models.mkdir(parents=True)
    (legacy_models / "ggml-large-v3-turbo-q8_0.bin").write_bytes(b"fake-model")

    imported = import_legacy_once()

    assert not (data_root / "voculinux" / "models").exists()
    assert {Path(p).name for p in imported} == {
        "config.json",
        "dictionary.txt",
        "custom-dictionary-corrections.json",
        "remote_desktop_restore_token",
    }


def test_partial_legacy_payload_imports_what_exists(monkeypatch, tmp_path) -> None:
    """A legacy install with only a config still imports just that file."""
    config_root, _data_root = _isolate(monkeypatch, tmp_path)
    legacy_config = config_root / "vocalinux"
    legacy_config.mkdir(parents=True)
    (legacy_config / "config.json").write_text("{}", encoding="utf-8")

    imported = import_legacy_once()

    assert imported == [str(config_root / "voculinux" / "config.json")]
    assert import_legacy_once() == []


def test_new_dictionary_never_overwritten_by_legacy(monkeypatch, tmp_path) -> None:
    """A new dictionary file is never clobbered even when config is absent."""
    config_root, data_root = _isolate(monkeypatch, tmp_path)
    _seed_legacy(config_root, data_root)
    new_config = config_root / "voculinux"
    new_config.mkdir(parents=True)
    (new_config / "dictionary.txt").write_text("Mine\n", encoding="utf-8")

    import_legacy_once()

    # No new config.json exists, so the import ran — but the existing new
    # dictionary file must survive untouched.
    assert (new_config / "dictionary.txt").read_text() == "Mine\n"
    assert legacy_import.CONFIG_FILENAME == "config.json"


#: A fixed old timestamp, so any write to the legacy file moves its mtime visibly.
_OLD_MTIME_NS = 1_000_000_000_000_000_000


def _import_with_stored_dictionary_path(
    monkeypatch, config_root: Path, data_root: Path, stored_path: str
) -> Path:
    """Seed a legacy install whose config stores ``stored_path``, then import it.

    Returns the legacy dictionary file. The ConfigManager paths are pointed at
    the copied configuration, so nothing outside the scratch directories is read
    or written.
    """
    seeded = _seed_legacy(config_root, data_root)
    seeded["config.json"].write_text(
        json.dumps({"dictionary": {"enabled": True, "file_path": stored_path}}),
        encoding="utf-8",
    )
    import_legacy_once()
    _point_config_manager_at(monkeypatch, config_root / "voculinux")
    return seeded["dictionary.txt"]


def _point_config_manager_at(monkeypatch, new_config_dir: Path) -> None:
    """Make ConfigManager read and write only the given (scratch) configuration dir."""
    from vocalinux.ui import config_manager

    monkeypatch.setattr(config_manager, "CONFIG_DIR", str(new_config_dir))
    monkeypatch.setattr(config_manager, "CONFIG_FILE", str(new_config_dir / "config.json"))


def _add_term_through_migrated_config(monkeypatch, new_config_dir: Path, term: str) -> None:
    """Add a dictionary term exactly as the app does, using the migrated configuration."""
    _point_config_manager_at(monkeypatch, new_config_dir)
    from vocalinux.custom_dictionary import CustomDictionaryManager
    from vocalinux.ui.config_manager import ConfigManager

    assert CustomDictionaryManager(ConfigManager()).add_term(term) is True


@pytest.mark.parametrize("form", ["absolute-legacy", "tilde-legacy"])
def test_migrated_dictionary_edit_never_writes_the_legacy_file(monkeypatch, tmp_path, form) -> None:
    """The copied config points the dictionary at the copy, so an edit leaves legacy alone."""
    if form == "tilde-legacy":
        # The tilde form names the legacy file only when XDG_CONFIG_HOME is ~/.config.
        home = tmp_path / "home"
        monkeypatch.setenv("HOME", str(home))
        config_root = home / ".config"
        monkeypatch.setenv("XDG_CONFIG_HOME", str(config_root))
        data_root = tmp_path / "data"
        monkeypatch.setenv("XDG_DATA_HOME", str(data_root))
        stored = "~/.config/vocalinux/dictionary.txt"
    else:
        config_root, data_root = _isolate(monkeypatch, tmp_path)
        stored = str(config_root / "vocalinux" / "dictionary.txt")
    legacy_terms = _import_with_stored_dictionary_path(monkeypatch, config_root, data_root, stored)
    os.utime(legacy_terms, ns=(_OLD_MTIME_NS, _OLD_MTIME_NS))
    legacy_bytes = legacy_terms.read_bytes()
    legacy_mtime = legacy_terms.stat().st_mtime_ns

    from vocalinux.custom_dictionary import CustomDictionaryManager
    from vocalinux.ui.config_manager import ConfigManager

    dictionary = CustomDictionaryManager(ConfigManager())
    assert dictionary.add_term("Kubernetes") is True

    assert legacy_terms.read_bytes() == legacy_bytes
    assert legacy_terms.stat().st_mtime_ns == legacy_mtime
    new_terms = config_root / "voculinux" / "dictionary.txt"
    assert "Kubernetes" in new_terms.read_text(encoding="utf-8")


def test_dictionary_path_outside_the_legacy_directory_is_kept(monkeypatch, tmp_path) -> None:
    """A path the user chose outside the legacy directory is not rewritten."""
    chosen = tmp_path / "shared" / "terms.txt"
    chosen.parent.mkdir(parents=True)
    chosen.write_text("Chosen\n", encoding="utf-8")
    config_root, data_root = _isolate(monkeypatch, tmp_path)
    _import_with_stored_dictionary_path(monkeypatch, config_root, data_root, str(chosen))

    migrated = json.loads((config_root / "voculinux" / "config.json").read_text(encoding="utf-8"))
    assert migrated["dictionary"]["file_path"] == str(chosen)
    assert chosen.read_text(encoding="utf-8") == "Chosen\n"


def test_absent_legacy_dictionary_edit_lands_in_the_new_file(monkeypatch, tmp_path) -> None:
    """The default dictionary path was saved but the file was never created."""
    config_root, _data_root = _isolate(monkeypatch, tmp_path)
    legacy_config = config_root / "vocalinux"
    legacy_config.mkdir(parents=True)
    legacy_terms = legacy_config / "dictionary.txt"
    (legacy_config / "config.json").write_text(
        json.dumps({"dictionary": {"enabled": True, "file_path": str(legacy_terms)}}),
        encoding="utf-8",
    )
    import_legacy_once()

    _add_term_through_migrated_config(monkeypatch, config_root / "voculinux", "Kubernetes")

    assert not legacy_terms.exists()
    assert "Kubernetes" in (config_root / "voculinux" / "dictionary.txt").read_text(
        encoding="utf-8"
    )


def test_existing_new_dictionary_takes_the_edit_and_legacy_stays_put(monkeypatch, tmp_path) -> None:
    """A new dictionary already present wins; the legacy file is never written."""
    config_root, data_root = _isolate(monkeypatch, tmp_path)
    seeded = _seed_legacy(config_root, data_root)
    legacy_terms = seeded["dictionary.txt"]
    seeded["config.json"].write_text(
        json.dumps({"dictionary": {"enabled": True, "file_path": str(legacy_terms)}}),
        encoding="utf-8",
    )
    new_config_dir = config_root / "voculinux"
    new_config_dir.mkdir(parents=True)
    (new_config_dir / "dictionary.txt").write_text("Mine\n", encoding="utf-8")
    os.utime(legacy_terms, ns=(_OLD_MTIME_NS, _OLD_MTIME_NS))
    legacy_bytes = legacy_terms.read_bytes()
    legacy_mtime = legacy_terms.stat().st_mtime_ns
    import_legacy_once()

    _add_term_through_migrated_config(monkeypatch, new_config_dir, "Kubernetes")

    assert legacy_terms.read_bytes() == legacy_bytes
    assert legacy_terms.stat().st_mtime_ns == legacy_mtime
    assert (new_config_dir / "dictionary.txt").read_text(encoding="utf-8") == "Mine\nKubernetes\n"


# --- #52: leftover upstream launchers -----------------------------------------
#
# A previous install can leave a user-level desktop entry and two wrapper
# scripts behind. When their venv is gone they start nothing, yet they show up
# as a second app in launchers. Every path here lives under a fake HOME, XDG
# roots and PATH, so the real home is never read.

_LEGACY_DESKTOP = "vocalinux.desktop"  # legacy launcher file name
_LEGACY_LAUNCH = "vocalinux"  # legacy CLI wrapper name
_LEGACY_GUI = "vocalinux-gui"  # legacy GUI wrapper name
_LEGACY_TITLE = "Vocalinux"  # legacy launcher display name
_GI_PREFIX = "env GI_TYPELIB_PATH=/usr/lib/girepository-1.0"


def _isolate_launchers(monkeypatch, tmp_path: Path) -> tuple[Path, Path, Path]:
    """Point HOME, XDG and PATH at scratch dirs; return (home, data_root, path_bin)."""
    home = tmp_path / "home"
    home.mkdir()
    data_root = tmp_path / "data"
    path_bin = tmp_path / "path-bin"
    path_bin.mkdir()
    monkeypatch.setenv("HOME", str(home))
    monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config"))
    monkeypatch.setenv("XDG_DATA_HOME", str(data_root))
    monkeypatch.setenv("PATH", str(path_bin))
    return home, data_root, path_bin


def _bin_dir(home: Path) -> Path:
    return home / ".local" / "bin"


def _write_desktop(data_root: Path, exec_value: str, extra: str = "") -> Path:
    """Write a desktop entry under the fake XDG data home."""
    entry = data_root / "applications" / _LEGACY_DESKTOP
    entry.parent.mkdir(parents=True, exist_ok=True)
    entry.write_text(
        f"[Desktop Entry]\nType=Application\nName={_LEGACY_TITLE}\nExec={exec_value}\n{extra}",
        encoding="utf-8",
    )
    return entry


def _make_venv(root: Path, entry: str) -> Path:
    """Create a live venv with an interpreter and the entry script."""
    venv = root / "venv"
    (venv / "bin").mkdir(parents=True)
    for name in ("python", entry):
        script = venv / "bin" / name
        script.write_text("#!/bin/sh\n", encoding="utf-8")
        script.chmod(0o755)
    return venv


def _write_wrapper(path: Path, venv: Path, entry: str) -> Path:
    """Write a wrapper shaped like the installer's, bound to ``venv``."""
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        "#!/bin/bash\n"
        "export GI_TYPELIB_PATH=/usr/lib/girepository-1.0\n"
        f'PY_SITE_PATHS=$("{venv / "bin" / "python"}" - <<\'PY\' 2>/dev/null\n'
        "PY\n"
        f'EXEC_CMD="{venv / "bin" / entry} \\$*"\n'
        'exec sg input -c "$EXEC_CMD"\n',
        encoding="utf-8",
    )
    path.chmod(0o755)
    return path


def _recorder(calls: list):
    """Return a notifier stand-in that records (title, message, icon)."""

    def _notify(title: str, message: str, icon: str = "") -> None:
        calls.append((title, message, icon))

    return _notify


def test_no_upstream_launchers_means_no_notice(monkeypatch, tmp_path) -> None:
    """A clean machine has nothing to report and writes no marker."""
    _, data_root, _ = _isolate_launchers(monkeypatch, tmp_path)
    calls: list = []

    assert legacy_import.report_stale_upstream_launchers(notifier=_recorder(calls)) == []

    assert calls == []
    assert not (data_root / "voculinux" / legacy_import.STALE_LAUNCHER_MARKER).exists()


def test_missing_exec_target_is_flagged_named_and_shown_once(monkeypatch, tmp_path, caplog) -> None:
    """A desktop entry whose target is gone is logged, notified once, and left untouched."""
    home, data_root, _ = _isolate_launchers(monkeypatch, tmp_path)
    entry = _write_desktop(data_root, str(_bin_dir(home) / _LEGACY_GUI))  # never created
    before = entry.read_bytes()
    calls: list = []

    with caplog.at_level("WARNING", logger="vocalinux.legacy_import"):
        stale = legacy_import.report_stale_upstream_launchers(notifier=_recorder(calls))

    assert stale == [str(entry)]
    assert any(str(entry) in r.getMessage() for r in caplog.records if r.levelname == "WARNING")
    assert len(calls) == 1
    title, message, _icon = calls[0]
    assert title == "Old Vocalinux launcher found"
    assert str(entry) in message
    assert entry.read_bytes() == before

    again: list = []
    assert legacy_import.report_stale_upstream_launchers(notifier=_recorder(again)) == []
    assert again == []


def test_env_prefixed_exec_into_wrapper_with_missing_venv_is_flagged(monkeypatch, tmp_path) -> None:
    """The reported case: a GI_TYPELIB_PATH env prefix, then a wrapper whose venv is gone."""
    home, data_root, _ = _isolate_launchers(monkeypatch, tmp_path)
    gone_venv = tmp_path / "gone" / "venv"
    gui = _write_wrapper(_bin_dir(home) / _LEGACY_GUI, gone_venv, _LEGACY_GUI)
    entry = _write_desktop(data_root, f"{_GI_PREFIX} {gui}")

    assert legacy_import.find_stale_upstream_launchers() == [str(entry), str(gui)]


def test_both_wrappers_are_flagged_when_their_venv_is_gone(monkeypatch, tmp_path) -> None:
    """Without a desktop entry, each stale wrapper is still named."""
    home, _, _ = _isolate_launchers(monkeypatch, tmp_path)
    gone_venv = tmp_path / "gone" / "venv"
    launch = _write_wrapper(_bin_dir(home) / _LEGACY_LAUNCH, gone_venv, _LEGACY_LAUNCH)
    gui = _write_wrapper(_bin_dir(home) / _LEGACY_GUI, gone_venv, _LEGACY_GUI)

    assert legacy_import.find_stale_upstream_launchers() == [str(launch), str(gui)]


def test_working_upstream_install_is_not_flagged(monkeypatch, tmp_path) -> None:
    """A launcher whose target and venv exist is the user's choice, never reported."""
    home, data_root, _ = _isolate_launchers(monkeypatch, tmp_path)
    venv = _make_venv(tmp_path / "live", _LEGACY_GUI)
    gui = _write_wrapper(_bin_dir(home) / _LEGACY_GUI, venv, _LEGACY_GUI)
    _write_desktop(data_root, f"{_GI_PREFIX} {gui}")
    calls: list = []

    assert legacy_import.report_stale_upstream_launchers(notifier=_recorder(calls)) == []

    assert calls == []
    assert not (data_root / "voculinux" / legacy_import.STALE_LAUNCHER_MARKER).exists()


def test_dangling_wrapper_symlink_is_flagged(monkeypatch, tmp_path) -> None:
    """A symlink whose target was removed starts nothing, so it is stale."""
    home, _, _ = _isolate_launchers(monkeypatch, tmp_path)
    link = _bin_dir(home) / _LEGACY_LAUNCH
    link.parent.mkdir(parents=True)
    link.symlink_to(tmp_path / "removed" / "bin" / _LEGACY_LAUNCH)

    assert legacy_import.find_stale_upstream_launchers() == [str(link)]


def test_compiled_binary_is_not_read_as_a_wrapper(monkeypatch, tmp_path) -> None:
    """Only a script with a shebang is a wrapper; a binary that names a dead path is left alone."""
    home, _, _ = _isolate_launchers(monkeypatch, tmp_path)
    binary = _bin_dir(home) / _LEGACY_LAUNCH
    binary.parent.mkdir(parents=True)
    binary.write_bytes(b"\x7fELF\x02\x01\x01" + str(tmp_path / "gone" / "bin" / "python").encode())

    assert legacy_import.find_stale_upstream_launchers() == []


def test_bare_exec_name_is_resolved_through_path(monkeypatch, tmp_path) -> None:
    """A bare Exec name is stale when PATH cannot find it, and fine when a working copy is there."""
    _, data_root, path_bin = _isolate_launchers(monkeypatch, tmp_path)
    entry = _write_desktop(data_root, _LEGACY_LAUNCH)

    assert legacy_import.find_stale_upstream_launchers() == [str(entry)]

    venv = _make_venv(tmp_path / "live", _LEGACY_LAUNCH)
    _write_wrapper(path_bin / _LEGACY_LAUNCH, venv, _LEGACY_LAUNCH)

    assert legacy_import.find_stale_upstream_launchers() == []


@pytest.mark.parametrize("exec_value", ["", "env", '"unterminated /usr/bin/x'])
def test_exec_that_names_no_usable_target_is_not_flagged(monkeypatch, tmp_path, exec_value) -> None:
    """Empty, env-only and unparseable Exec lines give no evidence, so they are not flagged."""
    _, data_root, _ = _isolate_launchers(monkeypatch, tmp_path)
    _write_desktop(data_root, exec_value)

    assert legacy_import.find_stale_upstream_launchers() == []


def test_only_the_main_entry_exec_counts(monkeypatch, tmp_path) -> None:
    """An Exec inside a Desktop Action group is not the launcher's target."""
    _, data_root, _ = _isolate_launchers(monkeypatch, tmp_path)
    live = tmp_path / "live-launcher"
    live.write_text("#!/bin/sh\n", encoding="utf-8")
    _write_desktop(
        data_root,
        str(live),
        extra=f"\n[Desktop Action Old]\nName=Old\nExec={tmp_path / 'gone'}\n",
    )

    assert legacy_import.find_stale_upstream_launchers() == []


def test_existing_marker_skips_the_scan_and_the_notice(monkeypatch, tmp_path) -> None:
    """Once the notice has been recorded, nothing is re-scanned or shown again."""
    home, data_root, _ = _isolate_launchers(monkeypatch, tmp_path)
    _write_desktop(data_root, str(_bin_dir(home) / _LEGACY_GUI))
    marker = data_root / "voculinux" / legacy_import.STALE_LAUNCHER_MARKER
    marker.parent.mkdir(parents=True)
    marker.write_text("shown\n", encoding="utf-8")
    calls: list = []

    assert legacy_import.report_stale_upstream_launchers(notifier=_recorder(calls)) == []
    assert calls == []


def test_empty_xdg_data_home_falls_back_to_the_default(monkeypatch, tmp_path) -> None:
    """An empty XDG_DATA_HOME means ~/.local/share, as the rest of the module reads it."""
    home, _, _ = _isolate_launchers(monkeypatch, tmp_path)
    monkeypatch.setenv("XDG_DATA_HOME", "")
    entry = _write_desktop(home / ".local" / "share", str(_bin_dir(home) / _LEGACY_GUI))

    assert legacy_import.find_stale_upstream_launchers() == [str(entry)]


def test_idle_callback_runs_the_check_once_and_then_stops(monkeypatch) -> None:
    """GLib repeats an idle callback while it returns truthy, so the wrapper must return False."""
    from vocalinux import main as voculinux_main

    calls: list[int] = []
    monkeypatch.setattr(
        legacy_import, "report_stale_upstream_launchers", lambda: calls.append(1) or []
    )

    assert voculinux_main._report_stale_upstream_launchers() is False
    assert calls == [1]


def test_idle_callback_swallows_a_failing_check(monkeypatch) -> None:
    """A failing check is logged and never escapes into the GTK main loop."""
    from vocalinux import main as voculinux_main

    def _boom() -> list[str]:
        raise OSError("disk gone")

    monkeypatch.setattr(legacy_import, "report_stale_upstream_launchers", _boom)

    assert voculinux_main._report_stale_upstream_launchers() is False