Fault injection recipes

What a client should do when a device or server misbehaves, and how to provoke it. Every recipe below is a real test against opcua-sim 0.1.0 with asyncua 2.1. They share this helper:

import asyncio

import pytest
from asyncua import Client, ua
from asyncua.ua import uaerrors

NS = "urn:steitec:opcua-sim"


def with_client(url, work, timeout=2):
    async def go():
        async with Client(url, timeout=timeout) as client:
            idx = await client.get_namespace_index(NS)
            return await work(client, idx)
    return asyncio.run(go())

A Bad value has no number

read_value() raises on a Bad status. To see the status, read the whole DataValue, and do not carry on with the last number as if nothing happened. In the Browser demo, a broken sensor looks like this:

def test_bad_status_has_no_value(opcua_bench_factory):
    bench = opcua_bench_factory({"devices": [{"name": "pt", "profile": "pressure-transmitter"}]})
    bench.inject("pt", "sensor_break")

    async def work(client, idx):
        return await client.get_node(ua.NodeId("pt.Pressure", idx)).read_data_value(
            raise_on_bad_status=False)

    data = with_client(bench.url, work)
    assert data.StatusCode.value == ua.StatusCodes.BadSensorFailure
    assert data.Value is None or data.Value.Value is None

Uncertain still has a number

Uncertain keeps its value. Whether to use it is your decision, not an error.

def test_uncertain_keeps_the_value(opcua_bench_factory):
    bench = opcua_bench_factory({"devices": [{"name": "pt", "profile": "pressure-transmitter"}]})
    bench.inject("pt", "bad_status", variable="Pressure", status="UncertainSensorNotAccurate")

    async def work(client, idx):
        return await client.get_node(ua.NodeId("pt.Pressure", idx)).read_data_value(
            raise_on_bad_status=False)

    data = with_client(bench.url, work)
    assert data.StatusCode.value == ua.StatusCodes.UncertainSensorNotAccurate
    assert data.Value.Value is not None

Old data looks perfectly good

With stale, the status stays Good and only the source timestamp gives it away. Compare it with the server timestamp if your application must not act on old values.

def test_stale_values_are_good_but_old(opcua_bench_factory):
    bench = opcua_bench_factory({"devices": [{"name": "m", "profile": "energy-meter"}]})
    bench.inject("m", "stale", seconds=120)

    async def work(client, idx):
        await asyncio.sleep(0.5)
        return await client.get_node(ua.NodeId("m.VoltageL1", idx)).read_data_value()

    data = with_client(bench.url, work)
    assert data.StatusCode.is_good()
    assert (data.ServerTimestamp - data.SourceTimestamp).total_seconds() > 100

A latched trip

Clearing the cause is not enough: Start stays refused until Reset.

def test_trip_latches_until_reset(opcua_bench):
    bench = opcua_bench
    bench.inject("drive", "trip", code=7)
    bench.clear("drive", "trip")

    async def work(client, idx):
        methods = client.get_node(ua.NodeId("drive.MethodSet", idx))
        with pytest.raises(uaerrors.BadInvalidState):
            await methods.call_method(ua.NodeId("drive.Start", idx))
        await methods.call_method(ua.NodeId("drive.Reset", idx))
        await methods.call_method(ua.NodeId("drive.Start", idx))

    with_client(bench.url, work)

Refused writes

Check the result of every write. A value that was refused was not written.

def test_write_refused(opcua_bench):
    opcua_bench.inject("drive", "reject_write", status="BadUserAccessDenied")

    async def work(client, idx):
        node = client.get_node(ua.NodeId("drive.SpeedSetpoint", idx))
        with pytest.raises(uaerrors.BadUserAccessDenied):
            await node.write_value(ua.DataValue(ua.Variant(10.0, ua.VariantType.Double)))

    with_client(opcua_bench.url, work)

Alarms: acknowledge with the right EventId

An Acknowledge needs the EventId of the latest event. An old one answers BadEventIdUnknown.

class Collect:
    def __init__(self):
        self.events = []

    def event_notification(self, event):
        self.events.append(event.get_event_props_as_fields_dict())


