forked from tangger/lerobot
move mock_motor in test_motors.py
This commit is contained in:
@@ -26,13 +26,14 @@ pytest -sx 'tests/test_motors.py::test_motors_bus[dynamixel-True]'
|
|||||||
# TODO(rcadene): add compatibility with other motors bus
|
# TODO(rcadene): add compatibility with other motors bus
|
||||||
|
|
||||||
import time
|
import time
|
||||||
|
import traceback
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from lerobot import available_motors
|
from lerobot import available_motors
|
||||||
from lerobot.common.robot_devices.utils import RobotDeviceAlreadyConnectedError, RobotDeviceNotConnectedError
|
from lerobot.common.robot_devices.utils import RobotDeviceAlreadyConnectedError, RobotDeviceNotConnectedError
|
||||||
from tests.utils import make_motors_bus, mock_builtins_input, mock_motor, require_motor
|
from tests.utils import make_motors_bus, mock_builtins_input, require_motor
|
||||||
|
|
||||||
|
|
||||||
def _test_configure_motors_all_ids_1(motor_type):
|
def _test_configure_motors_all_ids_1(motor_type):
|
||||||
@@ -152,3 +153,39 @@ def test_motors_bus_mock(monkeypatch, motor_type):
|
|||||||
@require_motor
|
@require_motor
|
||||||
def test_motors_bus(request, motor_type):
|
def test_motors_bus(request, motor_type):
|
||||||
_test_motors_bus(motor_type)
|
_test_motors_bus(motor_type)
|
||||||
|
|
||||||
|
|
||||||
|
def mock_motor(monkeypatch, motor_type):
|
||||||
|
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 motor_type == "dynamixel":
|
||||||
|
try:
|
||||||
|
import dynamixel_sdk
|
||||||
|
|
||||||
|
from tests.mock_dynamixel import (
|
||||||
|
MockGroupSyncRead,
|
||||||
|
MockGroupSyncWrite,
|
||||||
|
MockPacketHandler,
|
||||||
|
MockPortHandler,
|
||||||
|
mock_convert_to_bytes,
|
||||||
|
)
|
||||||
|
|
||||||
|
monkeypatch.setattr(dynamixel_sdk, "GroupSyncRead", MockGroupSyncRead)
|
||||||
|
monkeypatch.setattr(dynamixel_sdk, "GroupSyncWrite", MockGroupSyncWrite)
|
||||||
|
monkeypatch.setattr(dynamixel_sdk, "PacketHandler", MockPacketHandler)
|
||||||
|
monkeypatch.setattr(dynamixel_sdk, "PortHandler", MockPortHandler)
|
||||||
|
|
||||||
|
# Import dynamixel AFTER mocking dynamixel_sdk to use mocked classes
|
||||||
|
from lerobot.common.robot_devices.motors import dynamixel
|
||||||
|
|
||||||
|
# TODO(rcadene): remove need to mock `convert_to_bytes` by implemented the inverse transform
|
||||||
|
# `convert_bytes_to_value`
|
||||||
|
monkeypatch.setattr(dynamixel, "convert_to_bytes", mock_convert_to_bytes)
|
||||||
|
except ImportError:
|
||||||
|
traceback.print_exc()
|
||||||
|
pytest.skip("To avoid skipping tests mocking dynamixel motors, run `pip install dynamixel-sdk`.")
|
||||||
|
else:
|
||||||
|
raise NotImplementedError("Implement mocking logic for new motor.")
|
||||||
|
|||||||
@@ -314,22 +314,8 @@ def require_mock_camera(func):
|
|||||||
return wrapper
|
return wrapper
|
||||||
|
|
||||||
|
|
||||||
def require_mock_motor(func):
|
def mock_motor(**kwargs):
|
||||||
@wraps(func)
|
pass
|
||||||
def wrapper(*args, **kwargs):
|
|
||||||
# Access the pytest request context to get the mockeypatch fixture
|
|
||||||
monkeypatch = kwargs.get("monkeypatch")
|
|
||||||
motor_type = kwargs.get("motor_type")
|
|
||||||
|
|
||||||
if monkeypatch is None:
|
|
||||||
raise ValueError("The 'monkeypatch' 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.")
|
|
||||||
|
|
||||||
mock_motor(monkeypatch, motor_type)
|
|
||||||
return func(*args, **kwargs)
|
|
||||||
|
|
||||||
return wrapper
|
|
||||||
|
|
||||||
|
|
||||||
def mock_robot(monkeypatch, robot_type):
|
def mock_robot(monkeypatch, robot_type):
|
||||||
@@ -394,42 +380,6 @@ def mock_camera(monkeypatch, camera_type):
|
|||||||
raise NotImplementedError("Implement mocking logic for new camera.")
|
raise NotImplementedError("Implement mocking logic for new camera.")
|
||||||
|
|
||||||
|
|
||||||
def mock_motor(monkeypatch, motor_type):
|
|
||||||
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 motor_type == "dynamixel":
|
|
||||||
try:
|
|
||||||
import dynamixel_sdk
|
|
||||||
|
|
||||||
from tests.mock_dynamixel import (
|
|
||||||
MockGroupSyncRead,
|
|
||||||
MockGroupSyncWrite,
|
|
||||||
MockPacketHandler,
|
|
||||||
MockPortHandler,
|
|
||||||
mock_convert_to_bytes,
|
|
||||||
)
|
|
||||||
|
|
||||||
monkeypatch.setattr(dynamixel_sdk, "GroupSyncRead", MockGroupSyncRead)
|
|
||||||
monkeypatch.setattr(dynamixel_sdk, "GroupSyncWrite", MockGroupSyncWrite)
|
|
||||||
monkeypatch.setattr(dynamixel_sdk, "PacketHandler", MockPacketHandler)
|
|
||||||
monkeypatch.setattr(dynamixel_sdk, "PortHandler", MockPortHandler)
|
|
||||||
|
|
||||||
# Import dynamixel AFTER mocking dynamixel_sdk to use mocked classes
|
|
||||||
from lerobot.common.robot_devices.motors import dynamixel
|
|
||||||
|
|
||||||
# TODO(rcadene): remove need to mock `convert_to_bytes` by implemented the inverse transform
|
|
||||||
# `convert_bytes_to_value`
|
|
||||||
monkeypatch.setattr(dynamixel, "convert_to_bytes", mock_convert_to_bytes)
|
|
||||||
except ImportError:
|
|
||||||
traceback.print_exc()
|
|
||||||
pytest.skip("To avoid skipping tests mocking dynamixel motors, run `pip install dynamixel-sdk`.")
|
|
||||||
else:
|
|
||||||
raise NotImplementedError("Implement mocking logic for new motor.")
|
|
||||||
|
|
||||||
|
|
||||||
def mock_builtins_input(monkeypatch):
|
def mock_builtins_input(monkeypatch):
|
||||||
def print_text(text=None):
|
def print_text(text=None):
|
||||||
if text is not None:
|
if text is not None:
|
||||||
|
|||||||
Reference in New Issue
Block a user