Enhance integration testing guide with WaitForTopics feature

[LastStarDust]

Updated integration testing documentation to use the new WaitForTopics feature

Description

This PR addresses issue #5249 by updating the integration testing tutorial to showcase the WaitForTopics utility from launch_testing_ros.

• Replaced manual subscriber logic with WaitForTopics: The original example used manual rclpy.create_subscription() and spin_once() loops. The updated tutorial demonstrates the WaitForTopics approach, which is cleaner and serves as a better learning example for users.

• No more magic numbers: Removed the problematic 0.5-second TimerAction before ReadyToTest(). Tests now use event-driven waiting via PublisherEventCallbacks instead of "sleep" hacks. Moreover, it checks for specific velocity conditions.

• Added trigger callback example: Included an example of the trigger function, demonstrating how to publish control messages and verify robot response using the new trigger parameter of WaitForTopics. This was already proposed in the original document.

Did you use Generative AI?

Not for the source code and only partially for the text. But I double-checked each line of text.

Additional Information

• All code examples have been tested in this sample repository.

• I added a new section on Python-only package setup.

[asymingt]

@Shru can you take a look at this please?

[fujitatomoya]

@LastStarDust a couple of workflows are failing, can you check them? besides, i think we need to rebase this PR.

[Shru]

Thanks for putting this together! Adding documentation and examples for WaitForTopics and EnableRmwIsolation is a great addition to the integration testing guide.

I tested the tutorial code locally and noticed a couple of areas where the test patterns could be made more robust:

1. Bounded Timeout in test_moves_with_triggered_twist

In section 1.3.2, the test currently uses an unbounded while True: loop to wait for matching velocities:

while True:
    assert waiter.wait(linear_x=10, angular_z=2 * math.pi)
    ...
    if any(math.isclose(...) for p in poses):
        break

Because turtlesim streams pose messages continuously, waiter.wait() returns True on every iteration. If the node fails to reach the target velocity (e.g., a dropped packet during DDS discovery, a velocity limit clamp, or callback delays), the test hangs indefinitely instead of failing cleanly.

Adding an explicit deadline ensures the test fails fast with an actionable AssertionError and emits a proper XUnit report in CI:

def test_moves_with_triggered_twist(self) -> None:
    """Verify turtle motion after triggering Twist publication."""
    import time
    waiter = WaitForTopics([("turtle1/pose", Pose)], trigger=trigger_publish_twist)
    deadline = time.time() + 5.0  # 5-second bounded timeout

    try:
        while time.time() < deadline:
            assert waiter.wait(linear_x=10, angular_z=2 * math.pi)
            assert waiter.topics_received() == {"turtle1/pose"}
            poses = waiter.received_messages("turtle1/pose")
            assert len(poses) >= 1
            if any(
                math.isclose(p.linear_velocity, 10.0, rel_tol=1e-2)
                and math.isclose(p.angular_velocity, 2 * math.pi, rel_tol=1e-2)
                for p in poses
            ):
                return
        self.fail("Timed out waiting for expected turtle velocity after Twist trigger")
    finally:
        waiter.shutdown()

2. Publisher Lifecycle in trigger_publish_twist

In trigger_publish_twist, destroying and recreating the publisher on every invocation forces DDS to unregister endpoints and renegotiate discovery handshakes on each loop iteration.

Checking if the publisher already exists on the node and reusing it avoids unnecessary discovery churn and keeps execution deterministic:

def trigger_publish_twist(node: Node, linear_x: float, angular_z: float) -> None:
    """Publish Twist commands to trigger turtlesim pose updates."""
    # Create publisher once if it does not already exist on the node
    if not hasattr(node, "cmd_vel_publisher"):
        matched_event = threading.Event()

        def on_subscriber_matched(info: Any) -> None:
            if info.current_count > 0:
                matched_event.set()

        node.cmd_vel_publisher = node.create_publisher(
            Twist,
            "turtle1/cmd_vel",
            10,
            event_callbacks=PublisherEventCallbacks(matched=on_subscriber_matched),
        )

        if not matched_event.wait(timeout=5.0):
            raise RuntimeError("Timed out waiting for turtlesim cmd_vel subscriber")

    # Reuse established publisher to send command
    msg = Twist()
    msg.linear.x = float(linear_x)
    msg.angular.z = float(angular_z)
    node.cmd_vel_publisher.publish(msg)

Aside from these two points, the structure and explanations in the guide look great!

[LastStarDust]

@Shru Thank you so much for reviewing the code. I was aware of those shortcomings but I chose the most concise solution for brevity sake as this is just an example and not production code. But of course I have nothing against including your suggestions if @fujitatomoya and @christophebedard agree.

[fujitatomoya]

but I chose the most concise solution for brevity sake as this is just an example and not production code

i second this. this is tutorial and example, that i think what's important here is easy to see and understand.

[Shru]

@fujitatomoya @LastStarDust makes sense for a tutorial context, thanks! @LastStarDust could you rebase on rolling and push so CI can rerun?

[Shru]

@christophebedard The conflicts are resolved. Could you please approve the workflow run and review/approve when you have a moment?