path: root/packages/meshbay-node/tests/test_platform.py

"""Tests for meshbay_node.platform — the OS-specific paths and tool resolution.

The behaviour is selected on `sys.platform` at call time, so it is exercised
here by monkeypatching that (and `os.environ`) rather than only on the OS the
suite happens to run on.
"""

import os
import asyncio
import sys
from pathlib import Path
from unittest.mock import Mock

import pytest
from meshbay_node import platform as plat

# ── Directories ──────────────────────────────────────────────────────────────

@pytest.mark.parametrize("fn,tail", [
    (plat.config_dir, ()),
    (plat.data_dir, ("data",)),
    (plat.state_dir, ("state",)),
])
def test_windows_dirs_live_under_localappdata(fn, tail, monkeypatch, tmp_path):
    monkeypatch.setattr(sys, "platform", "win32")
    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
    assert fn() == tmp_path.joinpath("meshbay", *tail)


@pytest.mark.parametrize("fn,expected_tail", [
    (plat.config_dir, (".config", "meshbay")),
    (plat.data_dir, (".local", "share", "meshbay")),
    (plat.state_dir, (".local", "state", "meshbay")),
])
def test_posix_dirs_follow_xdg(fn, expected_tail, monkeypatch):
    monkeypatch.setattr(sys, "platform", "linux")
    assert fn() == Path.home().joinpath(*expected_tail)


def test_windows_dirs_fall_back_to_home_without_localappdata(monkeypatch):
    monkeypatch.setattr(sys, "platform", "win32")
    monkeypatch.delenv("LOCALAPPDATA", raising=False)
    assert plat.config_dir() == Path.home() / "meshbay"


# ── File permissions ─────────────────────────────────────────────────────────

def test_chmod_private_is_a_noop_on_windows(monkeypatch):
    monkeypatch.setattr(sys, "platform", "win32")
    fake = Mock(spec=Path)
    plat.chmod_private(fake)
    fake.chmod.assert_not_called()


def test_chmod_private_applies_mode_off_windows(monkeypatch, tmp_path):
    monkeypatch.setattr(sys, "platform", "linux")
    f = tmp_path / "secret"
    f.write_text("x", encoding="utf-8")
    called = {}
    monkeypatch.setattr(type(f), "chmod", lambda self, m: called.setdefault("mode", m))
    plat.chmod_private(f)
    assert called["mode"] == 0o600


# ── Media tools ──────────────────────────────────────────────────────────────

def test_check_media_tools_raises_when_ffmpeg_is_missing(monkeypatch):
    monkeypatch.setattr(plat.shutil, "which", lambda _n: None)
    with pytest.raises(RuntimeError, match="not found in PATH"):
        plat.check_media_tools()


def test_check_media_tools_stores_the_resolved_paths(monkeypatch):
    monkeypatch.setattr(plat.shutil, "which",
                        lambda n: f"/opt/bin/{n}.exe")
    plat.check_media_tools("ffmpeg", "ffprobe")
    assert plat.ffmpeg_cmd() == "/opt/bin/ffmpeg.exe"
    assert plat.ffprobe_cmd() == "/opt/bin/ffprobe.exe"


# ── Event loop ───────────────────────────────────────────────────────────────

def test_configure_event_loop_is_a_noop_off_windows(monkeypatch):
    monkeypatch.setattr(sys, "platform", "linux")
    monkeypatch.setenv("MESHBAY_NODE_EVENT_LOOP", "selector")
    before = asyncio.get_event_loop_policy()
    plat.configure_event_loop()
    assert asyncio.get_event_loop_policy() is before


def test_configure_event_loop_leaves_the_default_loop_alone_without_the_opt_in(monkeypatch):
    monkeypatch.setattr(sys, "platform", "win32")
    monkeypatch.delenv("MESHBAY_NODE_EVENT_LOOP", raising=False)
    before = asyncio.get_event_loop_policy()
    plat.configure_event_loop()
    assert asyncio.get_event_loop_policy() is before


