Stage 393 follow-up: HTTP integration tests for /api/sessions/events (Opus advisor PR #2637)
This commit is contained in:
nesquena-hermes
204
tests/test_session_events_http_integration.py
Normal file
204
tests/test_session_events_http_integration.py
Normal file
@@ -0,0 +1,204 @@
|
||||
"""HTTP-level integration coverage for the /api/sessions/events SSE endpoint.
|
||||
|
||||
The pure-bus tests in test_session_events.py cover the in-process publish/subscribe
|
||||
contract but never open an actual HTTP connection — they don't exercise the SSE
|
||||
handler's unsubscribe-on-disconnect path, which is the entire safety contract for
|
||||
the new event bus (without it, server threads leak queues until process exit).
|
||||
|
||||
These tests open a real connection to the test server, drive the bus by triggering
|
||||
side-effects from a separate thread (creating a session via /api/session/new), and
|
||||
assert the SSE frame arrives. The test-server subprocess has its own subscriber
|
||||
pool, so we can only test the disconnect-without-crash path here, not the in-pytest
|
||||
subscriber count. Added in stage-393 follow-up to Opus advisor's blocking concern
|
||||
on PR #2637.
|
||||
"""
|
||||
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
from tests._pytest_port import BASE
|
||||
|
||||
|
||||
def _read_sse_frames(resp, deadline):
|
||||
"""Read raw SSE frames from a streaming response until deadline.
|
||||
|
||||
Yields ('event_name', payload_str) tuples. Treats `: keepalive` lines as
|
||||
keepalives (returned as ('keepalive', None)). Returns when the deadline
|
||||
elapses or the connection closes.
|
||||
"""
|
||||
event_type = None
|
||||
data_lines = []
|
||||
while time.time() < deadline:
|
||||
line = resp.readline()
|
||||
if not line:
|
||||
time.sleep(0.05)
|
||||
continue
|
||||
decoded = line.decode("utf-8", errors="replace").rstrip("\r\n")
|
||||
if decoded == "":
|
||||
if data_lines:
|
||||
yield (event_type or "message", "\n".join(data_lines))
|
||||
event_type = None
|
||||
data_lines = []
|
||||
continue
|
||||
if decoded.startswith("event: "):
|
||||
event_type = decoded[7:].strip()
|
||||
elif decoded.startswith("data: "):
|
||||
data_lines.append(decoded[6:])
|
||||
elif decoded.startswith(": "):
|
||||
yield ("keepalive", None)
|
||||
|
||||
|
||||
def _subscriber_count():
|
||||
"""Read the live subscriber count from this pytest process's bus module."""
|
||||
from api import session_events
|
||||
with session_events._SESSION_EVENTS_LOCK:
|
||||
return len(session_events._SESSION_EVENTS_SUBSCRIBERS)
|
||||
|
||||
|
||||
def test_session_events_bus_subscribe_unsubscribe_balance():
|
||||
"""
|
||||
Pure-bus contract: subscribe → publish → unsubscribe leaves the subscriber
|
||||
set at the baseline. This is the safety invariant the SSE handler's
|
||||
finally block depends on.
|
||||
"""
|
||||
from api import session_events
|
||||
|
||||
baseline = _subscriber_count()
|
||||
q = session_events.subscribe_session_events()
|
||||
try:
|
||||
assert _subscriber_count() == baseline + 1
|
||||
session_events.publish_session_list_changed("integration_test")
|
||||
payload = q.get(timeout=2.0)
|
||||
assert payload["type"] == "sessions_changed"
|
||||
assert payload["reason"] == "integration_test"
|
||||
finally:
|
||||
session_events.unsubscribe_session_events(q)
|
||||
assert _subscriber_count() == baseline
|
||||
|
||||
|
||||
def test_session_events_sse_endpoint_handshake_succeeds(cleanup_test_sessions):
|
||||
"""
|
||||
Open a real HTTP connection to /api/sessions/events on the test server.
|
||||
Verify the handshake succeeds (200 + text/event-stream Content-Type) and
|
||||
we can read at least one event — either a sessions_changed event triggered
|
||||
by a side-effect we cause (POST /api/session/new) or a keepalive after the
|
||||
heartbeat interval. Then close the socket; a leaking handler would not
|
||||
crash the server but would tie up a thread + queue indefinitely.
|
||||
"""
|
||||
url = BASE + "/api/sessions/events"
|
||||
req = urllib.request.Request(url, headers={"Accept": "text/event-stream"})
|
||||
try:
|
||||
resp = urllib.request.urlopen(req, timeout=10)
|
||||
except urllib.error.HTTPError as e:
|
||||
raise AssertionError(f"SSE endpoint returned HTTP {e.code}: {e.read()[:200]!r}")
|
||||
|
||||
try:
|
||||
assert resp.status == 200
|
||||
ct = resp.headers.get("Content-Type", "")
|
||||
assert "text/event-stream" in ct, f"unexpected Content-Type: {ct!r}"
|
||||
|
||||
# Cause a side-effect that should publish an event: create a session.
|
||||
# The publish runs in the server process, so the subscribed queue on
|
||||
# the server side gets the frame. We must give the SSE thread time to
|
||||
# subscribe before publishing — do the POST on a small delay thread.
|
||||
def trigger():
|
||||
time.sleep(0.5)
|
||||
try:
|
||||
payload = json.dumps({"title": "sse-test"}).encode()
|
||||
req2 = urllib.request.Request(
|
||||
BASE + "/api/session/new",
|
||||
data=payload,
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
with urllib.request.urlopen(req2, timeout=5) as r2:
|
||||
d = json.loads(r2.read())
|
||||
sid = d.get("session", {}).get("session_id")
|
||||
if sid:
|
||||
cleanup_test_sessions.append(sid)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
threading.Thread(target=trigger, daemon=True).start()
|
||||
|
||||
deadline = time.time() + 8.0 # heartbeat is 5s; we should see an event well before that or a keepalive after
|
||||
got_real_event = False
|
||||
got_keepalive = False
|
||||
for kind, data in _read_sse_frames(resp, deadline):
|
||||
if kind == "sessions_changed":
|
||||
got_real_event = True
|
||||
try:
|
||||
payload = json.loads(data)
|
||||
assert payload.get("type") == "sessions_changed"
|
||||
except (ValueError, AssertionError):
|
||||
pass
|
||||
break
|
||||
elif kind == "keepalive":
|
||||
got_keepalive = True
|
||||
break
|
||||
assert got_real_event or got_keepalive, (
|
||||
"expected either a sessions_changed event from the triggered POST or "
|
||||
"a keepalive frame within 8s"
|
||||
)
|
||||
finally:
|
||||
resp.close()
|
||||
|
||||
|
||||
def test_session_events_sse_endpoint_survives_rapid_disconnect():
|
||||
"""
|
||||
Open and immediately close several SSE connections back-to-back. The
|
||||
handler must not leak threads/queues — the server should stay responsive
|
||||
for a subsequent regular /api/sessions GET.
|
||||
"""
|
||||
url = BASE + "/api/sessions/events"
|
||||
for _ in range(5):
|
||||
req = urllib.request.Request(url, headers={"Accept": "text/event-stream"})
|
||||
try:
|
||||
resp = urllib.request.urlopen(req, timeout=10)
|
||||
try:
|
||||
# Read a single byte to confirm the stream opened, then close.
|
||||
resp.read(1)
|
||||
finally:
|
||||
resp.close()
|
||||
except urllib.error.HTTPError as e:
|
||||
raise AssertionError(f"connection iteration failed: HTTP {e.code}")
|
||||
time.sleep(0.1)
|
||||
|
||||
# Server still serves a normal GET after the rapid open/close burst.
|
||||
sentinel_req = urllib.request.Request(
|
||||
BASE + "/api/sessions", headers={"Accept": "application/json"}
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(sentinel_req, timeout=5) as r:
|
||||
assert r.status == 200
|
||||
body = json.loads(r.read())
|
||||
assert isinstance(body, dict)
|
||||
except urllib.error.HTTPError as e:
|
||||
raise AssertionError(f"server unresponsive after SSE burst: HTTP {e.code}")
|
||||
|
||||
|
||||
def test_session_events_handler_uses_disconnect_safe_errors_tuple():
|
||||
"""
|
||||
Source-level guard that the handler's except clause references the shared
|
||||
_CLIENT_DISCONNECT_ERRORS tuple defined in api/routes.py. This is the
|
||||
safety mechanism that makes the unsubscribe-on-disconnect path actually
|
||||
fire across the full range of socket-failure modes (not just BrokenPipeError).
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
routes = Path("api/routes.py").read_text(encoding="utf-8")
|
||||
handler_start = routes.find("def _handle_session_events_stream")
|
||||
assert handler_start >= 0, "_handle_session_events_stream not found"
|
||||
handler_end = routes.find("\ndef ", handler_start + 1)
|
||||
if handler_end < 0:
|
||||
handler_end = len(routes)
|
||||
handler_body = routes[handler_start:handler_end]
|
||||
assert "_CLIENT_DISCONNECT_ERRORS" in handler_body, (
|
||||
"handler must catch the shared disconnect tuple to trigger unsubscribe"
|
||||
)
|
||||
assert "unsubscribe_session_events" in handler_body, (
|
||||
"handler must unsubscribe in the cleanup path"
|
||||
)
|
||||
assert "finally:" in handler_body, "handler must use a finally block for cleanup"
|
||||
Reference in New Issue
Block a user