Testing Helpers¶
chumicro_sockets.testing ships FakeSocket — an in-memory test double satisfying the duck-typed TCP socket surface (send / recv_into / close / setblocking) so downstream libraries (chumicro-mqtt, chumicro-requests) and your own network apps can reach high coverage without spinning up a network.
Usage¶
from chumicro_sockets.testing import FakeSocket
def test_handshake_writes_correct_prefix():
sock = FakeSocket()
# Script the response the server would have sent.
sock.enqueue_recv(b"\x20\x02\x00\x00") # MQTT CONNACK packet
client = MQTTClient(sock)
client.connect()
# Assert what we sent.
assert sock.sent.startswith(b"\x10") # MQTT CONNECT packet prefix
FakeSocket.sent is a bytearray that accumulates every send() call. enqueue_recv(chunk) queues a chunk for the next recv_into(); multiple chunks queue in FIFO order. enqueue_eagain_for_send(count) and enqueue_eagain_for_recv(count) script OSError(EAGAIN) raises so non-blocking-loop logic can be exercised deterministically.
A short read pushes the unconsumed tail back on the queue head — mimics real-socket fragmentation:
sock = FakeSocket()
sock.enqueue_recv(b"abcdef")
buffer = bytearray(8)
sock.recv_into(buffer, 3) # reads "abc"; "def" still queued
sock.recv_into(buffer, 8) # reads "def"
Usage from other libraries¶
Project convention: libraries that expose injectable services ship their own test fakes alongside the production code.
API Reference¶
chumicro_sockets.testing
¶
In-memory socket test doubles: FakeSocket, FakeUDPSocket, FakeSocketConnector.
FakeSocket
¶
In-memory TCP client socket for tests.
enqueue_recv(chunk)
¶
Append chunk to the recv-side queue.
enqueue_eagain_for_send(count=1)
¶
Script the next count :meth:send calls to raise EAGAIN.
enqueue_eagain_for_recv(count=1)
¶
Script the next count :meth:recv_into calls to raise EAGAIN.
simulate_peer_close()
¶
Simulate a clean peer FIN so recv_into returns 0 once the queue drains.
send(data)
¶
Write data into :attr:sent and return its length.
recv_into(buffer, nbytes=0)
¶
Pop the queue head into buffer and return the number of bytes written.
close()
¶
Mark the socket closed.
FakeUDPSocket
¶
In-memory UDP socket for tests.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
bind_host
|
str
|
Reported by :meth: |
'0.0.0.0'
|
bind_port
|
int
|
Reported by :meth: |
54321
|
enqueue_recv(data, *, host='0.0.0.0', port=0)
¶
Append a datagram to the recv-side queue.
enqueue_eagain_for_send(count=1)
¶
Script the next count :meth:sendto calls to raise EAGAIN.
enqueue_eagain_for_recv(count=1)
¶
Script the next count :meth:recvfrom_into calls to raise EAGAIN.
sendto(data, host, port)
¶
Append (bytes(data), host, port) to :attr:sent.
recvfrom_into(buffer, nbytes=0)
¶
Pop a queued datagram into buffer and return (n, (host, port)).
close()
¶
Mark the socket closed.
getsockname()
¶
Report the bound (host, port) tuple given at construction.
FakeSocketConnector
¶
Scriptable test double for :class:SocketConnector.