Testing

An in-memory mock backend ships as a pytest plugin. It is auto-registered via an entry point, so no conftest wiring is needed. It implements the same ModbusConnection / ModbusUnit APIs, so code typed against ModbusUnit runs against it unchanged. This is how you test a device library with no hardware and no Home Assistant in the loop.

Fixtures

• mock_modbus_connection is a MockModbusConnection.
• mock_modbus_unit is its unit 1.

MockModbusConnection / MockModbusUnit are also importable from modbus_connection.mock for direct construction.

Seeding registers

Set values on the per-space stores (holding, input, coil, discrete). A single value fills one register. A list fills consecutive registers. A callable is evaluated on every read:

async def test_reads_setpoint(mock_modbus_unit):
    mock_modbus_unit.holding[40] = 1234  # single value
    mock_modbus_unit.holding[2] = [0x0001, 0x86A0]  # list -> consecutive registers
    mock_modbus_unit.holding[9] = lambda: 7  # callable -> evaluated per read

    assert await mock_modbus_unit.read_holding_registers(40, 1) == [1234]
    assert await mock_modbus_unit.read_holding_registers(2, 2) == [0x0001, 0x86A0]

Reads resolve against these stores. Writes mutate them and fire on_write callbacks.

Replaying a raw snapshot

A raw dump from async_read_raw(), for example one captured from a real device in a bug report, loads straight into the mock with load_raw(). You can reproduce that device and check your model decodes it. The dump is keyed by the four Modbus spaces holding, input, coil and discrete, which load_raw() maps onto the stores:

async def test_decodes_a_captured_device(mock_modbus_unit):
    mock_modbus_unit.load_raw({"holding": {0: 2301}, "coil": {0: True}})
    meter = Meter(mock_modbus_unit)
    await meter.async_update()
    assert meter.voltage == 230.1

Testing a component

The mock is a real ModbusUnit, so you test a Component exactly as production code uses it:

from modbus_connection.model import Component, gauge


class Meter(Component):
    voltage = gauge(0, 0.1, unit="V")


async def test_meter(mock_modbus_unit):
    mock_modbus_unit.holding[0] = 2301  # raw
    meter = Meter(mock_modbus_unit)
    await meter.async_update()
    assert meter.voltage == 230.1  # raw * 0.1

Asserting on the reads a poll issued

read_events logs every block read the unit received, in order, as a ReadEvent. Where async_read_raw() reports which addresses a poll covered, this reports the blocks the planner asked for. A test can pin down how many round-trips a poll costs, and how wide each one was:

async def test_poll_respects_the_controller_limits(mock_modbus_unit):
    await MyDevice(mock_modbus_unit).async_update()

    blocks = mock_modbus_unit.read_events
    assert len(blocks) == 3  # the whole map in three round-trips
    assert all(b.count <= 100 for b in blocks)  # controller caps a read at 100
    assert all(b.register_type == "holding" for b in blocks)  # no coils on this device

This is the assertion a device library wants when its controller only answers declared ranges or caps a request’s width. It needs no wrapper around the unit.

A read is recorded when it is dispatched, so a read the device then rejects still appears: the request went out.

Simulating a read failure

Arm fail_read and any read whose block covers that address raises the given error instead of returning values. This mirrors a device that refuses a register block it does not serve, such as an uninstalled module. register_type defaults to "holding". Use "input", "coil" or "discrete" for the other tables, which are independent. Pass None to clear:

async def test_read_refused(mock_modbus_unit):
    mock_modbus_unit.fail_read(1100, IllegalDataAddressError())
    with pytest.raises(IllegalDataAddressError):
        await mock_modbus_unit.read_holding_registers(1100, 4)
    await mock_modbus_unit.read_holding_registers(0, 4)  # other blocks unaffected

    mock_modbus_unit.fail_read(1100, None)  # clear it

Simulating a device that answers nothing

fail_requests arms one error for every read and write on the unit. This models a device that is powered down, unplugged, or behind a dead gateway, where no address is special. Use it instead of fail_read when the test need not know which address a component’s read plan reaches first. Pass None to let the unit answer again:

async def test_device_unreachable(mock_modbus_unit):
    mock_modbus_unit.fail_requests(ModbusTimeoutError())
    with pytest.raises(ModbusTimeoutError):
        await Meter(mock_modbus_unit).async_update()

    mock_modbus_unit.fail_requests(None)  # the device answers again

This models the device rather than the link. connected still follows the connection. Use simulate_connection_lost() for a transport drop. Reads are still recorded in read_events before they raise, so a test can assert what was attempted. It is per unit, so one silent device on a shared gateway does not silence its neighbours. Per-address fail_read / fail_write keep applying on top.

Reacting to writes

Register an on_write callback to simulate a device that changes state in response to a command, for example one that flips a “ready” flag when a command register is written:

def test_command_sets_ready(mock_modbus_unit):
    def respond(event):
        if event.address == 0:  # a command was written
            mock_modbus_unit.holding[100] = 1  # device flips its "ready" flag

    mock_modbus_unit.on_write(respond)

Simulating a rejected write

Arm fail_write and the next write covering that address raises the given error before the store is touched. The value is left unchanged and on_write callbacks do not fire. register_type defaults to "holding". Use "coil" for coil writes. The tables are independent. Pass None to clear.

