ledgrab/server/tests/test_win_shutdown.py

"""Tests for the Windows shutdown guard.

The guard is a no-op outside Windows, so the cross-platform tests just
check that ``start()`` returns ``False`` and never touches Win32.

On Windows we exercise the full WM_QUERYENDSESSION → WM_ENDSESSION
sequence end-to-end by ``SendMessage``-ing the hidden window directly:
the guard should fire the callback synchronously, then block in
WM_ENDSESSION until the completion event is signalled.
"""

from __future__ import annotations

import sys
import threading
import time

import pytest

from ledgrab.utils.win_shutdown import WindowsShutdownGuard

IS_WINDOWS = sys.platform == "win32"


@pytest.mark.skipif(IS_WINDOWS, reason="Non-Windows behaviour")
def test_start_returns_false_off_windows() -> None:
    guard = WindowsShutdownGuard(
        on_shutdown=lambda: None,
        shutdown_complete=threading.Event(),
    )
    assert guard.start() is False


@pytest.mark.skipif(not IS_WINDOWS, reason="Requires Win32 user32")
def test_start_creates_hidden_window() -> None:
    guard = WindowsShutdownGuard(
        on_shutdown=lambda: None,
        shutdown_complete=threading.Event(),
    )
    try:
        assert guard.start() is True
        assert guard._hwnd is not None
    finally:
        guard.stop()


@pytest.mark.skipif(not IS_WINDOWS, reason="Requires Win32 user32")
def test_query_endsession_fires_callback_and_returns_true() -> None:
    import ctypes

    WM_QUERYENDSESSION = 0x0011

    fired: list[str] = []
    complete = threading.Event()
    guard = WindowsShutdownGuard(
        on_shutdown=lambda: fired.append("cb"),
        shutdown_complete=complete,
        wait_seconds=0.5,
    )
    try:
        assert guard.start() is True
        result = ctypes.windll.user32.SendMessageW(guard._hwnd, WM_QUERYENDSESSION, 0, 0)
        assert result == 1, "WM_QUERYENDSESSION must return TRUE so Windows ends the session"
        assert fired == ["cb"], "shutdown callback should fire exactly once on WM_QUERYENDSESSION"
    finally:
        guard.stop()


@pytest.mark.skipif(not IS_WINDOWS, reason="Requires Win32 user32")
def test_query_endsession_is_idempotent() -> None:
    """Two WM_QUERYENDSESSION messages must not run the callback twice."""
    import ctypes

    WM_QUERYENDSESSION = 0x0011

    fired: list[str] = []
    guard = WindowsShutdownGuard(
        on_shutdown=lambda: fired.append("cb"),
        shutdown_complete=threading.Event(),
        wait_seconds=0.5,
    )
    try:
        assert guard.start() is True
        ctypes.windll.user32.SendMessageW(guard._hwnd, WM_QUERYENDSESSION, 0, 0)
        ctypes.windll.user32.SendMessageW(guard._hwnd, WM_QUERYENDSESSION, 0, 0)
        assert fired == ["cb"]
    finally:
        guard.stop()


@pytest.mark.skipif(not IS_WINDOWS, reason="Requires Win32 user32")
def test_endsession_waits_for_completion_event() -> None:
    import ctypes

    WM_ENDSESSION = 0x0016

    complete = threading.Event()
    guard = WindowsShutdownGuard(
        on_shutdown=lambda: None,
        shutdown_complete=complete,
        wait_seconds=2.0,
    )
    try:
        assert guard.start() is True

        def signal_after(delay: float) -> None:
            time.sleep(delay)
            complete.set()

        threading.Thread(target=signal_after, args=(0.2,), daemon=True).start()
        t0 = time.monotonic()
        result = ctypes.windll.user32.SendMessageW(guard._hwnd, WM_ENDSESSION, 1, 0)
        elapsed = time.monotonic() - t0
        assert result == 0
        assert (
            0.15 < elapsed < 1.0
        ), f"WM_ENDSESSION should wait for completion, took {elapsed:.2f}s"
    finally:
        guard.stop()


@pytest.mark.skipif(not IS_WINDOWS, reason="Requires Win32 user32")
def test_endsession_gives_up_after_timeout() -> None:
    """If cleanup never finishes, WM_ENDSESSION must still return — Windows
    will hard-kill us otherwise."""
    import ctypes

    WM_ENDSESSION = 0x0016

    guard = WindowsShutdownGuard(
        on_shutdown=lambda: None,
        shutdown_complete=threading.Event(),  # never set
        wait_seconds=0.3,
    )
    try:
        assert guard.start() is True
        t0 = time.monotonic()
        result = ctypes.windll.user32.SendMessageW(guard._hwnd, WM_ENDSESSION, 1, 0)
        elapsed = time.monotonic() - t0
        assert result == 0
        assert (
            0.25 < elapsed < 1.0
        ), f"WM_ENDSESSION must time out near wait_seconds, took {elapsed:.2f}s"
    finally:
        guard.stop()


@pytest.mark.skipif(not IS_WINDOWS, reason="Requires Win32 user32")
def test_endsession_with_cancel_does_not_wait() -> None:
    """wParam=0 on WM_ENDSESSION means the session was cancelled — no cleanup needed."""
    import ctypes

    WM_ENDSESSION = 0x0016

    guard = WindowsShutdownGuard(
        on_shutdown=lambda: None,
        shutdown_complete=threading.Event(),  # never set
        wait_seconds=5.0,
    )
    try:
        assert guard.start() is True
        t0 = time.monotonic()
        result = ctypes.windll.user32.SendMessageW(guard._hwnd, WM_ENDSESSION, 0, 0)
        elapsed = time.monotonic() - t0
        assert result == 0
        assert elapsed < 0.2, f"WM_ENDSESSION with wParam=0 should be instant, took {elapsed:.2f}s"
    finally:
        guard.stop()