Skip to content

Testing

Stdlib-only instruments, shipped with the library so your integration tests and clientwright's own suite run on the same ground. Usage patterns: Guide → Testing your service.

Fault-injecting origin

Fault-injecting HTTP origin on the standard library, for conformance tests.

Routes: - /echo 200 with a small JSON body (method, path, headers echo) - /status/{code} responds with that status - /slow/{seconds} sleeps, then 200 - /redirect/{n} 302 chain of n hops ending at /echo - /redirect-loop 302 to itself forever - /flaky/{key}/{fails} first {fails} requests per key answer 503, then 200 - /retry-after/{seconds} 503 with a Retry-After header - /disconnect closes the connection without a response

Chaos routes (mid-stream and protocol-level faults): - /hang-body/{seconds} 200 announcing 10 bytes: 3 arrive, the rest after the stall - /drop-body 200 announcing 10 bytes but the connection dies after 3 - /garbage raw non-HTTP bytes instead of a status line - /reset hard TCP reset (SO_LINGER 0) instead of a response - /flaky-disconnect/{key}/{fails} first {fails} requests per key drop the connection, then 200

OriginServer

Bases: ThreadingHTTPServer

Context-managed origin bound to an ephemeral localhost port.

Source code in clientwright/core/testing/origin.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
class OriginServer(ThreadingHTTPServer):
    """Context-managed origin bound to an ephemeral localhost port."""

    daemon_threads = True

    def handle_error(self, request: object, client_address: object) -> None:
        """Silence per-connection tracebacks: aborted clients are the point here."""
        return None

    def __init__(self) -> None:
        super().__init__(("127.0.0.1", 0), _Handler)
        self.lock = threading.Lock()
        self.requests: list[tuple[str, str]] = []
        self.flaky_counters: dict[str, int] = {}
        self._thread = threading.Thread(target=self.serve_forever, daemon=True)

    @property
    def url(self) -> str:
        raw_host, port = self.server_address[0], self.server_address[1]
        host = raw_host.decode() if isinstance(raw_host, bytes) else raw_host
        return f"http://{host}:{port}"

    def request_count(self, path_prefix: str) -> int:
        with self.lock:
            return sum(1 for _, path in self.requests if path.startswith(path_prefix))

    def __enter__(self) -> OriginServer:
        self._thread.start()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        self.shutdown()
        self.server_close()

handle_error(request, client_address)

Silence per-connection tracebacks: aborted clients are the point here.

Source code in clientwright/core/testing/origin.py 
166
167
168
def handle_error(self, request: object, client_address: object) -> None:
    """Silence per-connection tracebacks: aborted clients are the point here."""
    return None

Deterministic doubles

Deterministic doubles for engine and telemetry tests.

ManualClock

Monotonic clock advanced by hand.

Source code in clientwright/core/testing/doubles.py 
 8
 9
10
11
12
13
14
15
16
17
18
class ManualClock:
    """Monotonic clock advanced by hand."""

    def __init__(self, start: float = 0.0) -> None:
        self._now = start

    def __call__(self) -> float:
        return self._now

    def advance(self, seconds: float) -> None:
        self._now += seconds

RecordingMetrics dataclass

ClientMetricsProtocol implementation that remembers every record.

Source code in clientwright/core/testing/doubles.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@dataclass(slots=True)
class RecordingMetrics:
    """ClientMetricsProtocol implementation that remembers every record."""

    calls: list[dict[str, object]] = field(default_factory=list)
    attempts: list[dict[str, object]] = field(default_factory=list)
    body_durations: list[dict[str, object]] = field(default_factory=list)
    inflight: list[dict[str, object]] = field(default_factory=list)
    circuit_states: list[dict[str, object]] = field(default_factory=list)
    redirect_hops: list[dict[str, object]] = field(default_factory=list)
    retry_skips: list[dict[str, object]] = field(default_factory=list)
    uninstrumented: list[dict[str, object]] = field(default_factory=list)

    def record_call(
        self,
        *,
        service: str,
        adapter: str,
        seam: str,
        method: str,
        origin: str,
        route: str,
        status: str,
        outcome: str,
        duration: float,
    ) -> None:
        self.calls.append(
            {
                "service": service,
                "adapter": adapter,
                "seam": seam,
                "method": method,
                "origin": origin,
                "route": route,
                "status": status,
                "outcome": outcome,
                "duration": duration,
            }
        )

    def record_body_duration(
        self,
        *,
        service: str,
        adapter: str,
        seam: str,
        method: str,
        origin: str,
        route: str,
        duration: float,
    ) -> None:
        self.body_durations.append({"method": method, "origin": origin, "route": route, "duration": duration})

    def record_attempt(
        self,
        *,
        service: str,
        adapter: str,
        seam: str,
        method: str,
        origin: str,
        outcome: str,
        duration: float,
    ) -> None:
        self.attempts.append({"method": method, "origin": origin, "outcome": outcome, "duration": duration})

    def inflight_delta(self, *, service: str, adapter: str, seam: str, origin: str, delta: int) -> None:
        self.inflight.append({"origin": origin, "delta": delta})

    def record_circuit_state(self, *, service: str, adapter: str, key: str, state: str) -> None:
        self.circuit_states.append({"key": key, "state": state})

    def record_redirect_hop(self, *, service: str, adapter: str, seam: str) -> None:
        self.redirect_hops.append({"adapter": adapter})

    def record_retry_skipped(self, *, service: str, adapter: str, seam: str, reason: str) -> None:
        self.retry_skips.append({"reason": reason})

    def record_uninstrumented_call(self, *, service: str, adapter: str, seam: str) -> None:
        self.uninstrumented.append({"adapter": adapter})

    @property
    def inflight_balance(self) -> int:
        return sum(int(record["delta"]) for record in self.inflight)  # type: ignore[call-overload]