@pytest.mark.skipif(sys.platform != "win32",
                    reason="WindowsSelectorEventLoopPolicy exists only on win32")
def test_configure_event_loop_selector_opt_in(monkeypatch):
    monkeypatch.setenv("MESHBAY_NODE_EVENT_LOOP", "selector")
    plat.configure_event_loop()
    assert isinstance(asyncio.get_event_loop_policy(),
                      asyncio.WindowsSelectorEventLoopPolicy)


# ── Autostart ────────────────────────────────────────────────────────────────
#
# On Windows autostart is a `.vbs` in the per-user Startup folder (a logon task
# would need elevation). Tests point APPDATA at a tmp dir so the real Startup
# folder is never touched, and force sys.platform.

@pytest.fixture
def win_startup(monkeypatch, tmp_path):
    monkeypatch.setattr(sys, "platform", "win32")
    monkeypatch.setenv("APPDATA", str(tmp_path))
    return (tmp_path / "Microsoft" / "Windows" / "Start Menu" / "Programs"
            / "Startup" / "MeshBay Node.vbs")


def test_autostart_supported_tracks_the_platform(monkeypatch):
    monkeypatch.setattr(sys, "platform", "win32")
    assert plat.autostart_supported() is True
    monkeypatch.setattr(sys, "platform", "linux")
    assert plat.autostart_supported() is False


def test_autostart_status_is_inert_off_windows(monkeypatch):
    monkeypatch.setattr(sys, "platform", "linux")
    assert plat.autostart_status() == {"installed": False, "state": ""}


def test_autostart_install_writes_a_hidden_launcher_in_the_startup_folder(win_startup):
    exe = r"C:\Program Files\MeshBay\meshbay-node.exe"
    plat.autostart_install(exe=exe)

    assert win_startup.exists()
    assert plat.autostart_status() == {"installed": True, "state": ""}
    raw = win_startup.read_bytes()
    assert b"\r\n" in raw and b"\n\n" not in raw          # CRLF, no stray LF
    text = raw.decode("utf-8")
    assert f'"{exe}"' in text                             # path is quote-wrapped
    assert "Chr(34)" in text and ", 0, False" in text     # hidden, non-blocking


def test_autostart_remove_deletes_the_launcher_and_is_idempotent(win_startup):
    plat.autostart_install(exe=r"C:\x\meshbay-node.exe")
    assert win_startup.exists()
    plat.autostart_remove()
    assert not win_startup.exists()
    plat.autostart_remove()                               # no error second time


def test_autostart_install_needs_a_locatable_launcher(win_startup, monkeypatch):
    monkeypatch.setattr(plat, "_node_exe", lambda: None)
    with pytest.raises(RuntimeError, match="locate the meshbay-node launcher"):
        plat.autostart_install()


def test_autostart_install_refuses_off_windows(monkeypatch):
    monkeypatch.setattr(sys, "platform", "linux")
    with pytest.raises(RuntimeError, match="Windows-only"):
        plat.autostart_install(exe="/usr/bin/meshbay-node")


def test_autostart_run_launches_the_resolved_exe_windowless_and_records_its_pid(
        win_startup, monkeypatch, tmp_path):
    monkeypatch.setattr(plat, "_node_exe", lambda: r"C:\x\meshbay-node.exe")
    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))     # state_dir() -> pidfile location
    calls = {}

    def fake_popen(argv, **kw):
        calls.update(argv=argv, kw=kw)
        return Mock(pid=4242)

    monkeypatch.setattr(plat.subprocess, "Popen", fake_popen)
    plat.autostart_run()
    assert calls["argv"] == [r"C:\x\meshbay-node.exe"]
    flags = calls["kw"]["creationflags"]
    assert flags & 0x08000000      # CREATE_NO_WINDOW
    assert flags & 0x00000200      # CREATE_NEW_PROCESS_GROUP
    assert not flags & 0x00000008  # not DETACHED_PROCESS -- that has no
                                   # console at all, so CTRL_BREAK_EVENT
                                   # would have nothing to signal
    assert plat._pid_file().read_text(encoding="utf-8") == "4242"


