`blackbull.testing.native`¶

`blackbull.testing.native` ¶

Native dispatch and full-stack loopback test instruments.

request() bypasses sockets; NativeTestServer includes protocol parsing. Synchronous wrappers own a background loop for their whole lifetime.

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

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

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