Remove forced raise_signal calls

tests/test_server.py

@@ -1,56 +1,27 @@
 from __future__ import annotations
 
 import asyncio
-import contextlib
 import signal
 import sys
-from typing import Callable, ContextManager, Generator
 
 import pytest
 
 from uvicorn.config import Config
 from uvicorn.server import Server
 
 
-# asyncio does NOT allow raising in signal handlers, so to detect
-# raised signals raised a mutable `witness` receives the signal
-@contextlib.contextmanager
-def capture_signal_sync(sig: signal.Signals) -> Generator[list[int], None, None]:
-    """Replace `sig` handling with a normal exception via `signal"""
-    witness: list[int] = []
-    original_handler = signal.signal(sig, lambda signum, frame: witness.append(signum))
-    yield witness
-    signal.signal(sig, original_handler)
-
-
-@contextlib.contextmanager
-def capture_signal_async(sig: signal.Signals) -> Generator[list[int], None, None]:  # pragma: py-win32
-    """Replace `sig` handling with a normal exception via `asyncio"""
-    witness: list[int] = []
-    original_handler = signal.getsignal(sig)
-    asyncio.get_running_loop().add_signal_handler(sig, witness.append, sig)
-    yield witness
-    signal.signal(sig, original_handler)
-
-
 async def dummy_app(scope, receive, send):  # pragma: py-win32
     pass
 
 
+signals = [signal.SIGTERM, signal.SIGINT]
 if sys.platform == "win32":
-    signals = [signal.SIGBREAK]
-    signal_captures = [capture_signal_sync]
-else:
-    signals = [signal.SIGTERM, signal.SIGINT]
-    signal_captures = [capture_signal_sync, capture_signal_async]
+    signals += [signal.SIGBREAK]
 
 
 @pytest.mark.anyio
 @pytest.mark.parametrize("exception_signal", signals)
-@pytest.mark.parametrize("capture_signal", signal_captures)
-async def test_server_interrupt(
-    exception_signal: signal.Signals, capture_signal: Callable[[signal.Signals], ContextManager[None]]
-):  # pragma: py-win32
+async def test_server_interrupt(exception_signal: signal.Signals):  # pragma: py-win32
     """Test interrupting a Server that is run explicitly inside asyncio"""
 
     async def interrupt_running(srv: Server):
@@ -59,9 +30,34 @@ async def interrupt_running(srv: Server):
         signal.raise_signal(exception_signal)
 
     server = Server(Config(app=dummy_app, loop="asyncio"))
+    original_handler = signal.getsignal(exception_signal)
+
     asyncio.create_task(interrupt_running(server))
-    with capture_signal(exception_signal) as witness:
-        await server.serve()
-    assert witness
+    await server.serve()
+
+    assert signal.getsignal(exception_signal) == original_handler
     # set by the server's graceful exit handler
     assert server.should_exit
+
+
+@pytest.mark.anyio
+async def test_server_interrupt_force_exit_on_double_sigint():  # pragma: py-win32
+    """Test interrupting a Server  on double SIGINT that is run explicitly inside asyncio"""
+
+    sigint = signal.SIGINT
+
+    async def interrupt_running(srv: Server):
+        while not srv.started:
+            await asyncio.sleep(0.01)
+        signal.raise_signal(sigint)
+        signal.raise_signal(sigint)
+
+    server = Server(Config(app=dummy_app, loop="asyncio"))
+    original_handler = signal.getsignal(sigint)
+
+    asyncio.create_task(interrupt_running(server))
+    await server.serve()
+    assert server.should_exit
+    assert server.force_exit
+
+    assert signal.getsignal(sigint) == original_handler

uvicorn/server.py

@@ -58,8 +58,6 @@ def __init__(self, config: Config) -> None:
         self.force_exit = False
         self.last_notified = 0.0
 
-        self._captured_signals: list[int] = []
-
     def run(self, sockets: list[socket.socket] | None = None) -> None:
         self.config.setup_event_loop()
         return asyncio.run(self.serve(sockets=sockets))
@@ -321,14 +319,8 @@ def capture_signals(self) -> Generator[None, None, None]:
         finally:
             for sig, handler in original_handlers.items():
                 signal.signal(sig, handler)
-        # If we did gracefully shut down due to a signal, try to
-        # trigger the expected behaviour now; multiple signals would be
-        # done LIFO, see https://stackoverflow.com/questions/48434964
-        for captured_signal in reversed(self._captured_signals):
-            signal.raise_signal(captured_signal)
 
     def handle_exit(self, sig: int, frame: FrameType | None) -> None:
-        self._captured_signals.append(sig)
         if self.should_exit and sig == signal.SIGINT:
             self.force_exit = True
         else: