Using this package for action tests.

[vik748]

Hi,
I am trying to use the ROS2TestEnvironment to help write unit tests for an action server. I think I have a basic test working. I believe this can be easily generalized and added to the ROS2TestEnvironment class. The code looks like:

class FibonacciActionClient:
    env: ROS2TestEnvironment
    action_client: ActionClient
    send_goal_future: Optional[Future]
    get_result_future: Optional[Future]
    goal_response = Optional[ClientGoalHandle]
    feedbacks = List[Fibonacci.Feedback]
    result = Optional[Fibonacci.Result]
    # A MutuallyExclusive CallbackGroup is required to force all the callbacks:
    # goal_response, feedback and goal_done to run sequentially.
    # Without this we get a race condition between goal_response and feedback.
    action_client_cb_group = CallbackGroup

    def __init__(self, env):
        self.env = env
        self.action_client_cb_group = MutuallyExclusiveCallbackGroup()
        self.action_client = ActionClient(
            env, Fibonacci, "fibonacci", callback_group=self.action_client_cb_group
        )
        self.goal_response = None
        self.send_goal_future = None
        self.get_result_future = None
        self.feedbacks = []
        self.result = None

    def feedback_callback(self, feedback_msg: Fibonacci.Feedback):
        feedback = feedback_msg.feedback  # pyright: ignore[reportAttributeAccessIssue]
        self.feedbacks.append(feedback)  # pyright: ignore[reportCallIssue]

    def goal_response_callback(self, future: Future):
        goal_handle = future.result()
        assert goal_handle is not None
        if not goal_handle.accepted:
            self.env.get_logger().info("Goal rejected :(")
            return

        self.env.get_logger().info("Goal accepted :)")

        self.get_result_future = goal_handle.get_result_async()

    def send_goal_and_wait_for_send_future(self, goal_msg: Fibonacci.Goal):
        timeout_availability = 2.0
        is_ready = self.action_client.wait_for_server()
        if not is_ready:
            raise TimeoutError(
                f"Action server did not become ready within {timeout_availability} seconds"
            )
        self.send_goal_future = self.action_client.send_goal_async(
            goal_msg, feedback_callback=self.feedback_callback
        )
        self.send_goal_future.add_done_callback(self.goal_response_callback)

        assert self.env.executor is not None
        self.env.executor.spin_until_future_complete(
            self.send_goal_future, timeout_sec=2.0
        )
        if self.send_goal_future.done():
            self.goal_response = self.send_goal_future.result()
        else:
            raise TimeoutError(f"Future did not complete within {2.0} seconds")

    def spin_until_goal_done(self):
        assert self.env.executor is not None
        assert self.get_result_future is not None

        GET_RESULT_TIMEOUT_S = 2.0
        self.env.executor.spin_until_future_complete(
            self.get_result_future, timeout_sec=GET_RESULT_TIMEOUT_S
        )
        if self.get_result_future.done():
            self.result = self.get_result_future.result()
        else:
            raise TimeoutError(
                f"Future did not complete within {GET_RESULT_TIMEOUT_S} seconds"
            )

        self.env.executor.spin_once()


@with_single_node(FibonacciActionServer)
def test_fibonacci_action(env: ROS2TestEnvironment) -> None:
    """Test action."""
    client = FibonacciActionClient(env)
    client.send_goal_and_wait_for_send_future(Fibonacci.Goal(order=25))

    assert client.env.executor is not None

    # Spin so we have a goal_response_callback and get_result_future is set.
    while client.get_result_future is None:
        client.env.executor.spin_once()

    client.spin_until_goal_done()

    assert client.feedbacks is not None
    assert client.result is not None
    print(f"{client.feedbacks=}")
    print(f"{client.result=}")

I imagine as part of ros2-test-easy this could be simplified to a single call like:
response, feedback, done_result = env.call_action("fibonacci", action_goal)

WDYT

[felixdivo]

Yes, this sounds useful! We don't use actions right now over at my group, so I haven't yet bothered implementing it. But sure, contributions are welcome. 😃

As a side note: This should already be possible, by simply following this tutorial and using the fact that the env: ROS2TestEnvironment is a Node instance. See also here.

We should think about this. Would it be a convenience function for calling the action "synchronously", i.e., in a way that only returns if the action is completed fully? I think that sounds reasonable.

Additionally, we might consider adding an example of how to create a service and an action and call it normally by using the env. This might be a bit too hidden?

[felixdivo]

Additionally, we might consider adding an example of how to create a service and an action and call it normally by using the env. This might be a bit too hidden?

You could have a look at #26, if you want to. @vik748

[felixdivo]

Done in #37.