hermes-webui/tests/test_session_events_http_integration.py

"""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"