Fault injection recipes

What a host should do when a device misbehaves, and how to provoke it. Every recipe below is a real test against hart-sim 0.1.2, using its fixtures and client.

import pytest

from hart_sim.client import DeviceHandle, HartError, HartTimeout

A failed sensor

The device still answers, so a host that only looks at the number misses it. The status byte says what is wrong, and the current goes to the alarm value. A value marked bad must not be used.

def test_sensor_failure(hart_bench):
    hart_bench.devices["PT-101"].faults.inject("sensor_failure")
    with hart_bench.client() as client:
        variables = client.handle(0).read_variables()
    assert variables.current_ma == pytest.approx(21.0)
    assert "device malfunction" in DeviceHandle.status_text(variables.status)
    assert "PV out of limits" in DeviceHandle.status_text(variables.status)

A saturated loop

The value on the loop stops following the process. A controller reading only the current would not know.

def test_saturated_loop(hart_bench):
    hart_bench.devices["PT-101"].faults.inject("saturate", direction="high")
    with hart_bench.client() as client:
        variables = client.handle(0).read_variables()
    assert variables.current_ma == pytest.approx(20.5)
    assert "loop current saturated" in DeviceHandle.status_text(variables.status)

No answer

HART hosts retry. Make sure yours does, and that a device coming back is noticed.

def test_retry_after_no_answer(hart_bench):
    with hart_bench.client(timeout=0.3) as client:
        device = client.handle(0)
        hart_bench.devices["PT-101"].faults.inject("no_response", count=1)
        with pytest.raises(HartTimeout):
            device.read_variables()
        assert device.read_variables().current_ma > 0

Busy

Response code 32 is not an error of the request, and nothing changed. Send it again later.

def test_busy_changes_nothing(hart_bench):
    with hart_bench.client() as client:
        device = client.handle(0)
        before = device.read_range()
        hart_bench.devices["PT-101"].faults.inject("busy", count=1)
        assert device.write_range(0, 600, unit=before[2]) == 32
        assert device.read_range() == before
        assert device.write_range(0, 600, unit=before[2]) == 0

A communication error

The device reports a problem with the frame it received, instead of a response code. 0x10 is a framing error in the example below. The client raises it as a HartError.

def test_communication_error(hart_bench):
    with hart_bench.client() as client:
        device = client.handle(0)
        hart_bench.devices["PT-101"].faults.inject("comm_error", flags=0x10, count=1)
        with pytest.raises(HartError, match="communication error"):
            device.read_variables()
        assert device.read_variables().current_ma > 0

Write protect

Response code 7 means the write was not done. Show that to the user instead of assuming the value is set.

def test_write_protect(hart_bench):
    hart_bench.devices["PT-101"].faults.inject("write_protect", on=True)
    with hart_bench.client() as client:
        device = client.handle(0)
        before = device.read_range()
        assert device.write_range(0, 600) == 7
        assert device.read_range() == before

More status available

The bit stays set until a host reads command 48. If yours never does, it never learns the details.

def test_more_status_until_command_48(hart_bench):
    faults = hart_bench.devices["PT-101"].faults
    faults.inject("more_status", bytes=[1, 2, 3, 4, 5, 6])
    with hart_bench.client() as client:
        device = client.handle(0)
        assert "more status available" in DeviceHandle.status_text(device.read_variables().status)
        frame = client.command(device.address, 48)
        assert frame.response_code == 0
        assert frame.data[:6] == bytes([1, 2, 3, 4, 5, 6])
        assert "more status available" not in DeviceHandle.status_text(device.read_variables().status)

A cold start

The device has restarted, so what you cached about it may be wrong. Read its configuration again. The bit is reported once per master: the first answer clears it, so a host that ignores that answer has missed it.

def test_cold_start_is_reported_once(hart_bench):
    with hart_bench.client() as client:
        device = client.handle(0)
        hart_bench.devices["PT-101"].faults.inject("cold_start")
        assert "cold start" in DeviceHandle.status_text(device.read_variables().status)
        assert "cold start" not in DeviceHandle.status_text(device.read_variables().status)

A drifting value

Everything looks fine: good status, no alarm, only a number that slowly moves away. clear stops the drift but keeps the offset so far.

def test_drift_has_good_status(hart_bench_factory):
    bench = hart_bench_factory({"devices": [{
        "profile": "pressure", "polling_address": 0, "device_id": 1, "tag": "PT-101",
        "pv": {"unit": "kPa", "sensor_limits": [-100, 2500], "range": [0, 1000], "damping": 0,
               "source": {"type": "constant", "value": 400}}}]})
    import time
    bench.devices["PT-101"].faults.inject("drift", per_second=100)
    with bench.client() as client:
        device = client.handle(0)
        first = device.read_variables().pv
        time.sleep(0.5)
        second = device.read_variables()
    assert second.pv > first + 20
    assert "device malfunction" not in DeviceHandle.status_text(second.status)