`blackbull.testing.native`¶

`blackbull.testing.native` ¶

Native-path test client — the two tiers that drive BlackBull's own request path rather than the ASGI compatibility boundary.

BlackBull threads a typed :class:~blackbull.connection.Connection end to end; the ASGI scope dict survives only at two boundaries. The compatibility client in :mod:blackbull.testing reaches the app through httpx.ASGITransport → scope dict → Connection.from_scope(), so the isinstance(conn, Connection) branch of BlackBull.__call__ — the branch every production request takes — is never exercised by it. A defect can therefore live on the native path while the whole compat-driven suite passes.

Two tiers close that, mirroring what every framework that owns its protocol stack provides:

Tier 1 — :func:request and the verb helpers build a Connection and call app(conn, receive, send) directly. No socket, no protocol actor: everything from Connection inward (dispatcher, middleware chain, router, handlers, DI, events, response serialisation). The equivalent of actix-web's init_service or Fastify's .inject()::

resp = await native.get(app, '/hello')
assert resp.status == 200

Tier 2 — :class:NativeTestServer binds a real loopback socket and runs BlackBull's own :class:~blackbull.server.server.Server, so a request travels accept → HTTP1Actor parse → Connection → native dispatch → wire bytes. The equivalent of aiohttp's TestServer or Go's httptest.NewServer::

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

Both tiers are async-first because the app entry point is a coroutine and a handler runs on the caller's event loop — which is also what production does. :class:NativeClient and the synchronous form of :class:NativeTestServer wrap them for tests written as plain def, each owning one background event loop for its whole lifetime rather than per request.

Which instrument to reach for is documented in docs/guide/testing.md: Tier 1 for application logic, Tier 2 for anything whose answer depends on the wire, and :class:~blackbull.testing.TestClient for the ASGI boundary itself.

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

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

`delete(app, path, **kwargs)` `async` ¶

Drive app with a DELETE through the native dispatch path.

`get(app, path, **kwargs)` `async` ¶

Drive app with a GET through the native dispatch path.

`head(app, path, **kwargs)` `async` ¶

Drive app with a HEAD through the native dispatch path.

The handler sees HEAD: rewriting it to GET and stripping the body is the H/1.1 actor's job (RFC 9110 §9.3.2), which is below this tier. Use :class:NativeTestServer to assert HEAD's wire behaviour.

`options(app, path, **kwargs)` `async` ¶

Drive app with an OPTIONS through the native dispatch path.

`patch(app, path, **kwargs)` `async` ¶

Drive app with a PATCH through the native dispatch path.

`post(app, path, **kwargs)` `async` ¶

Drive app with a POST through the native dispatch path.

`put(app, path, **kwargs)` `async` ¶

Drive app with a PUT through the native dispatch path.

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

Call app(conn, receive, send) and collect the response.

The full-control form: build (or mutate) a :class:Connection yourself and drive the app with it. The verb helpers below are thin wrappers over this.

The receive channel delivers body as one http.request event and then reports http.disconnect — the same terminal signal a real recipient gives once the body is drained, so a handler that over-reads gets the production answer rather than hanging.

blackbull.testing.native¶