Testing Helpers¶
chumicro_websockets.testing ships two in-memory fakes so libraries that depend on chumicro-websockets (and the library's own test suite) can drive WebSocketClient and WebSocketServer end-to-end without real sockets.
For the ticks domain, use chumicro_timing.testing.FakeTicks — pass it through the client's / server's ticks= kwarg.
Usage¶
FakeConnection¶
Bidirectional in-memory pipe satisfying the TCPClientSocket shape:
from chumicro_timing.testing import FakeTicks
from chumicro_sockets.testing import FakeSocketConnector
from chumicro_websockets import WebSocketClient
from chumicro_websockets.testing import FakeConnection
def test_client_handshake():
socket = FakeConnection()
clock = FakeTicks()
client = WebSocketClient(
transport_factory=lambda *_args, **_kwargs: FakeSocketConnector(
actions=["dns_ok", "tcp_ok"], socket=socket,
),
ticks=clock,
)
client.connect("ws://example.com/")
# First two ticks drive the connector; third sends the upgrade request.
client.handle(clock.ticks_ms())
client.handle(clock.ticks_ms())
client.handle(clock.ticks_ms())
assert b"GET / HTTP/1.1\r\n" in socket.peek_outbound()
Inject errors via raise_on_send / raise_on_recv:
socket = FakeConnection()
socket.raise_on_send = OSError(99, "send dead")
# Next client.handle() that calls send() raises this once, then resets.
Cap each send() call to simulate partial writes:
Signal peer-EOF (recv returns 0 instead of EAGAIN):
FakeListener¶
Stand-in for chumicro_sockets.listener:
from chumicro_timing.testing import FakeTicks
from chumicro_websockets import WebSocketServer
from chumicro_websockets.testing import FakeConnection, FakeListener
def test_server_accepts():
listener = FakeListener()
peer = FakeConnection()
listener.queue_accept(peer)
clock = FakeTicks()
server = WebSocketServer(
listener=listener,
on_connection=lambda conn: None,
ticks=clock,
)
server.handle(clock.ticks_ms()) # accepts the queued peer
assert server.connection_count == 1
Ticks domain¶
For ticks-domain fakes use chumicro_timing.testing.FakeTicks. clock.advance(ms) jumps the simulated clock forward to drive timeouts, auto-ping cadences, and pong-overdue watchdogs:
from chumicro_timing.testing import FakeTicks
from chumicro_websockets import WebSocketClient
from chumicro_websockets.testing import FakeConnection
clock = FakeTicks()
client = WebSocketClient(
transport_factory=lambda *_args, **_kwargs: FakeSocketConnector(
actions=["dns_ok", "tcp_ok"], socket=FakeConnection(),
),
handshake_timeout_ms=1000,
ticks=clock,
)
client.connect("ws://example.com/")
clock.advance(2000) # past the handshake deadline
client.handle(clock.ticks_ms())
# Client now CLOSED with WebSocketTimeoutError.
Usage from other libraries¶
Libraries that depend on chumicro-websockets can import the fakes directly:
from chumicro_timing.testing import FakeTicks
from chumicro_websockets.testing import FakeConnection, FakeListener
Project convention: libraries that expose injectable services ship their own test fakes alongside the production code.
For end-to-end client ↔ server loopback, see tests/test_integration.py in this library — it pumps bytes between paired FakeConnection objects to drive both runners through their full lifecycle in-process.
API Reference¶
chumicro_websockets.testing
¶
Test helpers for libraries that depend on chumicro-websockets.
The public fakes are :class:FakeConnection and :class:FakeListener.
FakeConnection
¶
Bidirectional in-memory pipe modeling a TCP client socket.
feed_inbound(data)
¶
Append data to the inbound queue (will be visible to recv_into).
read_outbound()
¶
Drain everything the local end has written and return it.
peek_outbound()
¶
Return the outbound buffer without draining (non-destructive).
close_inbound()
¶
Signal peer-EOF: next recv_into returns 0 instead of EAGAIN.
setblocking(flag)
¶
Accept the call; the fake is always non-blocking.
send(data)
¶
Append data to outbound; return how many bytes were "sent".
recv_into(buffer, nbytes=0)
¶
Pull up to nbytes (or len(buffer)) into buffer; EAGAIN if empty.
close()
¶
Mark the connection closed.