Testing

Use repost_client.testing to script attempts, record redacted requests, control time, and assert lifecycle events.

repost_client.testing ships inside repost-client with no extra dependencies. It drives the real generated method tree, validation, serialization, retry loop, cancellation, and observers through an injected ScriptedTransport that cannot open a network connection.

Deterministic client harness

deterministic_client_options() wires a scripted transport, recording observer, manual clock, fixed generators, fixed idempotency key, and zero retry entropy into one ClientOptions value:

import json
 
from repost_client.testing import deterministic_client_options
from repost_sdk import Book, Currency, RepostClient
 
harness = deterministic_client_options()
harness.transport.enqueue_response(
    202,
    json.dumps({
        "id": "msg_test",
        "type": "book.created",
        "customerId": "customer_test",
        "timestamp": "2026-01-01T00:00:00.000Z",
    }),
)
 
with RepostClient(harness.options) as repost:
    result = repost.webhooks.book.created(
        customer_id="customer_test",
        data=Book(
            title="Testing",
            authors=[],
            price=10.0,
            currency=Currency.USD,
        ),
    )
    harness.observer.await_quiescence()
 
assert result.id == "msg_test"
assert harness.transport.requests[0].attempt_number == 1
assert b'"title":"Testing"' in harness.transport.requests[0].body

Recorded requests retain the URL, serialized body, attempt number, idempotency key, and timeouts. The testing transport removes Authorization before storage.

Script responses and failures

ScriptedTransport consumes one FIFO script entry per attempt:

from repost_client import ErrorCode, FailureReason
 
harness.transport.enqueue_failure(
    ErrorCode.DNS,
    FailureReason.DNS_NOT_FOUND,
    committed=False,
).enqueue_response(202, accepted_body)

Use enqueue_pending() when a test must settle an attempt after cancellation, a deadline, or another event. ControlledResponse.respond(), .fail(), and .cancel() settle it exactly once. Script exhaustion fails as a custom-transport defect instead of falling through to the default network transport.

Drive retry timing

ManualScheduler pauses a nonzero retry delay until the test advances its ManualClock. Run the synchronous send on a worker thread, wait for the pending sleep, then advance time from the test thread:

import threading
 
from repost_client.testing import ManualClock, ManualScheduler
 
clock = ManualClock()
scheduler = ManualScheduler(clock)
completed: list[bool] = []
worker = threading.Thread(
    target=lambda: (scheduler.sleep(250, None), completed.append(True))
)
 
worker.start()
scheduler.await_pending(1)
clock.advance(250)
worker.join(1)
 
assert not worker.is_alive()
assert completed == [True]

For exact sequences, use sequence_entropy(), sequence_generators(), or sequence_idempotency_keys(). Fixed and failing variants cover stable values and pre-network failure paths.

Assert lifecycle events

RecordingObserver.events is an immutable snapshot. await_next() waits on a condition for the next event, and await_quiescence() waits until the runtime has dispatched every queued event:

from repost_client import ObserverEventKind
 
harness.observer.await_quiescence()
assert [event.kind for event in harness.observer.events] == [
    ObserverEventKind.OPERATION_START,
    ObserverEventKind.ATTEMPT_START,
    ObserverEventKind.ATTEMPT_END,
    ObserverEventKind.OPERATION_END,
]

Use the scripted transport for fast application tests. The SDK's own native pytest suite separately exercises the generated client through the default real HTTP transport and a local server.