`blackbull.testing`¶

`blackbull.testing` ¶

In-memory test client for ASGI 3.0 applications.

Lets you exercise a BlackBull (or any ASGI 3.0) app from a synchronous pytest test without binding a TCP socket. Built on httpx.ASGITransport for HTTP; uses a background-thread event loop to bridge sync calls to the (async-only) ASGI transport, to drive the lifespan protocol, and to host WebSocket sessions.

Typical usage::

from blackbull import BlackBull
from blackbull.testing import TestClient

app = BlackBull()

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

def test_hello():
    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.

`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

Raised when the server side closes (or rejects) the WebSocket.

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

blackbull.testing¶