1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
"""
Tests for Issue #132: Graceful shutdown when monkey-patch fails.
Issue #132: The monkey-patch of Server.handle_exit doesn't work with
`uvicorn app:app` because uvicorn registers signal handlers BEFORE
importing the app (uvicorn 0.29+).
The fix: _shutdown_watcher() now checks both AppStatus.should_exit
(monkey-patch worked) and uvicorn's Server.should_exit via signal
handler introspection (monkey-patch failed).
Manual Test:
# Terminal 1
uvicorn examples.example:app
# Terminal 2
curl http://localhost:8000/sse
# Terminal 1
Ctrl+C # Should exit gracefully now
"""
import signal
from unittest.mock import MagicMock, patch
import anyio
import pytest
from sse_starlette.sse import (
AppStatus,
_get_shutdown_state,
_get_uvicorn_server,
_shutdown_watcher,
)
class TestUvicornServerIntrospection:
"""Test _get_uvicorn_server() signal handler introspection."""
def test_returns_server_when_handler_is_bound_method(self):
"""Should extract Server from bound method handler."""
mock_server = MagicMock()
mock_server.should_exit = False
with patch("sse_starlette.sse.signal.getsignal") as mock_getsignal:
mock_handler = MagicMock()
mock_handler.__self__ = mock_server
mock_getsignal.return_value = mock_handler
result = _get_uvicorn_server()
assert result is mock_server
def test_returns_none_when_handler_is_sig_dfl(self):
"""Should return None for SIG_DFL (default handler)."""
with patch("sse_starlette.sse.signal.getsignal") as mock_getsignal:
mock_getsignal.return_value = signal.SIG_DFL
result = _get_uvicorn_server()
assert result is None
def test_returns_none_when_handler_is_sig_ign(self):
"""Should return None for SIG_IGN (ignored signal)."""
with patch("sse_starlette.sse.signal.getsignal") as mock_getsignal:
mock_getsignal.return_value = signal.SIG_IGN
result = _get_uvicorn_server()
assert result is None
def test_returns_none_when_handler_lacks_self(self):
"""Should return None when handler is a function (not bound method)."""
with patch("sse_starlette.sse.signal.getsignal") as mock_getsignal:
mock_getsignal.return_value = lambda sig, frame: None
result = _get_uvicorn_server()
assert result is None
def test_returns_none_when_self_lacks_should_exit(self):
"""Should return None when __self__ doesn't have should_exit attribute."""
mock_obj = MagicMock(spec=[]) # No attributes
with patch("sse_starlette.sse.signal.getsignal") as mock_getsignal:
mock_handler = MagicMock()
mock_handler.__self__ = mock_obj
mock_getsignal.return_value = mock_handler
result = _get_uvicorn_server()
assert result is None
def test_returns_none_on_exception(self):
"""Should return None if introspection fails with any exception."""
with patch(
"sse_starlette.sse.signal.getsignal", side_effect=Exception("test error")
):
result = _get_uvicorn_server()
assert result is None
class TestShutdownWatcherDualSource:
"""Test _shutdown_watcher() detects shutdown from both sources."""
@pytest.mark.asyncio
async def test_detects_appstatus_should_exit(self):
"""Should detect when AppStatus.should_exit is set (monkey-patch worked)."""
state = _get_shutdown_state()
event = anyio.Event()
state.events.add(event)
async def set_should_exit():
await anyio.sleep(0.1)
AppStatus.should_exit = True
async with anyio.create_task_group() as tg:
tg.start_soon(_shutdown_watcher)
tg.start_soon(set_should_exit)
# Wait for event to be signaled (with timeout)
with anyio.fail_after(2):
await event.wait()
assert AppStatus.should_exit is True
@pytest.mark.asyncio
async def test_detects_uvicorn_server_should_exit(self):
"""Should detect when uvicorn Server.should_exit is set (Issue #132)."""
# Create a mock uvicorn server
mock_server = MagicMock()
mock_server.should_exit = False
state = _get_shutdown_state()
event = anyio.Event()
state.events.add(event)
async def set_server_should_exit():
await anyio.sleep(0.1)
mock_server.should_exit = True
# Patch _get_uvicorn_server to return our mock
with patch("sse_starlette.sse._get_uvicorn_server", return_value=mock_server):
async with anyio.create_task_group() as tg:
tg.start_soon(_shutdown_watcher)
tg.start_soon(set_server_should_exit)
# Wait for event to be signaled (with timeout)
with anyio.fail_after(2):
await event.wait()
# AppStatus.should_exit should be synced
assert AppStatus.should_exit is True
@pytest.mark.asyncio
async def test_fallback_when_no_uvicorn_server(self):
"""Should work when _get_uvicorn_server returns None."""
state = _get_shutdown_state()
event = anyio.Event()
state.events.add(event)
async def set_should_exit():
await anyio.sleep(0.1)
AppStatus.should_exit = True
# Ensure no uvicorn server is found
with patch("sse_starlette.sse._get_uvicorn_server", return_value=None):
async with anyio.create_task_group() as tg:
tg.start_soon(_shutdown_watcher)
tg.start_soon(set_should_exit)
with anyio.fail_after(2):
await event.wait()
assert event.is_set()
|
