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 asyncio
import os
import sys
import time
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):
    # This call writes two module globals, and what undoes them is the autouse
    # `_restore_media_tool_paths` fixture in conftest.py -- see it for what went
    # wrong when nothing did. Deliberately not repeated here: one mechanism, one
    # explanation, or the two drift.
    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_windowless_and_inherits_nothing(win_startup, monkeypatch):
    """close_fds: the desktop app starts its node through here because a child
    of Electron inherited Electron's sockets and held them after the app quit."""
    monkeypatch.setattr(plat, "_node_exe", lambda: r"C:\x\meshbay-node.exe")
    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"]
    assert calls["kw"]["creationflags"] & 0x08000000      # CREATE_NO_WINDOW
    assert not calls["kw"]["creationflags"] & 0x00000008  # a console of its own:
    # CTRL_LOGOFF/SHUTDOWN reach install_console_close_handler
    assert calls["kw"]["close_fds"] is True


def test_the_launcher_is_this_very_executable_when_frozen(monkeypatch, tmp_path):
    """PATH came first, and with two copies on it `autostart start` from one
    install launched the other -- found on a machine with an MSIX build on PATH."""
    stray = tmp_path / "stray"
    stray.mkdir()
    (stray / "meshbay-node.exe").write_text("", encoding="utf-8")
    monkeypatch.setenv("PATH", str(stray))
    monkeypatch.setattr(plat.sys, "frozen", True, raising=False)
    frozen = r"C:\Programs\MeshBay\node-runtime\meshbay-node.exe"
    monkeypatch.setattr(plat.sys, "executable", frozen)
    assert plat._node_exe() == frozen


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


# ── Stopping ────────────────────────────────────────────────────────────────
#
# The CTRL_BREAK_EVENT path these tests used to pin was the bug: aimed at a
# node with a console of its own, GenerateConsoleCtrlEvent reached every process
# on the caller's console, so `autostart stop` and `restart-daemon` killed
# themselves. The tests mocked os.kill, so they agreed with it by construction.

def test_autostart_end_force_stops_by_image_name(monkeypatch):
    monkeypatch.setattr(sys, "platform", "win32")
    run_calls = []
    monkeypatch.setattr(plat.subprocess, "run",
                        lambda argv, **kw: run_calls.append(argv))
    plat.autostart_end()
    [argv] = run_calls
    assert argv[:5] == ["taskkill", "/F", "/T", "/IM", "meshbay-node.exe"]


def test_the_forced_stop_spares_the_command_that_runs_it(monkeypatch):
    """The CLI is meshbay-node.exe too: `taskkill /IM meshbay-node.exe` killed
    `autostart stop` and `restart-daemon` themselves -- found on a real install."""
    monkeypatch.setattr(sys, "platform", "win32")
    run_calls = []
    monkeypatch.setattr(plat.subprocess, "run",
                        lambda argv, **kw: run_calls.append(argv))
    plat.autostart_end()
    [argv] = run_calls
    assert f"PID ne {os.getpid()}" in argv and f"PID ne {os.getppid()}" in argv