def test_autostart_run_refuses_off_windows(monkeypatch):
    monkeypatch.setattr(sys, "platform", "linux")
    with pytest.raises(RuntimeError, match="Windows-only"):
        plat.autostart_run()


# ── Graceful stop (CTRL_BREAK_EVENT + taskkill fallback) ────────────────────
#
# autostart_end() references signal.CTRL_BREAK_EVENT, which genuinely does not
# exist in the `signal` module off Windows -- monkeypatching sys.platform
# cannot manufacture it, unlike the pure-Python behaviour tested above. Skip
# rather than mock around it, matching test_configure_event_loop_selector_opt_in.

@pytest.mark.skipif(sys.platform != "win32",
                    reason="signal.CTRL_BREAK_EVENT exists only on win32")
def test_autostart_end_stops_gracefully_when_ctrl_break_is_enough(monkeypatch, tmp_path):
    monkeypatch.setattr(sys, "platform", "win32")
    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
    plat._pid_file().parent.mkdir(parents=True, exist_ok=True)
    plat._pid_file().write_text("4242", encoding="utf-8")

    kill_calls = []
    monkeypatch.setattr(plat.os, "kill", lambda pid, sig: kill_calls.append((pid, sig)))
    # Alive (our exe) on the pre-signal check, gone by the first poll after --
    # a plain constant can't tell those two calls apart.
    seen = {"n": 0}

    def fake_check(pid):
        seen["n"] += 1
        return seen["n"] == 1

    monkeypatch.setattr(plat, "_pid_is_meshbay_node", fake_check)
    run_calls = []
    monkeypatch.setattr(plat.subprocess, "run",
                        lambda argv, **kw: run_calls.append(argv))

    plat.autostart_end()

    assert kill_calls == [(4242, plat.signal.CTRL_BREAK_EVENT)]
    assert run_calls == []                       # no taskkill needed
    assert not plat._pid_file().exists()


@pytest.mark.skipif(sys.platform != "win32",
                    reason="signal.CTRL_BREAK_EVENT exists only on win32")
def test_autostart_end_falls_back_to_taskkill_when_the_pid_never_exits(
        monkeypatch, tmp_path):
    monkeypatch.setattr(sys, "platform", "win32")
    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
    plat._pid_file().parent.mkdir(parents=True, exist_ok=True)
    plat._pid_file().write_text("4242", encoding="utf-8")

    monkeypatch.setattr(plat.os, "kill", lambda pid, sig: None)
    monkeypatch.setattr(plat, "_pid_is_meshbay_node", lambda pid: True)  # never exits
    monkeypatch.setattr(plat.time, "sleep", lambda s: None)       # don't really wait
    clock = iter([0.0, 1.0, 6.0])       # deadline = 0.0 + 5.0; third read is past it
    monkeypatch.setattr(plat.time, "monotonic", lambda: next(clock))
    run_calls = []
    monkeypatch.setattr(plat.subprocess, "run",
                        lambda argv, **kw: run_calls.append(argv))

    plat.autostart_end()

    assert run_calls == [["taskkill", "/IM", "meshbay-node.exe", "/F"]]
    assert not plat._pid_file().exists()


def test_autostart_end_falls_back_to_taskkill_without_a_pidfile(monkeypatch, tmp_path):
    """No CTRL_BREAK_EVENT dependency here -- there is no pid to signal, so
    this one runs everywhere, same as the pre-existing behaviour it replaces."""
    monkeypatch.setattr(sys, "platform", "win32")
    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
    run_calls = []
    monkeypatch.setattr(plat.subprocess, "run",
                        lambda argv, **kw: run_calls.append(argv))
    plat.autostart_end()
    assert run_calls == [["taskkill", "/IM", "meshbay-node.exe", "/F"]]


