Python equivalent of ros2_cpp_test

[thomasegriffith]

Hi- want to say excellent work on this project; it's great.

Quick question: is there a way to run python tests that don't operate on nodes but still need ament setup? Maybe something akin to ros2_cpp_test but for python?

Thanks!

[ahans]

You probably can use ros2_test for this. The name of the nodes parameter is a bit misleading, those don't need to be nodes in the ROS 2 sense, but simply executable Bazel targets. You can have a look at launch_pytest for an example. Note that there are two flavors of this testing framework: a more recent pytest-based one (used here) and the traditional launch_testing one (used, e.g., for the rosbag tests).

In both launch testing frameworks, you define the "nodes" (which can be plain executables) you want to test in a "launch description" and then actual tests which can interact with the other processes and/or check some aspect of their execution, such as stdout/stderr output or exit codes. The launch test framework will take care of creating and calling code for the ament setup.

If you want to do everything from a single Python file, you can instantiate ros2_ament_setup manually in your BUILD file and then depend on it in your py_test. Something like this:

ros2_ament_setup(
    name = "my_ament_setup",
    idl_deps = [ ... ],
    deps = [ ...],
)

py_test(
    name = "my_test",
    srcs = ["my_test.py"],
    deps = [
        ":my_ament_setup",
    ]
)

And then in my_test.py:

from path.to.my_ament_setup import ament_setup

ament_setup.set_up_ament()

[ahans]

The ros2_ament_setup part does not work any more as of the latest version of rules_ros2. You have to create a py_launcher for this and provide your own template. Looks involved and not exactly straightforward, so not sure that is what you want ...

[thomasegriffith]

Thanks!

For some context: I'm testing a class (living in a library) that reads from a bag using rosbag2_py, and want to make sure it's doing the right thing. So to do that, I have the test first write some synthetic data to a bag in the setup step, and then invoke an object of the class to read from the synthetic bag. Then I have some tests I run on the object. This is all python without any notion of binaries.

In ros2_test, does ament only get setup for the binary that's launched via launch.actions.ExecuteProcess()? If that's the case, it seems like it could be leveraged to write the synthetic data, but the object I want to test is probably in limbo? I ran a quick test and ran into this error if I try to do anything outside of the binary:

[rosbag2_storage]: Unable to create class load instance: Environment variable 'AMENT_PREFIX_PATH' is not set or empty

Your suggestion of using ros2_ament_setup as a dependency to a py_test target definitely seems like the ideal solution here, but sounds like that doesn't work anymore.

Anyway, I have the following workaround, but it requires a ROS install and isn't trivial to setup in CI. I have to use it as a manual test. Still better than nothing I suppose.

source /opt/ros/<version>/setup.bash
printenv | sort
bazel test --test_env=AMENT_PREFIX_PATH=<YOUR AMENT_PREFIX_PATH VALUE> \
                  --test_env=LD_LIBRARY_PATH=<YOUR LD_LIBRARY_PATH VALUE> \
                  --test_env=PYTHON_PATH=<YOUR PYTHON_PATH VALUE> \
                  //path/to/test:target

[ahans]

Your observation is correct, the ament setup happens for the "nodes" you define in the launch description. In the file you give via launch_file it is not available. What you could do is move all your test logic into the "node" and have that return a non-zero exit code upon failure. Then in the launch_file you wait for the "node" to finish and check its exit code to forward the success/failure result to Bazel. I'm fairly confident that would work. However, I'd still prefer the way where you call set_up_ament() manually. The way I described it above it doesn't work anymore, but that's about it. I will see if I can put together an example for you later.

[ahans]

@thomasegriffith The new ros2_py_test macro does pretty much what I would have done manually. Go ahead and use that, it will most probably solve your problem.

[thomasegriffith]

Will do. Thanks for all the help!

[mvukov]

I like to think of ros2_ament_setup as of an impl detail: most folks shouldn't need to deal with this. I created #81 where I implemented ros2_py_test macro.

If you don't involved nodes in a test, ros2_test is an overkill IMO -- takes too much time to get itself set up and you need at least two files instead of one...

[thomasegriffith]

@mvukov that works like a charm! Thanks!

Agreed with @ahans. It would be nice to have it work with py_binary and other rules, too. That's a use-case I wasn't thinking about originally, but I totally have the same problem with some python binaries that process bags.

[mvukov]

Sure, should be easy to extend to a ros2_{cc, py}_binary as well. I don't know whether I'll have time for this in the coming days, but please feel free to work out a PR.

[ahans]

I finally managed to start looking into this. My current plan is to put a ros2_py_binary next to ros2_py_test that does essentially the same, except it's not a test. For C++, we already have ros2_cpp_binary. I would extend that to also accept a set_up_ament parameter and behave similarly. Since the logic for setting up the launcher is identical for all cases (Python and C++), I think there's some potential to reduce redundancies. Not sure how to do that exactly, but first wanted to make sure you agree the overall plan sounds reasonable?