@pytest.mark.skipif(sys.platform != "win32", reason="taskkill")
def test_the_forced_stop_really_spares_its_caller(tmp_path):
    """For real: a process with the node's image name runs the forced stop, and
    must survive it while another process of that name does not. Under a name of
    its own, or it would also kill the developer's running node."""
    import shutil
    import subprocess
    import textwrap
    import uuid

    image = f"mb-test-{uuid.uuid4().hex[:8]}.exe"
    base = Path(getattr(sys, "_base_executable", sys.executable))
    env = {**os.environ, "PYTHONHOME": sys.base_prefix}

    def renamed_python(where: str) -> Path:
        d = tmp_path / where
        d.mkdir()
        shutil.copy(base, d / image)
        for dll in base.parent.glob("*.dll"):
            shutil.copy(dll, d)
        return d / image

    caller = renamed_python("caller")
    victim = renamed_python("victim")               # a "daemon" with a "transcode"
    child_pid_file = tmp_path / "child.pid"
    src = Path(plat.__file__).resolve().parents[1]
    script = textwrap.dedent(f"""
        import sys; sys.path.insert(0, {str(src)!r})
        from meshbay_node import platform as p
        p.NODE_IMAGE = {image!r}
        p.autostart_end()
        print("survived")
    """)
    other = subprocess.Popen(
        [str(victim), "-c",
         "import subprocess, sys; p = subprocess.Popen(['ping', '-n', '300', '127.0.0.1'],"
         " stdout=subprocess.DEVNULL); open(sys.argv[1], 'w').write(str(p.pid)); p.wait()",
         str(child_pid_file)], env=env)
    child_pid = 0
    try:
        for _ in range(100):
            if child_pid_file.exists() and child_pid_file.read_text(encoding="utf-8"):
                break
            time.sleep(0.1)
        child_pid = int(child_pid_file.read_text(encoding="utf-8"))
        r = subprocess.run([str(caller), "-c", script], capture_output=True, text=True,
                           timeout=60, env=env)
        assert "survived" in r.stdout, (r.returncode, r.stdout, r.stderr)
        assert other.wait(timeout=10) is not None, "the other process must be stopped"
        assert not plat._pid_alive(child_pid), "and its children with it"
    finally:
        if other.poll() is None:
            other.kill()
        if child_pid:
            subprocess.run(["taskkill", "/F", "/PID", str(child_pid)], capture_output=True)


def test_nothing_signals_a_console_group_any_more():
    """Read as code, not text: the docstrings say why CTRL_BREAK went."""
    import ast
    import inspect
    tree = ast.parse(inspect.getsource(plat))
    for node in ast.walk(tree):
        assert not (isinstance(node, ast.Attribute) and node.attr == "CTRL_BREAK_EVENT")
        if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
            assert not (node.func.attr == "kill" and getattr(node.func.value, "id", "") == "os")


def _serve_node(tmp_path, pid):
    """A control API that answers status and shutdown, as the daemon's does."""
    import json
    import threading
    from http.server import BaseHTTPRequestHandler, HTTPServer

    asked = []

    class H(BaseHTTPRequestHandler):
        def _send(self, body):
            data = json.dumps(body).encode()
            self.send_response(200)
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)

        def do_GET(self):  # noqa: N802
            self._send({"version": "x", "status": "running", "pid": pid})

        def do_POST(self):  # noqa: N802
            asked.append(self.path)
            self._send({"stopping": True})

        def log_message(self, *a):
            pass

    server = HTTPServer(("127.0.0.1", 0), H)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    (tmp_path / "ui-token").write_text("tok", encoding="utf-8")
    return server, asked


def test_a_graceful_stop_asks_the_node_and_waits_for_its_process(monkeypatch, tmp_path):
    server, asked = _serve_node(tmp_path, pid=4242)
    alive = iter([True, True, False])
    monkeypatch.setattr(plat, "_pid_alive", lambda pid: next(alive))
    try:
        assert plat.request_graceful_stop(tmp_path, server.server_address[1], timeout=10)
    finally:
        server.shutdown()
    assert asked == ["/api/shutdown?t=tok"]


def test_a_node_that_does_not_exit_is_reported_for_the_caller_to_force(monkeypatch, tmp_path):
    server, _ = _serve_node(tmp_path, pid=4242)
    monkeypatch.setattr(plat, "_pid_alive", lambda pid: True)
    try:
        assert not plat.request_graceful_stop(tmp_path, server.server_address[1], timeout=1)
    finally:
        server.shutdown()


def test_nothing_to_stop_when_no_node_answers(tmp_path):
    (tmp_path / "ui-token").write_text("tok", encoding="utf-8")
    assert not plat.request_graceful_stop(tmp_path, 1, timeout=1)


@pytest.mark.skipif(sys.platform != "win32", reason="tasklist")
def test_pid_alive_sees_this_process_and_not_a_dead_one():
    import os
    assert plat._pid_alive(os.getpid())
    assert not plat._pid_alive(9_999_991)


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.subprocess, "run",
        lambda argv, **kw: calls.append((argv, kw)) 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][0] == "powershell"
    env = calls[0][1]["env"]
    assert env["MESHBAY_SVC_USER"] == "DOMAIN\\user"
    assert env["MESHBAY_SVC_EXE"] == r"C:\x\meshbay-node.exe"


