PyHamcrest-based helpers for polling assertions.

Optional PyHamcrest-based helpers for polling assertions in async tests. Requires the hamcrest extras:

pip install remotivelabs-topology[hamcrest]

remotivelabs.topology.testing.hamcrest.await_at_most provides a fluent retry API that repeatedly calls a function until its return value satisfies a matcher or the timeout is reached:

from remotivelabs.broker import BrokerClient, FrameSubscription, SignalValue

from remotivelabs.topology.namespaces.can import CanNamespace
from remotivelabs.topology.testing.hamcrest import await_at_most, contains_signal_value_sequence


async def test_signal_sequence(broker_client: BrokerClient) -> None:
    async with CanNamespace("HazardLightControlUnit-DriverCan0", broker_client) as ns:
        stream = await ns.subscribe_frames(FrameSubscription(name="HazardLightButton"))
        values: list[SignalValue] = []

        async def consume_next() -> list[SignalValue]:
            frame = await anext(stream)
            values.append(frame.signals["HazardLightButton.HazardLightButton"])
            return values

        # Assert the button transitions 0 → 1 → 0 within 5 seconds
        await await_at_most(5).until(consume_next, contains_signal_value_sequence(0, 1, 0))

Signal value matchers:

• remotivelabs.topology.testing.hamcrest.contains_signal_value_sequence — passes if a list contains the given values as a contiguous subsequence (consecutive duplicates are collapsed).
• remotivelabs.topology.testing.hamcrest.ends_with_signal_value_sequence — passes if a list ends with the given sequence.
• remotivelabs.topology.testing.hamcrest.contains_signal_value_sequences — passes if each key in a dict maps to a list containing its expected subsequence.
• remotivelabs.topology.testing.hamcrest.has_last_element — passes if the last element of a list satisfies a nested matcher.