diff --git a/bmspy/server.py b/bmspy/server.py index 48e1b95..d676ef7 100755 --- a/bmspy/server.py +++ b/bmspy/server.py @@ -15,6 +15,7 @@ import re import struct import serial from dataclasses import asdict as dataclass_asdict, dataclass +from types import FrameType from typing import Any, NoReturn from bmspy.utilities import debugger @@ -52,7 +53,7 @@ class DeviceState: connected_clients = list() -def signalHandler() -> NoReturn: +def signalHandler(signum: int, frame: FrameType | None) -> NoReturn: raise SystemExit("terminating") diff --git a/tests/test_server.py b/tests/test_server.py index 0f9acf7..8e776ea 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,6 +1,9 @@ import json +import signal import socket as socket_module import struct +import subprocess +import sys import threading import pytest @@ -166,12 +169,40 @@ class TestSignalHandler: def test_raises_system_exit(self): from bmspy.server import signalHandler with pytest.raises(SystemExit): - signalHandler() + signalHandler(signal.SIGTERM, None) def test_exit_message_contains_terminating(self): from bmspy.server import signalHandler with pytest.raises(SystemExit, match="terminating"): - signalHandler() + signalHandler(signal.SIGTERM, None) + + def test_sigterm_during_accept_cleans_up_without_traceback(self, tmp_path): + socket_path = tmp_path / "bms.sock" + script = """ +import atexit +import os +import signal +import socket +import sys +import threading +from bmspy.server import signalHandler, socket_cleanup + +signal.signal(signal.SIGTERM, signalHandler) +listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) +listener.bind(sys.argv[1]) +listener.listen(1) +atexit.register(socket_cleanup, sys.argv[1]) +threading.Timer(0.1, os.kill, args=(os.getpid(), signal.SIGTERM)).start() +listener.accept() +""" + result = subprocess.run( + [sys.executable, "-c", script, str(socket_path)], + capture_output=True, text=True, timeout=10, + ) + # Preserve the existing SystemExit("terminating") behavior. + assert result.returncode == 1 + assert result.stderr.strip() == "terminating" + assert not socket_path.exists() # ---------------------------------------------------------------------------