def test_acknowledge(opcua_bench):
    async def work(client, idx):
        handler = Collect()
        subscription = await client.create_subscription(50, handler)
        await subscription.subscribe_events(ua.ObjectIds.Server, ua.ObjectIds.OffNormalAlarmType)
        await asyncio.to_thread(opcua_bench.inject, "drive", "overload")
        await asyncio.sleep(0.5)
        event = handler.events[-1]
        assert event["ConditionName"].Value == "Overload"
        condition = client.get_node(event["NodeId"].Value)
        acknowledge = client.get_node(ua.ObjectIds.AcknowledgeableConditionType_Acknowledge)
        comment = ua.LocalizedText("on it")
        with pytest.raises(uaerrors.BadEventIdUnknown):
            await condition.call_method(
                acknowledge, ua.Variant(b"old", ua.VariantType.ByteString), comment)
        await condition.call_method(acknowledge, event["EventId"], comment)
        await asyncio.sleep(0.5)
        assert handler.events[-1]["AckedState/Id"].Value is True

    with_client(opcua_bench.url, work)

After a reconnect: ConditionRefresh

A new subscription hears about new events only. Alarms that are already active come with ConditionRefresh. The Browser demo shows the same sequence:

def test_condition_refresh(opcua_bench):
    opcua_bench.inject("drive", "overload")

    async def work(client, idx):
        handler = Collect()
        subscription = await client.create_subscription(50, handler)
        await subscription.subscribe_events(ua.ObjectIds.Server, ua.ObjectIds.OffNormalAlarmType)
        await asyncio.sleep(0.3)
        assert handler.events == []
        conditions = client.get_node(ua.ObjectIds.ConditionType)
        refresh = client.get_node(ua.ObjectIds.ConditionType_ConditionRefresh)
        await conditions.call_method(
            refresh, ua.Variant(subscription.subscription_id, ua.VariantType.UInt32))
        await asyncio.sleep(0.5)
        assert [e["ConditionName"].Value for e in handler.events] == ["Overload"]

    with_client(opcua_bench.url, work)

A link that stays open and carries nothing

The connection never breaks, so only your timeout tells you. Make sure you have one.

def test_silent_link_needs_a_timeout(opcua_bench):
    async def work(client, idx):
        await asyncio.to_thread(opcua_bench.inject, "server", "blackhole")
        with pytest.raises((asyncio.TimeoutError, ConnectionError)):
            await client.get_node(ua.NodeId("drive.Speed", idx)).read_value()
        await asyncio.to_thread(opcua_bench.clear, "server", "blackhole")

    async def go():
        client = Client(opcua_bench.url, timeout=0.5)
        await client.connect()
        idx = await client.get_namespace_index(NS)
        try:
            await work(client, idx)
        finally:
            try:
                await client.disconnect()
            except Exception:
                pass                       # the session did not survive, which is the point

    asyncio.run(go())

A server restart ends your session

The port stays the same and the device keeps its state, but sessions and subscriptions are gone. Your client has to connect again and create its subscriptions again.

def test_restart_needs_a_new_session(opcua_bench):
    bench = opcua_bench

    async def write(client, idx):
        await client.get_node(ua.NodeId("drive.SpeedSetpoint", idx)).write_value(
            ua.DataValue(ua.Variant(11.0, ua.VariantType.Double)))

    with_client(bench.url, write)
    port = bench.port
    bench.inject("server", "restart", downtime=0.2)
    bench.wait_restarted()
    assert bench.port == port

    async def read(client, idx):
        return await client.get_node(ua.NodeId("drive.SpeedSetpoint", idx)).read_value()

    assert with_client(bench.url, read) == 11.0

A server that says it is not well

ServiceLevel and ServerState are how a redundancy-aware client decides to switch servers.

def test_server_state_and_service_level(opcua_bench):
    opcua_bench.inject("server", "server_state", state="Suspended")
    opcua_bench.inject("server", "service_level", value=10)

    async def work(client, idx):
        state = client.get_node(ua.ObjectIds.Server_ServerStatus_State)
        level = client.get_node(ua.ObjectIds.Server_ServiceLevel)
        return await state.read_value(), await level.read_value()

    assert with_client(opcua_bench.url, work) == (ua.ServerState.Suspended, 10)