Tyler
da29188dd0
Adds three live-tail streaming endpoints that emit an NDJSON snapshot
then forward new event-bus events as they arrive, with a 15s idle
heartbeat (overridable via CYCLONE_TAIL_HEARTBEAT_S for tests).
Each endpoint:
1. yields a snapshot of existing rows as {"type":"item","data":<row>}
2. terminates the snapshot with {"type":"snapshot_end","data":{"count":N}}
3. subscribes to its event kind and forwards each new event as an
{"type":"item","data":<event>} line
4. emits a {"type":"heartbeat","data":{"ts":<iso>}} line every
CYCLONE_TAIL_HEARTBEAT_S seconds when idle
5. checks request.is_disconnected() before each yield and unsubscribes
from the bus on cleanup so a closed stream releases its queue
The shared tail loop lives in api._tail_events, which polls
bus.subscribe_raw()'s queue directly instead of using the bus's
async-iterator wrapper — wait_for on an async generator cancels the
inner future on timeout, which poisons subsequent __anext__ calls with
StopAsyncIteration. Queue.get() is idempotent under cancellation, so
heartbeats don't break the subscription.
EventBus gains an unsubscribe(queue, kinds) method (idempotent) so
the tail loop can release its queue in a try/finally. The disconnect
test asserts the subscriber list is empty after the body iterator is
closed, validating no queue leak per open stream.
Tests in test_api_stream_live.py: 8 tests covering snapshot shape,
post-snapshot publish, heartbeat timing, multi-item snapshots, and
client disconnect cleanup. Plus 2 tests in test_pubsub.py for the
new unsubscribe method.
607 lines
22 KiB
Python
607 lines
22 KiB
Python
"""Live-tail NDJSON streaming endpoint tests (Phase 3).
|
|
|
|
Each ``/api/.../stream`` endpoint:
|
|
1. yields the current snapshot as ``{"type":"item","data":<row>}`` lines
|
|
2. terminates the snapshot with ``{"type":"snapshot_end","data":{"count":N}}``
|
|
3. subscribes to the relevant EventBus kind and forwards new events
|
|
4. emits a ``{"type":"heartbeat","data":{"ts":<iso>}}`` line every
|
|
``CYCLONE_TAIL_HEARTBEAT_S`` seconds (default 15) when idle
|
|
5. returns cleanly when the client disconnects
|
|
6. responds with ``application/x-ndjson``
|
|
|
|
Implementation note (deviation from plan)
|
|
-----------------------------------------
|
|
The plan suggested using ``httpx.AsyncClient(transport=ASGITransport(app))``
|
|
and ``client.stream()``. That doesn't work for true long-lived streams:
|
|
|
|
* ``httpx.ASGITransport``'s response is buffered (``ASGIResponseStream``
|
|
joins every body chunk at the end), so ``client.stream()`` blocks until
|
|
the app finishes — a live tail never finishes.
|
|
* ``starlette.requests.Request.is_disconnected()`` relies on a
|
|
``http.disconnect`` message in the ASGI receive channel, which
|
|
``ASGITransport`` only delivers *after* the response body completes
|
|
(the receive coroutine awaits ``response_complete``). So even if the
|
|
caller closes the client side, the endpoint never observes a
|
|
disconnect.
|
|
|
|
Instead, we call the endpoint coroutine directly, build a synthetic
|
|
``Request`` and ``StreamingResponse``, and iterate ``body_iterator``
|
|
ourselves. That gives us byte-by-byte streaming, a real disconnect
|
|
signal via ``body_iterator.aclose()`` (which throws ``CancelledError``
|
|
into the generator), and works for arbitrarily long-lived tails.
|
|
|
|
The public surface (snapshot + subscribe + heartbeat + disconnect
|
|
cleanup) is verified exactly as the plan requires; only the transport
|
|
differs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from datetime import date, datetime, timezone
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from starlette.requests import Request
|
|
|
|
from cyclone.api import app, claims_stream, remittances_stream, activity_stream
|
|
from cyclone.parsers.models import (
|
|
BatchSummary,
|
|
BillingProvider,
|
|
ClaimHeader,
|
|
ClaimOutput,
|
|
Envelope,
|
|
Payer,
|
|
ParseResult,
|
|
Subscriber,
|
|
ValidationReport,
|
|
)
|
|
from cyclone.parsers.models_835 import (
|
|
ClaimAdjustment,
|
|
ClaimPayment,
|
|
Envelope as Envelope835,
|
|
FinancialInfo,
|
|
ParseResult835,
|
|
Payer835,
|
|
Payee835,
|
|
ReassociationTrace,
|
|
ServicePayment,
|
|
BatchSummary as BatchSummary835,
|
|
)
|
|
from cyclone.pubsub import EventBus
|
|
from cyclone.store import (
|
|
BatchRecord,
|
|
BatchRecord835,
|
|
CycloneStore,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _short_heartbeat(monkeypatch):
|
|
"""Set a 0.2s heartbeat for all stream tests in this module.
|
|
|
|
The production default is 15s. Tests use a tight value so that
|
|
generators sitting idle after the test reads its lines exit
|
|
promptly — ``aclose()`` on the body iterator cancels the
|
|
generator and the heartbeat timeout is the de-facto idle
|
|
signal in these tests.
|
|
"""
|
|
monkeypatch.setenv("CYCLONE_TAIL_HEARTBEAT_S", "0.2")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Test helpers
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _make_claim_837(claim_id: str = "CLM-1", charge: str = "124.00") -> ClaimOutput:
|
|
return ClaimOutput(
|
|
claim_id=claim_id,
|
|
control_number="0001",
|
|
transaction_date=date(2026, 6, 19),
|
|
billing_provider=BillingProvider(name="Test", npi="1234567890"),
|
|
subscriber=Subscriber(
|
|
first_name="Jane", last_name="Doe", member_id=f"M-{claim_id}",
|
|
),
|
|
payer=Payer(name="Test Payer", id="P1"),
|
|
claim=ClaimHeader(
|
|
claim_id=claim_id, total_charge=Decimal(charge),
|
|
frequency_code="1", place_of_service="11",
|
|
),
|
|
diagnoses=[],
|
|
service_lines=[],
|
|
validation=ValidationReport(passed=True, errors=[], warnings=[]),
|
|
raw_segments=[],
|
|
)
|
|
|
|
|
|
def _make_result_837(n: int = 3, batch_id: str = "b-uuid-1") -> ParseResult:
|
|
claims = [
|
|
_make_claim_837(f"CLM-{i}", charge=str(100 + i))
|
|
for i in range(1, n + 1)
|
|
]
|
|
return ParseResult(
|
|
envelope=Envelope(
|
|
sender_id="S", receiver_id="R", control_number="0001",
|
|
transaction_date=date(2026, 6, 19),
|
|
),
|
|
claims=claims,
|
|
summary=BatchSummary(
|
|
input_file="test.txt", control_number="0001",
|
|
transaction_date=date(2026, 6, 19),
|
|
total_claims=n, passed=n, failed=0,
|
|
),
|
|
)
|
|
|
|
|
|
def _make_batch_record(kind: str = "837p", result=None, batch_id: str = "b-uuid-1",
|
|
n: int = 3):
|
|
return BatchRecord(
|
|
id=batch_id, kind=kind, input_filename="test.txt",
|
|
parsed_at=datetime(2026, 6, 19, 12, 0, tzinfo=timezone.utc),
|
|
result=result if result is not None else _make_result_837(n=n),
|
|
)
|
|
|
|
|
|
def _make_remit_with_cas(remit_id="CLP-1", status="1", charge="124.00",
|
|
paid="62.00", cas_amount="62.00", pcn=None):
|
|
cp = ClaimPayment(
|
|
payer_claim_control_number=pcn if pcn is not None else remit_id,
|
|
status_code=status,
|
|
status_label="Primary",
|
|
total_charge=charge, total_paid=paid,
|
|
service_payments=[
|
|
ServicePayment(
|
|
line_number=1, procedure_qualifier="HC", procedure_code="99213",
|
|
charge=charge, payment=paid,
|
|
adjustments=[ClaimAdjustment(group_code="CO", reason_code="45",
|
|
amount=cas_amount)],
|
|
),
|
|
],
|
|
)
|
|
return cp
|
|
|
|
|
|
def _make_835_result(claims):
|
|
return ParseResult835(
|
|
envelope=Envelope835(
|
|
sender_id="S", receiver_id="R", control_number="0001",
|
|
transaction_date=date(2026, 6, 19),
|
|
),
|
|
financial_info=FinancialInfo(
|
|
handling_code="C", paid_amount=Decimal("0"),
|
|
credit_debit_flag="C", payment_method=None,
|
|
),
|
|
trace=ReassociationTrace(
|
|
trace_type_code="1", trace_number="0001",
|
|
originating_company_id="S",
|
|
),
|
|
payer=Payer835(name="X", id="SKCO0"),
|
|
payee=Payee835(name="Y", npi="1234567890"),
|
|
claims=claims,
|
|
summary=BatchSummary835(
|
|
input_file="era.txt", control_number="0001",
|
|
transaction_date=date(2026, 6, 19),
|
|
total_claims=len(claims), passed=len(claims), failed=0,
|
|
),
|
|
)
|
|
|
|
|
|
def _make_request(path: str = "/api/claims/stream",
|
|
query_string: bytes = b"") -> Request:
|
|
"""Build a Starlette ``Request`` wired to ``app.state`` with a no-op
|
|
receive channel so the endpoint's ``is_disconnected()`` polling
|
|
works without raising ``RuntimeError: Receive channel has not
|
|
been made available``.
|
|
"""
|
|
# ``receive`` blocks until a message is delivered. We never deliver
|
|
# one, so the cancel scope inside ``is_disconnected()`` always
|
|
# returns ``False`` — exactly what we want for tests that close
|
|
# via ``body_iterator.aclose()`` rather than the disconnect path.
|
|
receive_started = asyncio.Event()
|
|
|
|
async def receive():
|
|
receive_started.set()
|
|
# Block forever; is_disconnected's cancel scope will cancel
|
|
# this await and we'll return None (treated as "no message").
|
|
await asyncio.Event().wait()
|
|
|
|
scope = {
|
|
"type": "http",
|
|
"method": "GET",
|
|
"headers": [],
|
|
"query_string": query_string,
|
|
"path": path,
|
|
"app": app,
|
|
"scheme": "http",
|
|
"server": ("testserver", 80),
|
|
"client": ("testclient", 50000),
|
|
}
|
|
return Request(scope, receive=receive)
|
|
|
|
|
|
def _resolve_query_defaults(endpoint) -> dict:
|
|
"""Extract the actual default values from a FastAPI endpoint signature.
|
|
|
|
``Query(None)`` etc. are marker objects — when called directly we
|
|
need to substitute their ``.default`` so the endpoint function
|
|
sees the same values it would receive over HTTP.
|
|
"""
|
|
import inspect
|
|
sig = inspect.signature(endpoint)
|
|
resolved: dict = {}
|
|
for name, param in sig.parameters.items():
|
|
default = param.default
|
|
if hasattr(default, "default"):
|
|
resolved[name] = default.default
|
|
else:
|
|
resolved[name] = default
|
|
return resolved
|
|
|
|
|
|
async def _call_endpoint(endpoint, path: str):
|
|
"""Invoke ``endpoint(request, **defaults)`` with resolved defaults."""
|
|
request = _make_request(path)
|
|
defaults = _resolve_query_defaults(endpoint)
|
|
# Drop ``request`` — it's the positional we pass explicitly.
|
|
defaults.pop("request", None)
|
|
return await endpoint(request, **defaults)
|
|
|
|
|
|
async def _read_lines(body_iter, n: int, timeout: float = 5.0) -> list[str]:
|
|
"""Read up to ``n`` lines from a StreamingResponse body iterator."""
|
|
buffer = b""
|
|
lines: list[str] = []
|
|
async with asyncio.timeout(timeout):
|
|
while len(lines) < n:
|
|
try:
|
|
chunk = await body_iter.__anext__()
|
|
except StopAsyncIteration:
|
|
break
|
|
buffer += chunk
|
|
while b"\n" in buffer and len(lines) < n:
|
|
raw, buffer = buffer.split(b"\n", 1)
|
|
if raw:
|
|
lines.append(raw.decode("utf-8"))
|
|
return lines
|
|
|
|
|
|
async def _read_one_line(body_iter, timeout: float = 5.0) -> str | None:
|
|
"""Read a single NDJSON line from the body iterator, or None on timeout."""
|
|
buffer = b""
|
|
async with asyncio.timeout(timeout):
|
|
while True:
|
|
try:
|
|
chunk = await body_iter.__anext__()
|
|
except StopAsyncIteration:
|
|
return None
|
|
buffer += chunk
|
|
if b"\n" in buffer:
|
|
raw, _ = buffer.split(b"\n", 1)
|
|
if raw:
|
|
return raw.decode("utf-8")
|
|
|
|
|
|
async def _drain_until_disconnect(body_iter, max_chunks: int = 20) -> None:
|
|
"""Close the body iterator and swallow the cancellation."""
|
|
try:
|
|
await body_iter.aclose()
|
|
except (asyncio.CancelledError, GeneratorExit):
|
|
pass
|
|
|
|
|
|
async def _read_until_type(
|
|
body_iter, target: str, *, timeout: float = 5.0, max_lines: int = 50
|
|
) -> dict | None:
|
|
"""Read lines from the body iterator until one has ``type == target``.
|
|
|
|
Used by tests that need to skip past heartbeats or multi-item
|
|
snapshot prefixes to assert on a specific event. Returns the first
|
|
matching parsed object, or ``None`` if ``max_lines`` is reached
|
|
without a match.
|
|
"""
|
|
buffer = b""
|
|
seen = 0
|
|
async with asyncio.timeout(timeout):
|
|
while seen < max_lines:
|
|
try:
|
|
chunk = await body_iter.__anext__()
|
|
except StopAsyncIteration:
|
|
return None
|
|
buffer += chunk
|
|
while b"\n" in buffer:
|
|
raw, buffer = buffer.split(b"\n", 1)
|
|
if not raw:
|
|
continue
|
|
seen += 1
|
|
obj = json.loads(raw.decode("utf-8"))
|
|
if obj.get("type") == target:
|
|
return obj
|
|
return None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Task 11 — claims stream
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
async def test_claims_stream_yields_snapshot_then_snapshot_end():
|
|
"""3 claims pre-loaded → 3 item lines + 1 snapshot_end line."""
|
|
s = CycloneStore()
|
|
s.add(_make_batch_record(n=3))
|
|
|
|
response = await _call_endpoint(claims_stream, "/api/claims/stream")
|
|
assert response.media_type.startswith("application/x-ndjson")
|
|
|
|
lines = await _read_lines(response.body_iterator, n=4)
|
|
items = [json.loads(line) for line in lines]
|
|
assert len(items) == 4
|
|
# First three are items; the order isn't guaranteed without an
|
|
# explicit sort, so check shape not exact ordering.
|
|
for item in items[:3]:
|
|
assert item["type"] == "item"
|
|
assert "id" in item["data"]
|
|
# Fourth is the snapshot_end terminator.
|
|
assert items[3] == {"type": "snapshot_end", "data": {"count": 3}}
|
|
await _drain_until_disconnect(response.body_iterator)
|
|
|
|
|
|
async def test_claims_stream_emits_new_item_after_publish():
|
|
"""After snapshot_end, a bus.publish('claim_written', …) → new item line."""
|
|
s = CycloneStore()
|
|
s.add(_make_batch_record(n=1))
|
|
|
|
response = await _call_endpoint(claims_stream, "/api/claims/stream")
|
|
# Drain snapshot phase: 1 item + 1 snapshot_end.
|
|
lines = await _read_lines(response.body_iterator, n=2)
|
|
assert json.loads(lines[1]) == {
|
|
"type": "snapshot_end", "data": {"count": 1},
|
|
}
|
|
# Advance the generator one more step so ``subscribe_raw`` runs
|
|
# and registers the subscriber queue. The next yield will be a
|
|
# heartbeat (with the 0.2s env var set in some tests); we don't
|
|
# care about its content here, just that we've moved past the
|
|
# sync subscription code.
|
|
_ = await _read_one_line(response.body_iterator, timeout=2.0)
|
|
|
|
# Now publish a claim_written event and assert a new item line
|
|
# streams out.
|
|
payload = {
|
|
"id": "NEW-CLM-1",
|
|
"patientName": "Live Tail",
|
|
"providerNpi": "1234567890",
|
|
"payerName": "Live Payer",
|
|
"cptCode": "99213",
|
|
"billedAmount": 999.99,
|
|
"receivedAmount": 0.0,
|
|
"status": "submitted",
|
|
"state": "submitted",
|
|
"denialReason": None,
|
|
"submissionDate": "2026-06-19T12:00:00Z",
|
|
"batchId": "b-uuid-1",
|
|
"parsedAt": "2026-06-19T12:00:00Z",
|
|
}
|
|
await app.state.event_bus.publish("claim_written", payload)
|
|
|
|
new_line = await _read_one_line(response.body_iterator, timeout=1.0)
|
|
assert new_line is not None, "no item line arrived after publish"
|
|
obj = json.loads(new_line)
|
|
assert obj["type"] == "item"
|
|
assert obj["data"]["id"] == "NEW-CLM-1"
|
|
await _drain_until_disconnect(response.body_iterator)
|
|
|
|
|
|
async def test_claims_stream_heartbeat_after_idle(monkeypatch):
|
|
"""With heartbeat=0.2s, a heartbeat line appears within ~1s of idle."""
|
|
monkeypatch.setenv("CYCLONE_TAIL_HEARTBEAT_S", "0.2")
|
|
s = CycloneStore()
|
|
s.add(_make_batch_record(n=1))
|
|
|
|
response = await _call_endpoint(claims_stream, "/api/claims/stream")
|
|
# Drain snapshot (2 lines).
|
|
lines = await _read_lines(response.body_iterator, n=2)
|
|
assert json.loads(lines[1]) == {
|
|
"type": "snapshot_end", "data": {"count": 1},
|
|
}
|
|
|
|
# Read subsequent lines until we see a heartbeat or time out.
|
|
deadline = asyncio.get_event_loop().time() + 1.5
|
|
saw_heartbeat = False
|
|
while asyncio.get_event_loop().time() < deadline:
|
|
new_line = await _read_one_line(
|
|
response.body_iterator, timeout=0.5,
|
|
)
|
|
if new_line is None:
|
|
continue
|
|
obj = json.loads(new_line)
|
|
if obj["type"] == "heartbeat":
|
|
assert "ts" in obj["data"]
|
|
saw_heartbeat = True
|
|
break
|
|
assert saw_heartbeat, "no heartbeat line arrived within 1.5s"
|
|
await _drain_until_disconnect(response.body_iterator)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Task 12 — remittances stream
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
async def test_remittances_stream_yields_snapshot_then_subscribes():
|
|
"""Pre-loaded 835 batch → snapshot (1 item + snapshot_end)."""
|
|
s = CycloneStore()
|
|
cp = _make_remit_with_cas(remit_id="CLP-1", pcn="PCN-1")
|
|
rec = BatchRecord835(
|
|
id="b-835-1", kind="835", input_filename="era.txt",
|
|
parsed_at=datetime(2026, 6, 19, 12, 0, tzinfo=timezone.utc),
|
|
result=_make_835_result([cp]),
|
|
)
|
|
s.add(rec)
|
|
|
|
response = await _call_endpoint(
|
|
remittances_stream, "/api/remittances/stream",
|
|
)
|
|
assert response.media_type.startswith("application/x-ndjson")
|
|
# Read the full snapshot (1 item + snapshot_end), skipping nothing.
|
|
snap_end = await _read_until_type(
|
|
response.body_iterator, "snapshot_end", timeout=2.0,
|
|
)
|
|
assert snap_end == {"type": "snapshot_end", "data": {"count": 1}}
|
|
await _drain_until_disconnect(response.body_iterator)
|
|
|
|
|
|
async def test_remittances_stream_emits_after_upsert_remittance():
|
|
"""A bus.publish('remittance_written', …) → new item line on stream."""
|
|
s = CycloneStore()
|
|
cp = _make_remit_with_cas(remit_id="CLP-0", pcn="PCN-0")
|
|
rec = BatchRecord835(
|
|
id="b-835-0", kind="835", input_filename="era0.txt",
|
|
parsed_at=datetime(2026, 6, 19, 12, 0, tzinfo=timezone.utc),
|
|
result=_make_835_result([cp]),
|
|
)
|
|
s.add(rec)
|
|
|
|
response = await _call_endpoint(
|
|
remittances_stream, "/api/remittances/stream",
|
|
)
|
|
# Drain snapshot to the terminator.
|
|
snap_end = await _read_until_type(
|
|
response.body_iterator, "snapshot_end", timeout=2.0,
|
|
)
|
|
assert snap_end == {"type": "snapshot_end", "data": {"count": 1}}
|
|
# Advance one more step so the subscription is registered before
|
|
# we publish. The next yielded line may be a heartbeat (with the
|
|
# 0.2s heartbeat fixture) or a real event — either is fine, we
|
|
# just need ``subscribe_raw`` to have run.
|
|
_ = await _read_one_line(response.body_iterator, timeout=2.0)
|
|
|
|
payload = {
|
|
"id": "PCN-NEW",
|
|
"claimId": "",
|
|
"payerName": "Tail Payer",
|
|
"paidAmount": 12.34,
|
|
"adjustmentAmount": 0.0,
|
|
"status": "received",
|
|
"denialReason": None,
|
|
"validationWarnings": [],
|
|
"receivedDate": "2026-06-19T12:00:00Z",
|
|
"batchId": "b-835-1",
|
|
"parsedAt": "2026-06-19T12:00:00Z",
|
|
"adjustments": [],
|
|
}
|
|
await app.state.event_bus.publish("remittance_written", payload)
|
|
|
|
# Read until the new item line appears (skip any heartbeats that
|
|
# race in before the published event).
|
|
new_obj = await _read_until_type(
|
|
response.body_iterator, "item", timeout=2.0,
|
|
)
|
|
assert new_obj is not None, "no item line arrived after publish"
|
|
assert new_obj["data"]["id"] == "PCN-NEW"
|
|
await _drain_until_disconnect(response.body_iterator)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Task 13 — activity stream
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
async def test_activity_stream_yields_snapshot_then_subscribes():
|
|
"""An 837 batch seeds activity rows → snapshot has at least 1 item."""
|
|
s = CycloneStore()
|
|
s.add(_make_batch_record(n=2))
|
|
|
|
response = await _call_endpoint(
|
|
activity_stream, "/api/activity/stream",
|
|
)
|
|
assert response.media_type.startswith("application/x-ndjson")
|
|
# Read to snapshot_end — activity may have multiple items.
|
|
snap_end = await _read_until_type(
|
|
response.body_iterator, "snapshot_end", timeout=2.0,
|
|
)
|
|
assert snap_end is not None
|
|
assert snap_end["data"]["count"] >= 1
|
|
await _drain_until_disconnect(response.body_iterator)
|
|
|
|
|
|
async def test_activity_stream_emits_after_record_activity():
|
|
"""A bus.publish('activity_recorded', …) → new item line on stream."""
|
|
s = CycloneStore()
|
|
s.add(_make_batch_record(n=1))
|
|
|
|
response = await _call_endpoint(
|
|
activity_stream, "/api/activity/stream",
|
|
)
|
|
# Drain snapshot.
|
|
snap_end = await _read_until_type(
|
|
response.body_iterator, "snapshot_end", timeout=2.0,
|
|
)
|
|
assert snap_end is not None
|
|
# Advance one step so subscription is registered.
|
|
_ = await _read_one_line(response.body_iterator, timeout=2.0)
|
|
|
|
payload = {
|
|
"kind": "claim_submitted",
|
|
"ts": "2026-06-19T12:00:00Z",
|
|
"batchId": "b-uuid-1",
|
|
"claimId": "LIVE-CLM",
|
|
"remittanceId": None,
|
|
"payload": {"message": "live tail event"},
|
|
}
|
|
await app.state.event_bus.publish("activity_recorded", payload)
|
|
|
|
new_obj = await _read_until_type(
|
|
response.body_iterator, "item", timeout=2.0,
|
|
)
|
|
assert new_obj is not None, "no item line arrived after publish"
|
|
assert new_obj["data"]["kind"] == "claim_submitted"
|
|
assert new_obj["data"]["claimId"] == "LIVE-CLM"
|
|
await _drain_until_disconnect(response.body_iterator)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Task 14 — disconnect cancels subscription
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
async def test_client_disconnect_cancels_subscription(monkeypatch):
|
|
"""After closing the body iterator, no claim_written subscribers remain.
|
|
|
|
Closes the body iterator (the test-side disconnect signal — see
|
|
the module docstring). The generator must propagate the
|
|
``CancelledError`` and tear down its bus subscription.
|
|
"""
|
|
monkeypatch.setenv("CYCLONE_TAIL_HEARTBEAT_S", "0.1")
|
|
s = CycloneStore()
|
|
s.add(_make_batch_record(n=1))
|
|
|
|
bus: EventBus = app.state.event_bus
|
|
response = await _call_endpoint(claims_stream, "/api/claims/stream")
|
|
# Read the full snapshot to the terminator.
|
|
snap_end = await _read_until_type(
|
|
response.body_iterator, "snapshot_end", timeout=2.0,
|
|
)
|
|
assert snap_end == {"type": "snapshot_end", "data": {"count": 1}}
|
|
# Advance one step into the subscription phase so subscribe_raw
|
|
# has run and registered the queue.
|
|
_ = await _read_one_line(response.body_iterator, timeout=2.0)
|
|
|
|
# Pre-condition: at least one subscriber registered while the
|
|
# stream was open.
|
|
assert len(bus._subscribers.get("claim_written", [])) >= 1
|
|
|
|
# Client disconnect: close the body iterator. The endpoint
|
|
# generator receives ``CancelledError`` via Starlette's
|
|
# cancellation path and returns; its `bus.subscribe_raw` queue is
|
|
# released. Wait briefly for cleanup.
|
|
await _drain_until_disconnect(response.body_iterator)
|
|
for _ in range(50):
|
|
if not bus._subscribers.get("claim_written"):
|
|
break
|
|
await asyncio.sleep(0.02)
|
|
|
|
assert bus._subscribers.get("claim_written", []) == [], (
|
|
f"subscribers leaked after disconnect: "
|
|
f"{bus._subscribers.get('claim_written')!r}"
|
|
) |