"""Tests for the ``__main__`` entry-point helpers. These cover the bits that aren't exercised by the FastAPI test client — the signal-handler install path and the shutdown-state plumbing — so a regression in the launcher can't silently break the user's "stop targets on PC shutdown" guarantee. """ from __future__ import annotations import signal import threading from types import SimpleNamespace from ledgrab.__main__ import ( _build_server, _install_signal_handlers, _request_shutdown, ) from ledgrab.shutdown_state import GRACEFUL_SHUTDOWN_TIMEOUT def test_request_shutdown_sets_should_exit() -> None: server = SimpleNamespace(should_exit=False) _request_shutdown(server) assert server.should_exit is True def test_build_server_bounds_graceful_shutdown() -> None: """uvicorn defaults ``timeout_graceful_shutdown`` to ``None`` (wait forever). A lingering events WebSocket then blocks ``Server.shutdown()`` from ever reaching the lifespan shutdown, so LED targets are never stopped and the process can't exit. The bound is what guarantees the lifespan shutdown runs — assert we never regress to the unbounded default. """ fake_config = SimpleNamespace( server=SimpleNamespace(host="127.0.0.1", port=8080, log_level="INFO") ) server = _build_server(fake_config) # type: ignore[arg-type] assert server.config.timeout_graceful_shutdown == GRACEFUL_SHUTDOWN_TIMEOUT assert server.config.timeout_graceful_shutdown is not None assert server.config.timeout_graceful_shutdown > 0 def test_graceful_shutdown_timeout_fits_os_budget() -> None: """The graceful wait runs BEFORE the lifespan's own ~16 s shutdown budget, and OS shutdown gives the whole process only ~20 s before a force-kill. Keep the bound small so target restore + DB checkpoint still fit. """ assert isinstance(GRACEFUL_SHUTDOWN_TIMEOUT, int) assert 0 < GRACEFUL_SHUTDOWN_TIMEOUT <= 5 def test_install_signal_handlers_installs_for_known_signals() -> None: """Tray path runs uvicorn on a background thread, so our handlers must actually survive — verify each catchable signal is replaced. """ server = SimpleNamespace(should_exit=False) previous = { name: signal.getsignal(getattr(signal, name)) for name in ("SIGINT",) if hasattr(signal, name) } try: _install_signal_handlers(server) for name in ("SIGINT", "SIGTERM", "SIGBREAK"): sig = getattr(signal, name, None) if sig is None: continue current = signal.getsignal(sig) # The handler is our local closure — its qualname starts with the function it's defined in. assert callable(current), f"{name} handler should be installed" assert getattr(current, "__qualname__", "").startswith( "_install_signal_handlers" ), f"{name} should be replaced by our handler, got {current!r}" finally: # Restore original handlers so the rest of the test suite isn't poisoned. for name, handler in previous.items(): signal.signal(getattr(signal, name), handler) def test_shutdown_state_is_shared_threading_event() -> None: """``__main__`` and ``main`` must share the same Event instance — if a fresh one is constructed on either side, WM_ENDSESSION waits forever. """ from ledgrab.shutdown_state import shutdown_complete as state_event assert isinstance(state_event, threading.Event) # If main.py is importable, confirm it re-exports the same object. try: from ledgrab.main import shutdown_complete as main_event except Exception: return # main.py needs full app state — fine to skip on a bare test run. assert main_event is state_event, "main.py must re-export the same Event, not create a new one"