test: update legacy SSE-break regression test for SSE_RELAY_CLOSE_EVENTS (#6527 maintainer fold-in)

The R5b static-pattern test pinned the old inline ('stream_end','error','cancel')
tuple; #6527 replaced it with the named SSE_RELAY_CLOSE_EVENTS frozenset. Accept
either shape and assert behavior (cancel+apperror in the resolved close set)
rather than source syntax. Intent-preserving; non-vacuous (fails on master:
SSE_RELAY_CLOSE_EVENTS absent + apperror not in old tuple).
This commit is contained in:
nesquena-hermes 2026-07-26 19:58:43 +00:00
parent c49568101c
commit c6e9ad79d2

View file

@ -195,10 +195,20 @@ def test_server_py_sse_loop_breaks_on_cancel(cleanup_test_sessions):
src = (REPO_ROOT / "server.py").read_text()
routes_src = (REPO_ROOT / "api" / "routes.py").read_text() if (REPO_ROOT / "api" / "routes.py").exists() else ""
combined = src + routes_src
m = re.search(r"if event in \([^)]+\):\s*break", combined)
assert m, "SSE break condition not found in server.py or api/routes.py"
assert "cancel" in m.group(), \
f"'cancel' missing from SSE break condition: {m.group()}"
# #6527: the break condition may be an inline tuple ("stream_end", ...) OR
# the named SSE_RELAY_CLOSE_EVENTS frozenset. Accept either shape and pin
# the BEHAVIOR (cancel closes the relay) against the resolved close set,
# not the source syntax.
m = re.search(
r"if event in (?:SSE_RELAY_CLOSE_EVENTS|\([^)]*cancel[^)]*\)):\s*break",
combined,
)
assert m, "SSE break/close condition not found in server.py or api/routes.py"
from api.run_journal import SSE_RELAY_CLOSE_EVENTS
assert "cancel" in SSE_RELAY_CLOSE_EVENTS, \
f"'cancel' missing from SSE relay close set: {SSE_RELAY_CLOSE_EVENTS}"
assert "apperror" in SSE_RELAY_CLOSE_EVENTS, \
f"'apperror' missing from SSE relay close set: {SSE_RELAY_CLOSE_EVENTS}"
# ── R6: Test cron isolation (Sprint 10) ──────────────────────────────────────