`blackbull.testing`¶

`blackbull.testing` ¶

Test clients for BlackBull applications — three instruments, three layers.

BlackBull threads a typed :class:~blackbull.connection.Connection end to end and keeps the ASGI scope dict at two boundaries only. That is why there is more than one test client here, and why picking the right one matters: each drives a different layer, and a defect on one is invisible to the others.

============================ ============================================= Instrument What it exercises ============================ ============================================= :mod:blackbull.testing.native Application logic — routing, middleware, handlers, DI, events — through the native app(conn, receive, send) entry point that every production request takes. The default choice for everyday tests. :class:~blackbull.testing.native.NativeTestServer The full stack on a real loopback socket: protocol parsing, framing, keep-alive, connection lifecycle, wire bytes. :class:TestClient The ASGI compatibility boundary — the as_scope() / from_scope() round-trip, driven the way an external ASGI host (uvicorn, httpx.ASGITransport) drives it. ============================ =============================================

:class:TestClient is deliberately not the default. It reaches the app through httpx.ASGITransport → ASGI scope dict → from_scope(), so the isinstance(conn, Connection) branch of BlackBull.__call__ is never taken by it. What it uniquely covers is the conversion chain itself: a missing _CONNECTION_FIELDS entry or a from_scope coercion bug shows up here and nowhere else in the suite, which is exactly why it stays.

TestClient usage — a boundary-conformance instrument::

from blackbull import BlackBull
from blackbull.testing import TestClient

app = BlackBull()

@app.route('/')
async def hello():
    return "hi"

def test_asgi_boundary():
    with TestClient(app) as client:
        response = client.get('/')
        assert response.status_code == 200
        assert response.text == "hi"

The client is a context manager so that ASGI lifespan.startup runs before any request and lifespan.shutdown runs on exit. Apps that don't implement the lifespan protocol are tolerated silently.

For everything else, start from 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.

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

Full-control form — see :func: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` ¶

In-memory HTTP+WebSocket test client for ASGI 3.0 applications.

Provides a synchronous façade over httpx.AsyncClient + httpx.ASGITransport by hosting an event loop in a background thread. HTTP request methods (get, post, put, …) forward to the underlying httpx.AsyncClient; WebSocket sessions use a dedicated bridge to the ASGI receive/send channels.

Use as a context manager so that the ASGI lifespan protocol runs around the test::

with TestClient(app) as client:
    ...

`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.

`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 :class:WebSocketTestSession that should itself be used as a context manager.

`WebSocketDisconnect` ¶

Bases: Exception

The peer closed the connection.

Raised by :meth: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 WebSocket session against an ASGI application.

Open via :meth:TestClient.websocket_connect as a context manager::

with client.websocket_connect('/ws') as ws:
    ws.send_text('ping')
    assert ws.receive_text() == 'pong'

Raises :class:WebSocketDisconnect when the server closes (or rejects) the connection.

`iter_bytes()` ¶

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

Mirror of :meth:iter_text for binary frames.

`iter_text()` ¶

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

Stops cleanly when the server emits a websocket.close — the :class:WebSocketDisconnect raised by the underlying receive is caught and converted into normal iterator termination, so the test can write::

with client.websocket_connect('/stream') as ws:
    for msg in ws.iter_text():
        ...

without an explicit try/except around the loop.

`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 the :class:Connection an H/1.1 request line would have produced.

The field derivations mirror :meth:HTTP1Actor._parse so a Tier 1 test and a real request agree on what the handler sees:

the query string is split off path and carried in query_string, never in raw_path;
path is percent-decoded, raw_path keeps the undecoded bytes;
a host header is supplied when the caller gave none, because every HTTP/1.1 request carries one (RFC 9112 §3.2) and code that reads it would otherwise behave differently under test than on the wire;
a content-length is derived from body for the same reason — an explicit one from the caller wins, so a test can still synthesise a mismatched framing header on purpose.

blackbull.testing¶