def test_autostart_end_ignores_a_stale_pid_reused_by_another_process(monkeypatch, tmp_path):
    """The recorded pid is alive but is not meshbay-node.exe -- Windows reused
    it after the daemon exited. Must not send CTRL_BREAK_EVENT to whatever
    that is; falls straight to taskkill (by image name, so harmless here)."""
    monkeypatch.setattr(sys, "platform", "win32")
    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
    plat._pid_file().parent.mkdir(parents=True, exist_ok=True)
    plat._pid_file().write_text("4242", encoding="utf-8")

    monkeypatch.setattr(plat, "_pid_is_meshbay_node", lambda pid: False)
    kill_calls = []
    monkeypatch.setattr(plat.os, "kill", lambda pid, sig: kill_calls.append((pid, sig)))
    run_calls = []
    monkeypatch.setattr(plat.subprocess, "run",
                        lambda argv, **kw: run_calls.append(argv))

    plat.autostart_end()

    assert kill_calls == []                      # never signalled the reused pid
    assert run_calls == [["taskkill", "/IM", "meshbay-node.exe", "/F"]]
    assert not plat._pid_file().exists()


def test_autostart_end_is_a_noop_off_windows(monkeypatch):
    monkeypatch.setattr(sys, "platform", "linux")
    run_calls = []
    monkeypatch.setattr(plat.subprocess, "run",
                        lambda argv, **kw: run_calls.append(argv))
    plat.autostart_end()
    assert run_calls == []


# ── Console close / logoff / shutdown handler ───────────────────────────────

def test_install_console_close_handler_is_a_noop_off_windows(monkeypatch):
    monkeypatch.setattr(sys, "platform", "linux")
    loop = Mock()
    stop_event = Mock()
    assert plat.install_console_close_handler(loop, stop_event) is None


@pytest.mark.skipif(sys.platform != "win32",
                    reason="ctypes.windll/wintypes exist only on win32")
def test_install_console_close_handler_registers_and_the_callback_sets_stop_event():
    import asyncio as _asyncio

    loop = _asyncio.new_event_loop()
    try:
        stop_event = _asyncio.Event()
        shutdown_done = plat.install_console_close_handler(loop, stop_event)
        assert shutdown_done is not None
        # Drive the registered handler directly rather than actually closing a
        # console window -- exercises the same code path SetConsoleCtrlHandler
        # would invoke, without needing a live console to close.
        handler = plat._CONSOLE_HANDLER_REFS[-1]
        shutdown_done.set()   # so the handler's bounded wait returns immediately
        # ctypes marshals the WINFUNCTYPE's BOOL restype back as a plain int
        # (1/0), not a Python bool, when called directly like this.
        assert handler(plat.CTRL_CLOSE_EVENT)
        loop.run_until_complete(_asyncio.sleep(0))   # let call_soon_threadsafe land
        assert stop_event.is_set()
        # An event this handler does not own (CTRL_C_EVENT) is left unhandled
        # so Python's own console handler (or the default action) gets it.
        assert not handler(0)
    finally:
        loop.close()


# ── Service mode ─────────────────────────────────────────────────────────────

def test_service_install_removes_the_startup_launcher_first(win_startup, monkeypatch):
    """The two mechanisms are mutually exclusive by design -- both installed
    would start the daemon twice, once at boot and again at sign-in. This is
    the CLI's own front door to that invariant, separate from (but agreeing
    with) the Node page's startup-mode selector."""
    plat.autostart_install(exe=r"C:\x\meshbay-node.exe")
    assert win_startup.exists()

    monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user")
    calls = []
    monkeypatch.setattr(
        plat, "_schtasks",
        lambda *args: calls.append(args) or Mock(returncode=0, stdout="", stderr=""))

    plat.service_install(exe=r"C:\x\meshbay-node.exe")

    assert not win_startup.exists()       # removed as part of service_install
    assert calls and calls[0][0] == "/create"


