Add capability for testing actions.

[vik748]

Per discussion in #25 this PR adds the capability to test actions with a single call like:

    goal_handle, feedbacks, result_response = env.send_action_goal_and_wait_for_response(
        name="fibonacci",
        action_type=Fibonacci,
        goal_msg=Fibonacci.Goal(order=4),
    )

I couldn't figure out how to improve typing hints, so have a bunch of Any types.
Also I couldn't figure out how to auto-infer action type similar to _get_service_client, could use help with that.

[vik748]

@felixdivo PTAL.

[felixdivo]

Hey @vik748, I know I owe you a few comments, also regarding mock testing. Sorry for taking so long. I'll probably respond by the end of the week. I really value your efforts, and at first glance, it looked great. I'll have to find a few more minutes to go through it more slowly, though, hopefully this weekend. :)

[felixdivo]

Looks great so far!

Regarding the many Anys: I think this cannot be avoided since ROS does not model special types for actions. For documentation, you could add an alias similar to RosMessage = Any.

[felixdivo]

Regarding the automatic detection of action type. We always get a goal, e.g. Fibonacci.Goal(order=4). Can't we then parse and import the "main" Type (in this case Fibonacci) ver similar to how it is done for services in _get_service_client?

This might works if you pass in goal_class = Fibonacci.Goal:

# Get the type of the service
# This is a bit tricky but relieves the user from passing it explicitly
module = import_module(goal_class.__module__)
# We cut away the trailing "_Goal" from the type name, which has length 5
base_type_name = goal_class.__name__[:-5]
base_type_class: Type = getattr(module, base_type_name)

[felixdivo]

Great work so far!

[felixdivo]

@vik748 I am sorry for my slow responses. I will make sure to answer at least on a weekly basis from May one!

[Timple]

Feedback can happen a lot! So this print might become very verbose.

[felixdivo]

Suggested change

print(f"{feedback_msg.feedback=}")

[Timple]

For the record, this is how I do this kind of stuff now. It's an implementation without sleeps, so perhaps can serve as inspiration:

from action_msgs.srv import CancelGoal
from my_package.action import MyAction

def test():
    client = ActionClient(node, MyAction, "my_action")
    assert client.wait_for_server(10.0), "No actionclient for my_action found"

    event = threading.Event()

    def unblock(future):
        nonlocal event
        event.set()

    send_goal_future = self.send_goal_async(goal, **kwargs)
    send_goal_future.add_done_callback(unblock)

    event.wait()
    goal_handle = send_goal_future.result()
    assert goal_handle.accepted

    print("Canceling action")
    response: CancelGoal.Response = goal_handle.cancel_goal()
    assert response.return_code == CancelGoal.Response.ERROR_NONE, "Canceling action failed"

[felixdivo]

event.wait()

You could even pass a timeout to it.

[felixdivo]

I'll react to PRs etc. more regularly from now on.

[felixdivo]

Hey @vik748, after finally completing #33 and many minor cleanups, I was able to have a closer look at this. I was inspired by the ROS source code and @Timple's very similar suggestions. Since I did not want to bother you with more requests, I implemented the changes myself in #37. Thank you for the great work here and your patience. :)