def test_service_install_uses_s4u_not_a_stored_password(monkeypatch):
    """Register-ScheduledTask -LogonType S4U, not schtasks: schtasks only
    infers the logon type from whether /rp is present, and both readings
    broke live on a blank-password account (2026-09-05) -- /rp "" fails
    credential validation, and omitting /rp registers "Interactive only",
    which never runs at boot or on demand. See platform.py's service mode
    comment for the full story."""
    # Service mode is Windows-only and refuses outright anywhere else;
    # every other test in this file says so, these two never did.
    monkeypatch.setattr(sys, "platform", "win32")
    monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user")
    calls = []
    monkeypatch.setattr(
        plat.subprocess, "run",
        lambda argv, **kw: calls.append((argv, kw)) or Mock(returncode=0, stdout="", stderr=""))

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

    script = calls[0][0][-1]
    assert "-LogonType S4U" in script
    assert "New-ScheduledTaskTrigger -AtStartup" in script


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.subprocess, "run",
                        lambda argv, **kw: Mock(returncode=0, stdout="", stderr=""))
    plat.service_install(exe=r"C:\x\meshbay-node.exe")   # no error
    assert not win_startup.exists()


def test_service_install_raises_with_powershells_error_message(monkeypatch):
    # Service mode is Windows-only and refuses outright anywhere else;
    # every other test in this file says so, these two never did.
    monkeypatch.setattr(sys, "platform", "win32")
    monkeypatch.setattr(plat, "_current_user", lambda: "DOMAIN\\user")
    monkeypatch.setattr(
        plat.subprocess, "run",
        lambda argv, **kw: Mock(returncode=1, stdout="", stderr="Access is denied."))

    with pytest.raises(RuntimeError, match="Access is denied"):
        plat.service_install(exe=r"C:\x\meshbay-node.exe")


# ── 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", encoding="utf-8")
    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_load_node_env_sets_names(monkeypatch, tmp_path):
    monkeypatch.setattr(plat, "packaged_default_env", lambda: None)
    (tmp_path / "node.env").write_text(
        "# a comment\n"
        "\n"
        "MESHBAY_TMDB_DEFAULT_TOKEN=eyJloaded\n"
        'QUOTED="value"\n'
    , encoding="utf-8")
    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", encoding="utf-8")
    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(monkeypatch, tmp_path):
    monkeypatch.setattr(plat, "packaged_default_env", lambda: None)
    assert plat.load_node_env(tmp_path) == 0


def test_load_node_env_reads_the_packaged_default_without_a_node_env(monkeypatch, tmp_path):
    """A node onboarded by the desktop client has no node.env: nothing ran
    `init` to copy one. The packaged token must reach it anyway."""
    src = tmp_path / "default.env"
    src.write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJpackaged\n", encoding="utf-8")
    monkeypatch.setattr(plat, "packaged_default_env", lambda: src)
    monkeypatch.delenv("MESHBAY_TMDB_DEFAULT_TOKEN", raising=False)
    cfg = tmp_path / "config"
    cfg.mkdir()

    assert plat.load_node_env(cfg) == 1
    assert os.environ["MESHBAY_TMDB_DEFAULT_TOKEN"] == "eyJpackaged"


def test_load_node_env_prefers_the_operator_node_env(monkeypatch, tmp_path):
    src = tmp_path / "default.env"
    src.write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJpackaged\n", encoding="utf-8")
    monkeypatch.setattr(plat, "packaged_default_env", lambda: src)
    monkeypatch.delenv("MESHBAY_TMDB_DEFAULT_TOKEN", raising=False)
    cfg = tmp_path / "config"
    cfg.mkdir()
    (cfg / "node.env").write_text("MESHBAY_TMDB_DEFAULT_TOKEN=eyJoperator\n", encoding="utf-8")

    plat.load_node_env(cfg)
    assert os.environ["MESHBAY_TMDB_DEFAULT_TOKEN"] == "eyJoperator"