def test_service_install_tolerates_no_startup_launcher_present(win_startup, monkeypatch):
    assert not win_startup.exists()
    monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user")
    monkeypatch.setattr(plat, "_schtasks",
                        lambda *args: Mock(returncode=0, stdout="", stderr=""))
    plat.service_install(exe=r"C:\x\meshbay-node.exe")   # no error
    assert not win_startup.exists()


# ── Packaged defaults ────────────────────────────────────────────────────────

def test_frozen_build_finds_default_env_beside_the_executable(monkeypatch, tmp_path):
    """Where build-node-runtime.ps1 puts it, alongside ffmpeg."""
    exe = tmp_path / "meshbay-node.exe"
    exe.write_bytes(b"")
    (tmp_path / "default.env").write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJtest\n")
    monkeypatch.setattr(sys, "frozen", True, raising=False)
    monkeypatch.setattr(sys, "executable", str(exe))
    assert plat.packaged_default_env() == tmp_path / "default.env"


def test_source_checkout_has_no_packaged_default(monkeypatch, tmp_path):
    monkeypatch.setattr(sys, "frozen", False, raising=False)
    monkeypatch.setattr(sys, "executable", str(tmp_path / "python"))
    monkeypatch.setattr(plat, "Path", Path)
    # /opt/meshbay-node/share/default.env is absent on a dev machine
    assert plat.packaged_default_env() is None


def test_install_node_env_copies_once(monkeypatch, tmp_path):
    src = tmp_path / "default.env"
    src.write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJfirst\n")
    monkeypatch.setattr(plat, "packaged_default_env", lambda: src)
    cfg = tmp_path / "config"
    cfg.mkdir()

    written = plat.install_node_env(cfg)
    assert written == cfg / "node.env"
    assert "eyJfirst" in written.read_text()


def test_install_node_env_never_overwrites_operator_values(monkeypatch, tmp_path):
    """An existing node.env holds the operator's own token; clobbering it would
    silently downgrade a configured node to the shared default."""
    src = tmp_path / "default.env"
    src.write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJpackaged\n")
    monkeypatch.setattr(plat, "packaged_default_env", lambda: src)
    cfg = tmp_path / "config"
    cfg.mkdir()
    (cfg / "node.env").write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJoperator\n")

    assert plat.install_node_env(cfg) is None
    assert "eyJoperator" in (cfg / "node.env").read_text()


def test_install_node_env_is_a_noop_without_a_package(monkeypatch, tmp_path):
    monkeypatch.setattr(plat, "packaged_default_env", lambda: None)
    assert plat.install_node_env(tmp_path) is None
    assert not (tmp_path / "node.env").exists()


def test_load_node_env_sets_names(monkeypatch, tmp_path):
    (tmp_path / "node.env").write_text(
        "# a comment\n"
        "\n"
        "MESHBAY_TMDB_DEFAULT_TOKEN=eyJloaded\n"
        'QUOTED="value"\n'
    )
    monkeypatch.delenv("MESHBAY_TMDB_DEFAULT_TOKEN", raising=False)
    monkeypatch.delenv("QUOTED", raising=False)
    assert plat.load_node_env(tmp_path) == 2
    assert os.environ["MESHBAY_TMDB_DEFAULT_TOKEN"] == "eyJloaded"
    assert os.environ["QUOTED"] == "value"


def test_load_node_env_does_not_override_the_environment(monkeypatch, tmp_path):
    """systemd may have loaded the same file already, and an operator export
    must win over a packaged default."""
    (tmp_path / "node.env").write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJfromfile\n")
    monkeypatch.setenv("MESHBAY_TMDB_DEFAULT_TOKEN", "eyJfromenv")
    assert plat.load_node_env(tmp_path) == 0
    assert os.environ["MESHBAY_TMDB_DEFAULT_TOKEN"] == "eyJfromenv"


def test_load_node_env_tolerates_a_missing_file(tmp_path):
    assert plat.load_node_env(tmp_path) == 0