fix: add SIGKILL fallback when SIGTERM does not stop the alarm

This commit is contained in:
2026-06-18 17:17:56 +02:00
parent 4d9f08f21f
commit 0472de5a5d
2 changed files with 40 additions and 3 deletions
+10
View File
@@ -2,6 +2,7 @@ import strawberry
from typing import List, Optional from typing import List, Optional
import os import os
import sys import sys
import time
import subprocess import subprocess
import signal import signal
from pathlib import Path from pathlib import Path
@@ -138,6 +139,15 @@ class Mutation:
with open(PID_FILE) as f: with open(PID_FILE) as f:
pid = int(f.read().strip()) pid = int(f.read().strip())
os.kill(pid, signal.SIGTERM) os.kill(pid, signal.SIGTERM)
# Give the process a short grace period, then escalate to SIGKILL.
for _ in range(20):
time.sleep(0.1)
try:
os.kill(pid, 0)
except ProcessLookupError:
break
else:
os.kill(pid, signal.SIGKILL)
if os.path.exists(PID_FILE): if os.path.exists(PID_FILE):
os.remove(PID_FILE) os.remove(PID_FILE)
return True return True
+30 -3
View File
@@ -2,6 +2,7 @@ from unittest.mock import patch, MagicMock
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
import tempfile import tempfile
import os import os
import signal
# We need to set the environment variable before importing the app # We need to set the environment variable before importing the app
os.environ["API_KEY"] = "test-secret" os.environ["API_KEY"] = "test-secret"
@@ -295,14 +296,38 @@ def test_start_ringing_ignores_when_already_ringing():
mock_popen.assert_not_called() mock_popen.assert_not_called()
def test_stop_ringing_kills_process_when_ringing(): def test_stop_ringing_sends_sigterm_when_ringing():
"""stopRinging returns True and kills the process when ringing.""" """stopRinging sends SIGTERM and cleans up the PID file."""
from api.schema import Mutation from api.schema import Mutation
mutation = Mutation() mutation = Mutation()
with patch("api.schema.is_wecker_ringing", return_value=True), \ with patch("api.schema.is_wecker_ringing", return_value=True), \
patch("api.schema.os.kill") as mock_kill, \ patch("api.schema.os.kill") as mock_kill, \
patch("api.schema.os.path.exists", return_value=True), \
patch("api.schema.os.remove") as mock_remove, \
patch("common.PID_FILE", "/tmp/fake_wecker.pid"), \
patch("builtins.open") as mock_open:
mock_f = mock_open.return_value.__enter__.return_value
mock_f.read.return_value = "1234"
mock_kill.side_effect = [None, ProcessLookupError]
result = mutation.stop_ringing()
assert result is True
mock_kill.assert_any_call(1234, signal.SIGTERM)
mock_remove.assert_called_once()
def test_stop_ringing_falls_back_to_sigkill():
"""stopRinging escalates to SIGKILL if SIGTERM does not terminate the process."""
from api.schema import Mutation
mutation = Mutation()
with patch("api.schema.is_wecker_ringing", return_value=True), \
patch("api.schema.os.kill") as mock_kill, \
patch("api.schema.time.sleep"), \
patch("api.schema.os.path.exists", return_value=True), \
patch("api.schema.os.remove"), \
patch("common.PID_FILE", "/tmp/fake_wecker.pid"), \ patch("common.PID_FILE", "/tmp/fake_wecker.pid"), \
patch("builtins.open") as mock_open: patch("builtins.open") as mock_open:
mock_f = mock_open.return_value.__enter__.return_value mock_f = mock_open.return_value.__enter__.return_value
@@ -310,7 +335,9 @@ def test_stop_ringing_kills_process_when_ringing():
mock_kill.return_value = None mock_kill.return_value = None
result = mutation.stop_ringing() result = mutation.stop_ringing()
assert result is True assert result is True
mock_kill.assert_called_with(1234, 15) signals_sent = [call_args[0][1] for call_args in mock_kill.call_args_list]
assert signal.SIGTERM in signals_sent
assert signal.SIGKILL in signals_sent
def test_stop_ringing_does_nothing_when_not_ringing(): def test_stop_ringing_does_nothing_when_not_ringing():