blackbull.testing

blackbull.testing

Application test instruments.

Native helpers test dispatch; NativeTestServer tests the wire stack; TestClient tests the external ASGI boundary. Choosing one does not cover the other layers. TestClient runs lifespan and raises reported startup or shutdown failures; see docs/guide/testing.md.

NativeClient

Synchronous Tier 1 client, for tests written as plain def.

Owns one background event loop for the whole session — not one per request — and drives the ASGI lifespan protocol around it, so startup hooks have run before the first request the way they have in production::

with NativeClient(app) as client:
    resp = client.get('/hello')
    assert resp.status == 200

Prefer the module-level coroutines from an async def test: they call the app on the test's own loop with no thread hand-off at all.

__enter__()

Start the background loop and run the application's lifespan startup.

__init__(app)

Prepare a client for app; the loop thread starts on __enter__.

delete(path, **kwargs)

DELETE path on the app's own dispatch, no socket involved.

get(path, **kwargs)

GET path on the app's own dispatch, no socket involved.

head(path, **kwargs)

HEAD path on the app's own dispatch, no socket involved.

options(path, **kwargs)

OPTIONS path on the app's own dispatch, no socket involved.

patch(path, **kwargs)

PATCH path on the app's own dispatch, no socket involved.

post(path, **kwargs)

POST path on the app's own dispatch, no socket involved.

put(path, **kwargs)

PUT path on the app's own dispatch, no socket involved.

request(conn, *, body=b'')

Full-control form — see request.

NativeTestResponse dataclass

What Tier 1 collected from the app's send channel.

body is the concatenation of every http.response.body chunk, so a streaming handler and a single-shot one are compared the same way. events keeps the raw ASGI events for tests that assert on the emission sequence itself (chunk boundaries, more_body flags, trailers).

json()

Parse the body as JSON.

text(encoding='utf-8')

Decode the body as text (errors='replace').

NativeTestServer

BlackBull's own server on a loopback port, for full-stack tests.

Every layer runs: TCP accept, ConnectionActor, HTTP1Actor parsing, the native Connection, dispatch, and the bytes the sender puts on the wire. Anything whose answer depends on the wire — keep-alive reuse, the HEAD body strip, chunked framing, connection close semantics — is only observable here.

Async form (preferred: the server shares the test's event loop, as it shares the process loop in production)::

async with NativeTestServer(app) as server:
    resp = await server.client.get('/hello')

Synchronous form, for plain def tests — the server runs on one background loop for the session::

with NativeTestServer(app) as server:
    resp = server.client.get('/hello')

The listener binds 127.0.0.1 only, so a test never exposes a port beyond the machine. Plaintext HTTP/1.1 and WebSocket; TLS and HTTP/2 are out of scope for this tier — reach for blackbull.fault_injection or the conformance suites there.

client property

An HTTP client bound to this server.

httpx.AsyncClient under async with; a synchronous façade over the same client under plain with.

url property

Base URL of the running server, e.g. http://127.0.0.1:54321.

TestClient

Synchronous HTTP/ASGITransport and native BlackBull WebSocket tests.

HTTP methods forward to httpx. WebSocket sessions pass a Connection and do not exercise a generic scope-only ASGI app. Use as a context manager to run lifespan startup and shutdown.

cookies property

Persistent cookie jar, forwarded from the underlying httpx.AsyncClient.

Same semantics as httpx.Client.cookies: cookies set by responses persist across requests on the same client.

headers property

Default headers applied to every request, forwarded from the underlying httpx.AsyncClient.

__enter__()

Start the background loop and run the application's lifespan startup.

__init__(app, base_url='http://testserver', raise_app_exceptions=True, root_path='', cookies=None, headers=None, follow_redirects=False)

Prepare a client; the loop thread and lifespan start on __enter__.

Parameters:

Name	Type	Description	Default
app	Any	The ASGI application under test.	required
base_url	str	Prefix for relative request URLs.	'http://testserver'
raise_app_exceptions	bool	Re-raise an exception from the application instead of turning it into a 500 response.	True
root_path	str	ASGI root_path for every request.	''
cookies	Any | None	Initial cookie jar.	None
headers	Any | None	Default headers for every request.	None
follow_redirects	bool	Follow 3xx responses automatically.	False

delete(url, **kwargs)

DELETE url through request.

get(url, **kwargs)

GET url through request.

head(url, **kwargs)

HEAD url through request. Response carries headers only.

options(url, **kwargs)

OPTIONS url through request.

patch(url, **kwargs)

PATCH url through request.

post(url, **kwargs)

POST url through request.

put(url, **kwargs)

PUT url through request.

request(method, url, **kwargs)

Send method to url and return the response.

Keyword arguments go to httpx.AsyncClient.request. Raises RuntimeError if the client is used outside its with block, since the loop thread and lifespan are what that block owns.

websocket_connect(url, subprotocols=None, headers=None, cookies=None, timeout=5.0)

Open a WebSocket session against the application.

url is a path (relative to the app), e.g. /ws or /ws?token=abc. Returns a WebSocketTestSession that should itself be used as a context manager.

WebSocketDisconnect

Bases: Exception

The peer closed the connection.

Raised by WebSocket.receive and its typed variants. Iterating with async for handles this for you — the loop simply ends — so catch it only when you call receive() directly and need the close code.

code is the RFC 6455 close code the peer sent, or 1005 when it sent none. reason is its optional UTF-8 explanation.

WebSocketTestSession

Synchronous BlackBull WebSocket session using its native Connection boundary.

Open through TestClient.websocket_connect as a context manager. Server close or handshake rejection raises WebSocketDisconnect.

__init__(app, path, subprotocols=None, headers=None, cookies=None, timeout=5.0)

Prepare a session; the handshake runs on __enter__.

Parameters:

Name	Type	Description	Default
app	Any	The ASGI application to connect to.	required
path	str	Request path, including any query string.	required
subprotocols	list[str] | None	Offered in Sec-WebSocket-Protocol.	None
headers	list[tuple[str, str]] | None	Extra request headers, as (name, value) string pairs.	None
cookies	dict[str, str] | None	Sent as a single Cookie header.	None
timeout	float	Seconds any one receive will wait before giving up.	5.0

close(code=1000)

Disconnect with code and tear the session down.

Idempotent, and safe on a session whose handshake never completed. __exit__ calls it, so a with block needs no explicit close.

iter_bytes()

Yield successive binary messages from the server until the WebSocket closes.

Mirror of iter_text for binary frames.

iter_text()

Yield text messages until close; convert WebSocketDisconnect to iterator termination.

receive_bytes()

The next binary message.

Raises WebSocketDisconnect if the server closed instead, and RuntimeError if what arrived was a text message.

receive_json()

The next text message, parsed as JSON.

receive_text()

The next text message.

Raises WebSocketDisconnect if the server closed instead, and RuntimeError if what arrived was a binary message.

send_bytes(data)

Send data to the application as a binary message.

send_json(data)

Serialise data as JSON and send it as a text message.

send_text(text)

Send text to the application as a text message.

build_connection(method, path, *, headers=None, body=b'', http_version='1.1', scheme='http', root_path='', client=_TEST_CLIENT_ADDR, server=_TEST_SERVER_ADDR)

Build a Connection without protocol parsing.

Split query from path; retain undecoded raw_path and percent-decode path. Supply missing Host and Content-Length; an explicit Content-Length wins so tests can synthesize inconsistent framing.