testing.chunks

Page actions AI-ready formats and sharing
Open LLM text
Share with AI
Ask Claude Ask ChatGPT Ask Gemini Ask Copilot

In-process ASGI chunk capture for streaming-response assertions.

TestClient.request() joins every http.response.bodyASGI message into a single bufferedResponse, which is the right shape for ordinary handlers but erases the framingof…

In-process ASGI chunk capture for streaming-response assertions.

TestClient.request() joins every http.response.bodyASGI message into a single bufferedResponse, which is the right shape for ordinary handlers but erases the framing of a streaming response. For Stream / TemplateStream / Suspensethe boundaries between body messages are load-bearing — they are the difference between a progressive, shell-first render and a fully buffered one.

CapturedStreampreserves that framing: it records the ordered list of non-empty body chunks exactly as the ASGIsend()callable received them (one entry perhttp.response.bodymessage), so a test can assert that a response arrives as more than one chunk, that the shell bytes precede the deferred/streamed-block bytes, and that specific markers land in the expected chunk.

This is a socket-free counterpart to a real chunked-transfer client: the same send()messages an ASGI server would write to the wire, captured in memory. Real socket binding is unavailable in the CI sandbox, so structural and ordering assertions live here; wall-clock timing assertions do not (they need a socket and are flaky).

testing.chunks

Name Type Default Description
type
qualified_name
element_type
description
source_file
line_number
is_autodoc
autodoc_element
_autodoc_template
_autodoc_url_path
_autodoc_page_type
title
doc_content_hash

Symbols on this page

CapturedStream
class

The ordered body chunks captured from a single ASGI response.

Returned byrequest_chunks() (and the *_chunks convenience wrappers). Each entry in chunksis the decoded body of onehttp.response.bodyASGI message with a non-empty body, in send order. The terminal empty close message is not recorded.

For a bufferedResponsethis is a single chunk; for a streaming response it is one chunk per flushed boundary (Suspense shell + one chunk per deferred block,Stream/TemplateStream flush boundaries, etc.).

View source · /home/runner/work/chirp/chirp/site/../src/chirp/testing/chunks.py:1