Skip to content

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

from chumicro_sockets.testing import FakeSocket

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:getsockname as the bound host.

'0.0.0.0'
bind_port int

Reported by :meth:getsockname as the bound port.

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.

io_socket property

The socket for Runner.wait once built, or None before and after.

io_interest(now_ms)

Poll-interest bitmask matching the real SocketConnector: the handshake direction during awaiting_tls, write during awaiting_tcp, nothing else.