[thomasegriffith]

Sounds reasonable to me!

[mvukov]

I already tried to reduce code-dup with py_launcher/sh_launcher and to me that already looked reasonable. I'd say it probably makes sense to group this by language: factor out common code per language: ros2_cc_{binary, test} and ros2_py_{binary_test}. C++ and Py targets are similar, but, for Python targets you need a symlink target because you cannot do rootpath on a py-exec-target (yet). Sounds good? Something like:

def _ros2_exec_target(target, name, set_up_ament, ...):
  if set_up_ament == False:
    _ros2_cc_target(target, ...)

  target_name = name + "_impl"
  tags = kwargs.pop("tags", None)
  _ros2_cc_target(cc_binary, tags = ["manual"], ...)  # You really need cc_binary here AFAIK.

  # Set up launcher
  # Set up sh_{binary|test}, will need to figure out whether we can extract this from `target` arg.

def ros2_cpp_binary(name, ...):
  _ros2_exec_target(cc_binary, ...)

def ros2_cpp_test(name, ...):
  _ros2_exec_target(cc_test, ...)

It's also OK to rename test.sh.tpl to something like launcher.sh.tpl.

[thomasegriffith]

@mvukov if you use try to write to an mcap bag from a ros2_py_test target, the create_topic() call from a rosbag2_py.SequentialWriter() object will fail with the following:

IndexError: package '<msg package name; nav_msgs in my case>' not found, searching: [path/to/your/test_launch_ament_setup]

sqlite seems to work fine though.

[mvukov]

I think you need to put @ros2_common_interfaces//:py_nav_msgs in deps of your target. If that doesn't help, lemme know. Also, if it doesn't work, please create a min working example.

[thomasegriffith]

Right- I have that as a dep. But if that was a requirement, shouldn't sqlite fail too?

Anyway, I can't seem to push a branch or create a PR (do I need permission?).

Here's a test. No issue from test_write_to_sqlite(), but test_write_to_mcap() raises the above IndexError.

ros2/test/rosbag/BUILD.bazel:

ros2_py_test(
    name = "write_to_bag_test",
    srcs = ["write_to_bag_test.py"],
    deps = [
        "@ros2_rclpy//:rclpy",
        "@ros2_common_interfaces//:py_nav_msgs",
        "@ros2_rosbag2//:rosbag2_py",
        requirement("pytest")
    ],
    main = "write_to_bag_test.py",
    set_up_ament = True,
)

ros2/test/rosbag/write_to_bag_test.py:

"""Pytest for writing to a bag file"""

import pathlib
import pytest

import rosbag2_py

SQLITE3_STORAGE_ID = "sqlite3"
MCAP_STORAGE_ID = "mcap"
ODOMETRY_MESSAGE_TYPE = "nav_msgs/msg/Odometry"
ODOMETRY_TOPIC = "/foo"
SERIALIZATION_FORMAT = "cdr"

def make_bag(bag_filepath: str, storage_id: str):
    """Make a bag with a single message on a single topic.
    
    Args:
        bag_filepath: The path to the bag file to create.
        storage_id: The storage format to use.
    """
    writer = rosbag2_py.SequentialWriter()
    writer.open(
        rosbag2_py.StorageOptions(
            uri=bag_filepath,
            storage_id=storage_id,
        ),
        rosbag2_py.ConverterOptions(input_serialization_format=SERIALIZATION_FORMAT, 
                                    output_serialization_format=SERIALIZATION_FORMAT),
    )
    writer.create_topic(
        rosbag2_py.TopicMetadata(
            name=ODOMETRY_TOPIC, type=ODOMETRY_MESSAGE_TYPE, serialization_format=SERIALIZATION_FORMAT
        )
    )

def test_write_to_sqlite(tmp_path):
    bag_filepath = str(pathlib.Path(tmp_path) / "test.db3")
    make_bag(bag_filepath=bag_filepath, storage_id=SQLITE3_STORAGE_ID)

def test_write_to_mcap(tmp_path):
    bag_filepath = str(pathlib.Path(tmp_path) / "test.mcap")
    make_bag(bag_filepath=bag_filepath, storage_id=MCAP_STORAGE_ID)


if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__]))

[mvukov]

Hi Thomas, thanks for the test. I added it in #85 together with some fixes. You were right, if the deps were not set up correctly, sqlite wouldn't work either. Anyway, some functionality was missing and I added it.

[mvukov]

BTW, if you want to submit a PR: please fork the repo, make changes in your fork and then open a PR against this repo.

[thomasegriffith]

Great! Thank you!!