Repository navigation
Fault injection recipes
Real devices and networks fail in a handful of typical ways. Each recipe below shows the failure, the modbus-sim fault that reproduces it, and a pytest test for the code that has to cope. They use pymodbus as the master (tested with 3.16) and the devices of the Browser demo; all of them run as they are against modbus-sim 0.2.0.
import time
import pytest
from pymodbus.client import ModbusTcpClient
from pymodbus.exceptions import ConnectionException, ModbusIOException
BENCH = {"devices": [
{"profile": "eastron-sdm630", "unit": 1, "name": "meter"},
{"profile": "omron-e5cc", "unit": 2, "name": "controller"},
{"profile": "sunspec-inverter", "unit": 3, "name": "inverter"},
]}
@pytest.fixture
def bench(modbus_bench_factory):
return modbus_bench_factory(BENCH)
@pytest.fixture
def client(bench):
c = ModbusTcpClient("127.0.0.1", port=bench.port, timeout=0.5, retries=0)
c.connect()
yield c
c.close()
def read_voltage(client):
r = client.read_input_registers(0, count=2, device_id=1)
return client.convert_from_registers(r.registers, client.DATATYPE.FLOAT32)Faults are switched with bench.inject(device, kind, **params) in tests, or
modbus-sim inject <device> <kind> name=value … against a running server. The
complete list with parameters is in the
README. Faults with a
count wear off after that many requests; the others stay until cleared.
Real life: a meter with a loose RS-485 wire, a gateway that lost the slave, a
device rebooting. pymodbus raises ModbusIOException after its timeout; code
that does not catch it stops polling everything else.
def read_voltage_or_none(client):
try:
return read_voltage(client)
except ModbusIOException:
return None
def test_meter_silent(bench, client):
bench.inject("meter", "no_response") # count=1
assert read_voltage_or_none(client) is None
assert read_voltage_or_none(client) > 200 # the next one answersno_response with count=null keeps the device silent until you clear it;
delay seconds=1.5 answers late instead — useful to find the timeout your
code really uses.
Real life: a device answers exception 06 (server device busy) while it
writes its flash, or 04 (server device failure) when something inside broke.
The request did not fail at the transport level — isError() is the check.
def test_busy(bench, client):
bench.inject("meter", "busy")
r = client.read_input_registers(0, count=2, device_id=1)
assert r.isError() and r.exception_code == 6
def test_exception_on_one_function(bench, client):
bench.inject("controller", "exception", code=4, function=3, count=3)
r = client.read_holding_registers(0x2000, count=1, device_id=2)
assert r.isError() and r.exception_code == 4function=3 limits the fault to one function code, so writes still work —
good for testing that a read failure does not stop the control loop.
Real life: a gateway closes idle connections, a switch restarts, a firewall
drops the session. pymodbus raises ConnectionException; the client must
reconnect.
def test_reconnect(bench, client):
bench.inject("meter", "drop_connection")
with pytest.raises(ConnectionException):
read_voltage(client)
client.close()
client.connect()
assert read_voltage(client) > 200Real life: a drifting sensor, a bad current transformer, a wrong scale factor in a configuration. Your plausibility checks and alarms should catch it.
def test_drift(bench, client):
bench.inject("meter", "drift", point="voltage_l1", per_second=20)
time.sleep(1.5)
assert read_voltage(client) > 253 # +10 %: your over-voltage alarm should firefreeze is the opposite: values stop changing, as when a device hangs but
its Modbus stack still answers. Watchdogs that look for changing values should
catch that.
Device faults change what the model reports — the registers stay valid, the values say something is wrong.
def test_phase_loss(bench, client):
bench.inject("meter", "phase_loss", phase=2)
r = client.read_input_registers(0, count=6, device_id=1)
v = client.convert_from_registers(r.registers, client.DATATYPE.FLOAT32)
assert v[1] == 0.0 and v[0] > 200
def test_sensor_break(bench, client):
bench.inject("controller", "sensor_break")
r = client.read_holding_registers(0x2000, count=2, device_id=2)
assert r.registers[0] == 0x7FFF and r.registers[1] & 1 << 6 # PV, input error bit
def test_grid_loss(bench, client):
bench.inject("inverter", "grid_loss")
r = client.read_holding_registers(40108, count=4, device_id=3)
state, _vendor, evt_hi, evt_lo = r.registers
assert state == 7 and (evt_hi << 16 | evt_lo) & 1 << 4 # fault, grid disconnectReal life: not a fault but a rule of the device — and a classic first-day bug. The E5CC answers exception 04 to every write while communications writing is off (see Real-device profiles). A driver should handle it once, explicitly:
def set_point(client, celsius, unit=2):
r = client.write_register(0x2103, round(celsius * 10), device_id=unit)
if r.isError() and r.exception_code == 4: # communications writing off
client.write_register(0x0000, 0x0001, device_id=unit)
r = client.write_register(0x2103, round(celsius * 10), device_id=unit)
return not r.isError()
def test_set_point_enables_writing(client):
assert set_point(client, 150)
r = client.read_holding_registers(0x2002, count=1, device_id=2)
assert r.registers == [1500]For code that parses frames itself (gateways, protocol stacks, your own master):
| Fault | What happens |
|---|---|
bad_crc |
RTU: the answer has a wrong CRC — a correct master discards it and times out |
wrong_transaction |
TCP: the answer carries another transaction ID — a correct master must not accept it as the answer to its request |
wrong_unit |
the answer comes from another unit ID |
truncate bytes=N |
the answer is cut short by N bytes |
modbus-sim · MIT licensed · Eastron, Omron and SunSpec are trademarks of their owners; modbus-sim is not affiliated with them.