Arm the typed exception for the condition. It constructs with its code implied, and it is what the backends raise:

async def test_write_rejected(mock_modbus_unit):
    mock_modbus_unit.holding[40] = 7
    mock_modbus_unit.fail_write(40, IllegalDataValueError())
    with pytest.raises(IllegalDataValueError):
        await mock_modbus_unit.write_register(40, 99)
    assert await mock_modbus_unit.read_holding_registers(40, 1) == [7]  # unchanged

    mock_modbus_unit.fail_write(40, None)  # clear it
    await mock_modbus_unit.write_register(40, 99)  # now succeeds

The error you arm is the condition you simulate:

mock_modbus_unit.fail_write(40, IllegalDataValueError())  # device rejects the value
mock_modbus_unit.fail_write(40, ModbusTimeoutError())  # device does not answer
mock_modbus_unit.fail_write(40, ModbusConnectionError())  # device unreachable
mock_modbus_unit.fail_write(40, ModbusProtocolError())  # corrupt reply

Simulating a dropped link

simulate_connection_lost() on the connection drops the link and fires every on_connection_lost callback. Use it to test code that observes the transport, like a coordinator marking entities unavailable. The drop is transient, as it is on a real connection: the next request establishes the link again.

async def test_reacts_to_a_drop(mock_modbus_connection, mock_modbus_unit):
    events = []
    mock_modbus_connection.on_connection_lost(lambda: events.append("lost"))

    mock_modbus_connection.simulate_connection_lost()
    assert events == ["lost"]
    assert mock_modbus_connection.connected is False

    await mock_modbus_unit.read_holding_registers(0, 1)  # reconnects on demand
    assert mock_modbus_connection.connected is True

close() behaves like the real thing too: it is permanent, does not fire the callbacks, and later requests raise ClientClosedError. disconnect() also matches the real connection: it drops the link without firing the callbacks, and the next request reconnects.

Canned responses for the other operations

The register and bit operations resolve against the stores. The diagnostic, file-record, and identification operations have no natural store. Arm each one you use with set_response(method, value). A callable value is evaluated per call. Without one, the mock raises NotImplementedError telling you which response to configure:

async def test_reads_server_id(mock_modbus_unit):
    mock_modbus_unit.set_response("report_server_id", b"\x11ACME v2")
    assert await mock_modbus_unit.report_server_id() == b"\x11ACME v2"

The operations that take a canned response: read_exception_status, report_server_id, read_fifo_queue, read_device_identification, read_file_record, diagnostics, get_comm_event_counter, and get_comm_event_log. (mask_write_register and read_write_registers work against the register stores directly, and write_file_record is accepted as a no-op.)

Mock API reference

MockModbusConnection

Implements the full ModbusConnection API in memory: connected, for_unit(unit_id), connect(), close(), and on_connection_lost(callback). It adds the test hook simulate_connection_lost(). for_unit returns the same MockModbusUnit per unit id, so the unit you seed is the unit the code under test reads.

MockModbusUnit

Implements the full ModbusUnit API against in-memory stores, plus the test configuration surface:

Member	Purpose
holding, input, coil, discrete	The per-space stores: dict of address to a value, a list (consecutive addresses), or a callable (evaluated per read). See RegisterSpec / CoilSpec.
on_write(callback)	Register a callback invoked with a WriteEvent for register and coil writes. Returns an unsubscribe callable.
read_events	The ReadEvent log of every block read the unit received, in order.
fail_write(address, error, *, register_type="holding")	Arm the exception matching writes raise ("holding" or "coil"); None clears it.
fail_read(address, error, *, register_type="holding")	Arm the exception reads covering the address raise ("holding", "input", "coil", or "discrete"); None clears it.
fail_requests(error)	Arm the exception every read and write on this unit raises; None clears it.
set_response(method, value)	Arm a canned response for a non-store operation.
load_raw(raw)	Load an async_read_raw() snapshot into the stores; raises ValueError for an unknown space.
set_message_spacing(seconds)	Records the interval on the message_spacing attribute for assertions; raises ValueError if negative.
require_timeout(seconds)	Records the requirement on the required_timeout attribute, None where nothing is required; raises ValueError if negative.
require_connect_delay(seconds)	Records the requirement on the required_connect_delay attribute, None where nothing is required; raises ValueError if negative.

WriteEvent

The frozen dataclass on_write callbacks receive:

Field	Type	Meaning
register_type	"holding" | "coil"	Which table was written.
address	int	The first written address.
values	list[int] | list[bool]	The written values, one per address.
function_code	int	The function code the write went out as: 0x06/0x10 for registers (force_fc16 makes a one-register write 0x10), 0x05/0x0F for coils, 0x16 for a mask write.

ReadEvent

The frozen dataclass read_events collects:

Field	Type	Meaning
register_type	"holding" | "input" | "coil" | "discrete"	Which table was read, named as in an async_read_raw() snapshot.
address	int	The block’s first address.
count	int	How many addresses the block covers.

RegisterSpec and CoilSpec

The store value types: int | list[int] | Callable[[], int | list[int]] for the register stores, and the bool equivalent for the bit stores.