Classes
SSETestResult
4
▼
Collected events from an SSE endpoint.
Returned by ``TestClient.sse()`` after the connection close…
SSETestResult
4
▼
Collected events from an SSE endpoint.
Returned byTestClient.sse()after the connection closes.
Attributes
| Name | Type | Description |
|---|---|---|
events |
tuple[SSEEvent, ...]
|
— |
heartbeats |
int
|
— |
status |
int
|
— |
headers |
dict[str, str]
|
— |
_SseAttrExtractor
2
▼
Collect htmx-sse wiring attributes from rendered HTML.
_SseAttrExtractor
2
▼
Collect htmx-sse wiring attributes from rendered HTML.
Methods
handle_starttag
2
▼
handle_starttag
2
▼
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None
Parameters
| Name | Type | Description |
|---|---|---|
tag |
— |
|
attrs |
— |
Internal Methods 1 ▼
__init__
0
▼
__init__
0
▼
def __init__(self) -> None
Functions
parse_sse_frames
1
tuple[list[SSEEvent], in…
▼
Parse raw SSE text into structured events and heartbeat count.
Splits on doubl…
parse_sse_frames
1
tuple[list[SSEEvent], in…
▼
def parse_sse_frames(raw: str) -> tuple[list[SSEEvent], int]
Parse raw SSE text into structured events and heartbeat count.
Splits on double-newline boundaries. Each block is parsed into an
SSEEvent. Comment lines (starting with :) are counted as
heartbeats if they contain "heartbeat".
Parameters
| Name | Type | Description |
|---|---|---|
raw |
str |
Returns
tuple[list[SSEEvent], int]
extract_sse_attrs
1
tuple[list[str], set[str…
▼
Return ``(sse-connect values, sse-swap event names)`` from rendered HTML.
Used…
extract_sse_attrs
1
tuple[list[str], set[str…
▼
def extract_sse_attrs(html: str) -> tuple[list[str], set[str]]
Return(sse-connect values, sse-swap event names)from rendered HTML.
Used byassert_sse_wired() to cross-check wiring against the
actual event names a stream emits.
Parameters
| Name | Type | Description |
|---|---|---|
html |
str |
Returns
tuple[list[str], set[str]]
assert_sse_wired
4
None
▼
Cross-check SSE wiring between the page and the stream.
Fails if the page has …
async
assert_sse_wired
4
None
▼
async def assert_sse_wired(client: TestClient, page_path: str, sse_path: str, *, max_events: int = 5) -> None
Cross-check SSE wiring between the page and the stream.
Fails if the page hassse-connect but no sse-swap, or if a
listener in the page waits for an event name the stream never emits
(the class of silent failure you would otherwise only catch in a
browser).
Stream-emitted events that no listener consumes are allowed — streams
may emitstatus/closemetadata that is not a swap target.
Parameters
| Name | Type | Description |
|---|---|---|
client |
TestClient |
|
page_path |
str |
|
sse_path |
str |
|
max_events |
int |
Default:5
|