Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions app/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -144,9 +144,12 @@ async def _desktop_parent_watchdog(parent_pid: int) -> None:
while True:
if not _process_exists(parent_pid):
_log.info("desktop parent process exited; stopping backend")
# Send SIGTERM to ourselves so uvicorn runs its shutdown sequence
# instead of bypassing cleanup with os._exit().
os.kill(os.getpid(), signal.SIGTERM)
# Raise SIGTERM in-process so uvicorn's handler runs its shutdown
# sequence. os.kill(pid, SIGTERM) would be wrong here: on Windows
# it is TerminateProcess -- a hard kill that bypasses cleanup
# (#282). raise_signal triggers the Python-level handler on both
# platforms.
signal.raise_signal(signal.SIGTERM)
return
await asyncio.sleep(1)

Expand Down
50 changes: 50 additions & 0 deletions tests/test_watchdog.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
"""Tests for the desktop parent watchdog's shutdown path (#282)."""

from __future__ import annotations

import signal

import pytest

from app import main as main_mod


@pytest.mark.asyncio
async def test_watchdog_raises_sigterm_in_process(monkeypatch):
"""When the parent dies, the watchdog must raise SIGTERM in-process
(uvicorn's handler runs) -- NOT os.kill, which on Windows is
TerminateProcess and bypasses the shutdown sequence."""
raised: list[int] = []
monkeypatch.setattr(main_mod, "_process_exists", lambda _pid: False)
monkeypatch.setattr(main_mod.signal, "raise_signal", raised.append)

killed: list = []
monkeypatch.setattr(main_mod.os, "kill", lambda *a: killed.append(a))

await main_mod._desktop_parent_watchdog(12345)

assert raised == [signal.SIGTERM]
assert killed == [] # the hard-kill path must be gone


@pytest.mark.asyncio
async def test_watchdog_keeps_waiting_while_parent_alive(monkeypatch):
"""While the parent lives, the watchdog sleeps and loops -- no signal."""
checks: list[int] = []

def alive(pid: int) -> bool:
checks.append(pid)
return len(checks) < 3 # alive twice, then gone

async def instant_sleep(_delay):
return None

raised: list[int] = []
monkeypatch.setattr(main_mod, "_process_exists", alive)
monkeypatch.setattr(main_mod.asyncio, "sleep", instant_sleep)
monkeypatch.setattr(main_mod.signal, "raise_signal", raised.append)

await main_mod._desktop_parent_watchdog(999)

assert len(checks) == 3
assert raised == [signal.SIGTERM]