Documentation Index

Fetch the complete documentation index at: https://docs.roboticks.io/llms.txt

Use this file to discover all available pages before exploring further.

​

Fault injection reference

from roboticks.fault_injection import (
    drop_messages,
    delay_messages,
    kill_node,
    corrupt_topic,
)

​

drop_messages

​

Signature

def drop_messages(
    topic: str,
    *,
    rate: float,
    seed: int | None = None,
) -> ContextManager[None]: ...

​

Behaviour

• Installs an interceptor on the topic at context entry; uninstalls at exit.
• Drops are evaluated per message, not per second. A rate of 0.5 drops roughly half the messages.
• If multiple drop_messages blocks nest on the same topic, the product of rates applies (composition is multiplicative).
• Out-of-range rate raises ValueError at entry.

​

Example

with drop_messages("/scan", rate=0.3):
    # 30% of /scan messages are dropped during this block
    run_test_body()

​

delay_messages

​

Signature

def delay_messages(
    topic: str,
    *,
    ms: int,
    jitter_ms: int = 0,
) -> ContextManager[None]: ...

​

Behaviour

• Each intercepted message is held for ms ± jitter_ms/2 before being released to subscribers.
• Holding is implemented as a threading.Timer per message. High-rate topics with high ms will hold many in-flight messages; budget memory accordingly.
• Order is preserved (per-topic FIFO).

​

Example

with delay_messages("/scan", ms=300, jitter_ms=50):
    # Each /scan message held 275-325 ms
    run_test_body()

​

kill_node

​

Signature

def kill_node(
    name: str,
    *,
    signal: int = signal.SIGTERM,
    wait_for_exit: float = 2.0,
) -> ContextManager[NodeKilledHandle]: ...

​

Returns (as context value)

​

Raises

• LookupError — no node with that name visible on the ROS graph at entry.
• RuntimeError — process refused to exit within wait_for_exit.

​

Example

with kill_node("/nav2_planner") as killed:
    # planner is down here
    time.sleep(3.0)
    assert killed.respawned, "watchdog did not bring the planner back"

​

corrupt_topic

​

Signature

def corrupt_topic(
    topic: str,
    *,
    mutator: Callable[[Any], Any],
) -> ContextManager[None]: ...

​

Behaviour

• The mutator runs between the publisher and subscribers — every subscriber sees the mutated value.
• Mutator exceptions propagate up to the publishing thread and abort the test. Catch inside the mutator if you want resilient behaviour.
• The mutator may mutate in place and return the same object, or return a fresh one.

​

Example

from sensor_msgs.msg import LaserScan

def clamp_max(msg: LaserScan) -> LaserScan:
    msg.ranges = [min(r, 1.0) for r in msg.ranges]
    return msg

with corrupt_topic("/scan", mutator=clamp_max):
    # planner sees a LIDAR that never reads beyond 1 m
    run_test_body()

​

Composition

with mcap_capture(topics=["/scan", "/cmd_vel"]):
    with delay_messages("/scan", ms=200):
        with drop_messages("/scan", rate=0.05):
            run_test_body()

​

DDS implementation notes

• Implemented via rclpy’s middleware plug-points — no DDS vendor extensions required.
• Per-test isolated: the interceptor installs into the current process only; other processes in a launch_testing graph see the un-faulted topic.
• For across-process fault injection (e.g. fault a topic between two separately-launched nodes), use launch_testing integration and inject from inside the test process that owns the topic.

​

Limits

​

Next