diff --git a/tests/test_cameras.py b/tests/test_cameras.py index 7b9d04eb..a2f2a42c 100644 --- a/tests/test_cameras.py +++ b/tests/test_cameras.py @@ -25,11 +25,7 @@ import numpy as np import pytest from lerobot.common.robot_devices.utils import RobotDeviceAlreadyConnectedError, RobotDeviceNotConnectedError -from tests.utils import ( - TEST_CAMERA_TYPES, - make_camera, - mock_camera_or_skip_test_when_not_available, -) +from tests.utils import TEST_CAMERA_TYPES, make_camera, require_camera # Maximum absolute difference between two consecutive images recored by a camera. # This value differs with respect to the camera. @@ -41,7 +37,8 @@ def compute_max_pixel_difference(first_image, second_image): @pytest.mark.parametrize("camera_type, mock", TEST_CAMERA_TYPES) -def test_camera(monkeypatch, camera_type, mock): +@require_camera +def test_camera(request, camera_type, mock): """Test assumes that `camera.read()` returns the same image when called multiple times in a row. So the environment should not change (you shouldnt be in front of the camera) and the camera should not be moving. @@ -51,8 +48,6 @@ def test_camera(monkeypatch, camera_type, mock): # TODO(rcadene): measure fps in nightly? # TODO(rcadene): test logs - mock_camera_or_skip_test_when_not_available(monkeypatch, camera_type, mock) - # Test instantiating camera = make_camera(camera_type) @@ -146,10 +141,9 @@ def test_camera(monkeypatch, camera_type, mock): @pytest.mark.parametrize("camera_type, mock", TEST_CAMERA_TYPES) -def test_save_images_from_cameras(tmpdir, monkeypatch, camera_type, mock): +@require_camera +def test_save_images_from_cameras(tmpdir, request, camera_type, mock): # TODO(rcadene): refactor - mock_camera_or_skip_test_when_not_available(monkeypatch, camera_type, mock) - if camera_type == "opencv": from lerobot.common.robot_devices.cameras.opencv import save_images_from_cameras elif camera_type == "intelrealsense": diff --git a/tests/test_control_robot.py b/tests/test_control_robot.py index 6bec8d03..efb32419 100644 --- a/tests/test_control_robot.py +++ b/tests/test_control_robot.py @@ -31,18 +31,12 @@ from lerobot.common.policies.factory import make_policy from lerobot.common.utils.utils import init_hydra_config from lerobot.scripts.control_robot import calibrate, get_available_arms, record, replay, teleoperate from tests.test_robots import make_robot -from tests.utils import ( - DEFAULT_CONFIG_PATH, - DEVICE, - TEST_ROBOT_TYPES, - mock_robot_or_skip_test_when_not_available, -) +from tests.utils import DEFAULT_CONFIG_PATH, DEVICE, TEST_ROBOT_TYPES, require_robot @pytest.mark.parametrize("robot_type, mock", TEST_ROBOT_TYPES) -def test_teleoperate(monkeypatch, robot_type, mock): - mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock) - +@require_robot +def test_teleoperate(request, robot_type, mock): robot = make_robot(robot_type) teleoperate(robot, teleop_time_s=1) teleoperate(robot, fps=30, teleop_time_s=1) @@ -51,18 +45,16 @@ def test_teleoperate(monkeypatch, robot_type, mock): @pytest.mark.parametrize("robot_type, mock", TEST_ROBOT_TYPES) -def test_calibrate(monkeypatch, robot_type, mock): - mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock) - +@require_robot +def test_calibrate(request, robot_type, mock): robot = make_robot(robot_type) calibrate(robot, arms=get_available_arms(robot)) del robot @pytest.mark.parametrize("robot_type, mock", TEST_ROBOT_TYPES) -def test_record_without_cameras(tmpdir, monkeypatch, robot_type, mock): - mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock) - +@require_robot +def test_record_without_cameras(tmpdir, request, robot_type, mock): root = Path(tmpdir) repo_id = "lerobot/debug" @@ -82,9 +74,8 @@ def test_record_without_cameras(tmpdir, monkeypatch, robot_type, mock): @pytest.mark.parametrize("robot_type, mock", TEST_ROBOT_TYPES) -def test_record_and_replay_and_policy(tmpdir, monkeypatch, robot_type, mock): - mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock) - +@require_robot +def test_record_and_replay_and_policy(tmpdir, request, robot_type, mock): env_name = "koch_real" policy_name = "act_koch_real" diff --git a/tests/test_motors.py b/tests/test_motors.py index 3a034a61..672ecc6f 100644 --- a/tests/test_motors.py +++ b/tests/test_motors.py @@ -31,22 +31,17 @@ import numpy as np import pytest from lerobot.common.robot_devices.utils import RobotDeviceAlreadyConnectedError, RobotDeviceNotConnectedError -from tests.utils import ( - TEST_MOTOR_TYPES, - make_motors_bus, - mock_input, - mock_motor_or_skip_test_when_not_available, -) +from tests.utils import TEST_MOTOR_TYPES, make_motors_bus, mock_input, require_motor @pytest.mark.parametrize("motor_type, mock", TEST_MOTOR_TYPES) -def test_find_port(monkeypatch, motor_type, mock): - mock_motor_or_skip_test_when_not_available(monkeypatch, motor_type, mock) - +@require_motor +def test_find_port(request, motor_type, mock): from lerobot.common.robot_devices.motors.dynamixel import find_port if mock: # To run find_port without user input + monkeypatch = request.getfixturevalue("monkeypatch") monkeypatch.setattr("builtins.input", mock_input) with pytest.raises(OSError): @@ -56,11 +51,11 @@ def test_find_port(monkeypatch, motor_type, mock): @pytest.mark.parametrize("motor_type, mock", TEST_MOTOR_TYPES) -def test_configure_motors_all_ids_1(monkeypatch, motor_type, mock): - mock_motor_or_skip_test_when_not_available(monkeypatch, motor_type, mock) - +@require_motor +def test_configure_motors_all_ids_1(request, motor_type, mock): if mock: # To run find_port without user input + monkeypatch = request.getfixturevalue("monkeypatch") monkeypatch.setattr("builtins.input", mock_input) input("Are you sure you want to re-configure the motors? Press enter to continue...") @@ -80,9 +75,8 @@ def test_configure_motors_all_ids_1(monkeypatch, motor_type, mock): @pytest.mark.parametrize("motor_type, mock", TEST_MOTOR_TYPES) -def test_motors_bus(monkeypatch, motor_type, mock): - mock_motor_or_skip_test_when_not_available(monkeypatch, motor_type, mock) - +@require_motor +def test_motors_bus(request, motor_type, mock): motors_bus = make_motors_bus(motor_type) # Test reading and writting before connecting raises an error diff --git a/tests/test_robots.py b/tests/test_robots.py index be650616..cd49049f 100644 --- a/tests/test_robots.py +++ b/tests/test_robots.py @@ -29,16 +29,15 @@ import pytest import torch from lerobot.common.robot_devices.utils import RobotDeviceAlreadyConnectedError, RobotDeviceNotConnectedError -from tests.utils import TEST_ROBOT_TYPES, make_robot, mock_robot_or_skip_test_when_not_available +from tests.utils import TEST_ROBOT_TYPES, make_robot, require_robot @pytest.mark.parametrize("robot_type, mock", TEST_ROBOT_TYPES) -def test_robot(tmpdir, monkeypatch, robot_type, mock): +@require_robot +def test_robot(tmpdir, request, robot_type, mock): # TODO(rcadene): measure fps in nightly? # TODO(rcadene): test logs # TODO(rcadene): add compatibility with other robots - mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock) - from lerobot.common.robot_devices.robots.manipulator import ManipulatorRobot if robot_type == "aloha" and mock: diff --git a/tests/utils.py b/tests/utils.py index ea10edbd..cff1970e 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -185,52 +185,134 @@ def require_package(package_name): return decorator -def mock_robot_or_skip_test_when_not_available(monkeypatch, robot_type, mock): - if mock: +def require_robot(func): + """ + Decorator that skips the test if a robot is not available + + The decorated function must have two arguments `request` and `robot_type`. + + Example of usage: + ```python + @pytest.mark.parametrize( + "robot_type", ["koch", "aloha"] + ) + @require_robot + def test_require_robot(request, robot_type): + pass + ``` + """ + + @wraps(func) + def wrapper(*args, **kwargs): + # Access the pytest request context to get the mockeypatch fixture + request = kwargs.get("request") + robot_type = kwargs.get("robot_type") + mock = kwargs.get("mock") + + if robot_type is None: + raise ValueError("The 'robot_type' must be an argument of the test function.") + if request is None: + raise ValueError("The 'request' fixture must be an argument of the test function.") + if mock is None: + raise ValueError("The 'mock' variable must be an argument of the test function.") + + if robot_type not in available_robots: + raise ValueError( + f"The camera type '{robot_type}' is not valid. Expected one of these '{available_robots}" + ) + # Run test with a monkeypatched version of the robot devices. - # TODO(rcadene): redesign mocking to not have this hardcoded logic - if robot_type in ["koch", "koch_bimanual"]: - camera_type = "opencv" - elif robot_type == "aloha": - camera_type = "intelrealsense" - else: - camera_type = "all" + if mock: + # TODO(rcadene): redesign mocking to not have this hardcoded logic + if robot_type in ["koch", "koch_bimanual"]: + camera_type = "opencv" + elif robot_type == "aloha": + camera_type = "intelrealsense" + else: + camera_type = "all" + mock_cameras(request, camera_type) + mock_motors(request) - mock_cameras(monkeypatch, camera_type) - mock_motors(monkeypatch) + # To run calibration without user input + monkeypatch = request.getfixturevalue("monkeypatch") + monkeypatch.setattr("builtins.input", mock_input) - # To run calibration without user input - monkeypatch.setattr("builtins.input", mock_input) - - elif not is_robot_available(robot_type): # Run test with a real robot. Skip test if robot connection fails. - pytest.skip(f"A {robot_type} robot is not available.") + else: + if not is_robot_available(robot_type): + pytest.skip(f"A {robot_type} robot is not available.") + + return func(*args, **kwargs) + + return wrapper -def mock_camera_or_skip_test_when_not_available(monkeypatch, camera_type, mock): - if camera_type not in available_cameras: - raise ValueError( - f"The camera type '{camera_type}' is not valid. Expected one of these '{available_cameras}" - ) - if mock: - # Run test with a monkeypatched version of the cameras - mock_cameras(monkeypatch, camera_type) - elif not is_camera_available(camera_type): - # Run test with a real camera. Skip test if camera connection fails - pytest.skip(f"A {camera_type} camera is not available.") +def require_camera(func): + @wraps(func) + def wrapper(*args, **kwargs): + # Access the pytest request context to get the mockeypatch fixture + request = kwargs.get("request") + camera_type = kwargs.get("camera_type") + mock = kwargs.get("mock") + + if request is None: + raise ValueError("The 'request' fixture must be an argument of the test function.") + if camera_type is None: + raise ValueError("The 'camera_type' must be an argument of the test function.") + if mock is None: + raise ValueError("The 'mock' variable must be an argument of the test function.") + + if camera_type not in available_cameras: + raise ValueError( + f"The camera type '{camera_type}' is not valid. Expected one of these '{available_cameras}" + ) + + # Run test with a monkeypatched version of the robot devices. + if mock: + mock_cameras(request, camera_type) + + # Run test with a real robot. Skip test if robot connection fails. + else: + if not is_camera_available(camera_type): + pytest.skip(f"A {camera_type} camera is not available.") + + return func(*args, **kwargs) + + return wrapper -def mock_motor_or_skip_test_when_not_available(monkeypatch, motor_type, mock): - if motor_type not in available_motors: - raise ValueError( - f"The motor type '{motor_type}' is not valid. Expected one of these '{available_motors}" - ) - if mock: - # Run test with a monkeypatched version of the motors - mock_motors(monkeypatch) - elif not is_motor_available(motor_type): - # Run test with a real motor. Skip test if motor connection fails - pytest.skip(f"A {motor_type} motor is not available.") +def require_motor(func): + @wraps(func) + def wrapper(*args, **kwargs): + # Access the pytest request context to get the mockeypatch fixture + request = kwargs.get("request") + motor_type = kwargs.get("motor_type") + mock = kwargs.get("mock") + + if request is None: + raise ValueError("The 'request' fixture must be an argument of the test function.") + if motor_type is None: + raise ValueError("The 'motor_type' must be an argument of the test function.") + if mock is None: + raise ValueError("The 'mock' variable must be an argument of the test function.") + + if motor_type not in available_motors: + raise ValueError( + f"The motor type '{motor_type}' is not valid. Expected one of these '{available_motors}" + ) + + # Run test with a monkeypatched version of the robot devices. + if mock: + mock_motors(request) + + # Run test with a real robot. Skip test if robot connection fails. + else: + if not is_motor_available(motor_type): + pytest.skip(f"A {motor_type} motor is not available.") + + return func(*args, **kwargs) + + return wrapper def mock_input(text=None): @@ -238,8 +320,9 @@ def mock_input(text=None): print(text) -def mock_cameras(monkeypatch, camera_type="all"): +def mock_cameras(request, camera_type="all"): # TODO(rcadene): Redesign the mocking tests + monkeypatch = request.getfixturevalue("monkeypatch") if camera_type in ["opencv", "all"]: try: @@ -276,8 +359,9 @@ def mock_cameras(monkeypatch, camera_type="all"): ) -def mock_motors(monkeypatch): +def mock_motors(request): # TODO(rcadene): Redesign the mocking tests + monkeypatch = request.getfixturevalue("monkeypatch") try: import dynamixel_sdk @@ -349,8 +433,6 @@ def is_robot_available(robot_type): config_path = ROBOT_CONFIG_PATH_TEMPLATE.format(robot=robot_type) robot_cfg = init_hydra_config(config_path) robot = make_robot(robot_cfg) - print("DEBUG", robot.leader_arms) - print("DEBUG", robot.cameras) robot.connect() del robot return True