From ad4fcc3749a6dc2304f4066192423135d9e40725 Mon Sep 17 00:00:00 2001 From: MGross21 Date: Thu, 15 Oct 2026 16:16:35 -0700 Subject: [PATCH 1/6] test(cli): share fixtures via conftest and assert exact cli behavior Resets the module-global robot between tests, which previously made results order-dependent. --- pyproject.toml | 5 + tests/conftest.py | 49 ++++ tests/test_cli.py | 649 +++++++++++++--------------------------------- 3 files changed, 230 insertions(+), 473 deletions(-) create mode 100644 tests/conftest.py diff --git a/pyproject.toml b/pyproject.toml index 09b8ea5..6f5b637 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -75,3 +75,8 @@ lint.ignore = [ lint.per-file-ignores = { "__init__.py" = ["F401", "F403", "E402"], "armctl/__main__.py" = ["UP006", "UP035", "UP045"] } format.quote-style = "double" format.indent-style = "space" + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = ["-ra", "--strict-markers"] +markers = ["integration: requires Docker and the ur extra (URSim)"] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..1a2f6d3 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,49 @@ +import logging +from unittest.mock import MagicMock + +import pytest +from typer.testing import CliRunner + +import armctl +import armctl.__main__ as cli + +JOINTS = [0.1, 0.2, 0.3, 0.4, 0.5, 0.6] +POSE = [0.5, 0.0, 0.3, 0.0, 1.57, 0.0] + + +@pytest.fixture(autouse=True) +def _reset_global_state(): + yield + cli._robot = None + logging.disable(logging.NOTSET) + + +@pytest.fixture(scope="session") +def runner(): + return CliRunner(env={"NO_COLOR": "1"}) + + +@pytest.fixture +def cli_robot(monkeypatch): + """Replace every exported robot class with a factory returning one shared MagicMock.""" + robot = MagicMock() + robot.get_joint_positions.return_value = JOINTS + robot.get_cartesian_position.return_value = POSE + robot.get_robot_state.return_value = "RUNNING" + robot.init_args = None + del robot.home + + def factory(*args): + robot.init_args = args + return robot + + for name in armctl.__all__: + monkeypatch.setattr(armctl, name, factory) + return robot + + +@pytest.fixture +def connected(runner, cli_robot): + result = runner.invoke(cli.app, ["connect", "--ip", "10.0.0.1", "--robot-type", "ur"]) + assert result.exit_code == 0, result.output + return cli_robot diff --git a/tests/test_cli.py b/tests/test_cli.py index af5dd72..74cf201 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,478 +1,181 @@ -"""Comprehensive CLI tests for armctl.""" - -from __future__ import annotations - -import re -import sys -from unittest.mock import MagicMock, patch - +import click import pytest -from typer.testing import CliRunner -import armctl from armctl.__main__ import app, get_robot_types from armctl.utils import NetworkScanner -# ANSI escape sequence regex for stripping color codes -ANSI_ESC_RE = re.compile(r"\x1B[@-_][0-?]*[ -/]*[@-~]") - - -def strip_ansi(s: str) -> str: - """Strip ANSI escape sequences from string.""" - if s is None: - return s - return ANSI_ESC_RE.sub("", s) - - -@pytest.fixture(scope="session") -def runner(): - """Session-scoped CLI runner fixture.""" - return CliRunner(env={"NO_COLOR": "1"}) - - -@pytest.fixture(scope="function") -def mock_network_scanner(monkeypatch): - """Mock NetworkScanner for network scan tests.""" - - def mock_scan_network(): - return ["192.168.1.10", "192.168.1.20"] - - monkeypatch.setattr(NetworkScanner, "scan_network", mock_scan_network) - - -@pytest.fixture(scope="function") -def mock_robot(monkeypatch): - """Mock robot classes dynamically from armctl.__all__.""" - - class MockRobot: - def __init__(self, ip=None, port=None): - self.ip = ip - self.port = port - self.connect = MagicMock() - self.disconnect = MagicMock() - self.move_joints = MagicMock() - self.move_cartesian = MagicMock() - self.get_joint_positions = MagicMock(return_value=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6]) - self.get_cartesian_position = MagicMock(return_value=[0.5, 0.0, 0.3, 0.0, 1.57, 0.0]) - self.get_robot_state = MagicMock(return_value="RUNNING") - - mock_instance = None - - def mock_robot_factory(ip=None, port=None): - nonlocal mock_instance - mock_instance = MockRobot(ip, port) - return mock_instance - - for robot_class_name in armctl.__all__: - try: - monkeypatch.setattr(f"armctl.{robot_class_name}", mock_robot_factory) - except AttributeError: - pass - - class MockRobotAccessor: - def get_instance(self): - return mock_instance - - return MockRobotAccessor() - - -class TestCLIHelp: - """Test help functionality for all commands.""" - - def setup_method(self): - self.runner = CliRunner(env={"NO_COLOR": "1"}) - - def test_main_help(self): - """Test main command help.""" - result = self.runner.invoke(app, ["--help"], color=False) - assert result.exit_code == 0 - out = strip_ansi(result.stdout) - assert "Agnostic Robotic Manipulation Controller" in out - assert "connect" in out - assert "disconnect" in out - assert "move" in out - assert "get" in out - assert "control" in out - assert "utils" in out - - def test_connect_help(self): - """Test connect command help.""" - result = self.runner.invoke(app, ["connect", "--help"], color=False) - assert result.exit_code == 0 - out = strip_ansi(result.stdout) - assert "Connect to robot" in out - assert "--ip" in out - assert "--robot-type" in out - assert "--port" in out - - def test_move_help(self): - """Test move command help.""" - result = self.runner.invoke(app, ["move", "--help"], color=False) - assert result.exit_code == 0 - out = strip_ansi(result.stdout) - assert "Movement commands" in out - assert "joints" in out - assert "cartesian" in out - assert "home" in out - - def test_get_help(self): - """Test get command help.""" - result = self.runner.invoke(app, ["get", "--help"], color=False) - assert result.exit_code == 0 - out = strip_ansi(result.stdout) - assert "Get robot information" in out - assert "joints" in out - assert "cartesian" in out - assert "state" in out - - def test_control_help(self): - """Test control command help.""" - result = self.runner.invoke(app, ["control", "--help"], color=False) - assert result.exit_code == 0 - out = strip_ansi(result.stdout) - assert "Robot control" in out - assert "stop" in out - assert "sleep" in out - - def test_utils_help(self): - """Test utils command help.""" - result = self.runner.invoke(app, ["utils", "--help"], color=False) - assert result.exit_code == 0 - out = strip_ansi(result.stdout) - assert "Utility commands" in out - assert "scan" in out - assert "list" in result.stdout - - -class TestUtilsCommands: - """Test utility commands.""" - - def setup_method(self): - self.runner = CliRunner(env={"NO_COLOR": "1"}) - - def test_utils_list(self): - result = self.runner.invoke(app, ["utils", "list"]) - assert result.exit_code == 0 - robot_types = get_robot_types() - for robot_type in robot_types: - assert robot_type in result.stdout - - def test_utils_scan_basic(self, runner, mock_network_scanner): - result = runner.invoke(app, ["utils", "scan"]) - assert result.exit_code == 0 - assert "192.168.1.10" in result.stdout - assert "192.168.1.20" in result.stdout - - -class TestConnectionCommands: - """Test connection commands.""" - - def setup_method(self): - self.runner = CliRunner(env={"NO_COLOR": "1"}) - - def test_connect_missing_ip(self): - result = self.runner.invoke(app, ["connect", "--robot-type", "ur5"]) - assert result.exit_code == 2 - - def test_connect_missing_robot_type(self): - result = self.runner.invoke(app, ["connect", "--ip", "192.168.1.10"]) - assert result.exit_code == 2 - - def test_connect_invalid_robot_type(self): - result = self.runner.invoke( - app, - [ - "connect", - "--ip", - "192.168.1.10", - "--robot-type", - "invalid_robot", - ], - ) - assert result.exit_code == 2 - - def test_connect_success(self, runner, mock_robot): - result = runner.invoke( - app, - [ - "connect", - "--ip", - "192.168.1.10", - "--robot-type", - "universalrobots", - ], - ) - assert result.exit_code == 0 - instance = mock_robot.get_instance() - assert instance is not None - instance.connect.assert_called_once() - - def test_connect_with_port(self, runner, mock_robot): - result = runner.invoke( - app, - [ - "connect", - "--ip", - "192.168.1.10", - "--robot-type", - "universalrobots", - "--port", - "30001", - ], - ) - assert result.exit_code == 0 - instance = mock_robot.get_instance() - assert instance is not None - instance.connect.assert_called_once() - - def test_connect_failure(self, runner, mock_robot): - class FailingRobot: - def __init__(self, ip=None, port=None): - self.ip = ip - self.port = port - - def connect(self): - raise Exception("Connection failed") - - def mock_get_robot_types_failing(): - types = get_robot_types() - types["universalrobots"] = FailingRobot - return types - - with patch("armctl.__main__.get_robot_types", mock_get_robot_types_failing): - result = runner.invoke( - app, - [ - "connect", - "--ip", - "192.168.1.10", - "--robot-type", - "universalrobots", - ], - ) - assert result.exit_code == 1 - try: - output = result.stderr - except (ValueError, AttributeError): - output = result.stdout - assert "Connection failed" in output - - def test_disconnect_no_connection(self, runner): - result = runner.invoke(app, ["disconnect"]) - assert result.exit_code in [0, 1] - - -class TestMovementCommands: - """Test movement commands.""" - - def setup_method(self): - self.runner = CliRunner(env={"NO_COLOR": "1"}) - - def test_move_joints_no_connection(self): - result = self.runner.invoke(app, ["move", "joints", "0", "0", "0", "0", "0", "0"]) - assert result.exit_code == 1 - - def test_move_joints_wrong_count(self): - result = self.runner.invoke(app, ["move", "joints", "0", "0", "0"]) - assert result.exit_code == 1 - - def test_move_cartesian_no_connection(self): - result = self.runner.invoke(app, ["move", "cartesian", "0", "0", "0", "0", "0", "0"]) - assert result.exit_code == 1 - - def test_move_cartesian_wrong_count(self): - result = self.runner.invoke(app, ["move", "cartesian", "0", "0", "0", "0", "0"]) - assert result.exit_code == 2 - - def test_move_home_no_connection(self): - result = self.runner.invoke(app, ["move", "home"]) - assert result.exit_code == 1 - - -class TestGetCommands: - """Test get commands.""" - - def setup_method(self): - self.runner = CliRunner(env={"NO_COLOR": "1"}) - - def test_get_joints_no_connection(self): - result = self.runner.invoke(app, ["get", "joints"]) - assert result.exit_code == 1 - - def test_get_cartesian_no_connection(self): - result = self.runner.invoke(app, ["get", "cartesian"]) - assert result.exit_code == 1 - - def test_get_state_no_connection(self): - result = self.runner.invoke(app, ["get", "state"]) - assert result.exit_code == 1 - - -class TestControlCommands: - """Test control commands.""" - - def setup_method(self): - self.runner = CliRunner(env={"NO_COLOR": "1"}) - - def test_control_stop_no_connection(self): - result = self.runner.invoke(app, ["control", "stop"]) - assert result.exit_code == 1 - - def test_control_sleep_no_connection(self): - result = self.runner.invoke(app, ["control", "sleep", "1.0"]) - assert result.exit_code == 1 - - def test_control_sleep_invalid_duration(self): - result = self.runner.invoke(app, ["control", "sleep", "invalid"]) - assert result.exit_code != 0 - - -class TestIntegratedWorkflow: - """Test integrated workflows.""" - - def setup_method(self): - self.runner = CliRunner(env={"NO_COLOR": "1"}) - - def test_full_workflow(self, runner, mock_robot, monkeypatch): - result = runner.invoke( - app, - [ - "connect", - "--ip", - "192.168.1.10", - "--robot-type", - "universalrobots", - ], - ) - assert result.exit_code == 0 - - instance = mock_robot.get_instance() - assert instance is not None - instance.connect.assert_called_once() - - result = runner.invoke(app, ["disconnect"]) - assert result.exit_code == 0 - - -class TestRobotTypeHandling: - """Test robot type handling.""" - - def setup_method(self): - self.runner = CliRunner(env={"NO_COLOR": "1"}) - - def test_get_robot_types(self): - types = get_robot_types() - assert isinstance(types, dict) - assert len(types) > 0 - expected_types = [ - "universalrobots", - "ur", - "jaka", - "vention", - "elephant", - ] - for expected in expected_types: - assert expected in types - - @pytest.mark.parametrize( - "robot_type", - [ - "universalrobots", - "ur", - "ur5", - "ur10", - "ur3", - "ur5e", - "ur16", - "jaka", - "vention", - "elephant", - "elephantrobotics", - "pro600", - ], - ) - def test_all_robot_types_connect(self, robot_type, runner, mock_robot): - result = runner.invoke(app, ["connect", "--ip", "192.168.1.10", "--robot-type", robot_type]) - assert result.exit_code in [0, 2] - if result.exit_code == 2: - pytest.skip(f"Robot type {robot_type} not in current enum") - - -class TestErrorHandling: - """Test error handling.""" - - def _version_compatible_exit_code(self, result): - if sys.version_info >= (3, 10): - return result.exit_code != 0 - else: - return result.exit_code == 0 - - def setup_method(self): - self.runner = CliRunner(env={"NO_COLOR": "1"}) - - def test_invalid_command(self): - result = self.runner.invoke(app, ["invalid_command"]) - assert result.exit_code != 0 - - def test_missing_subcommand(self): - result = self.runner.invoke(app, ["move"]) - assert self._version_compatible_exit_code(result) - - result = self.runner.invoke(app, ["get"]) - assert self._version_compatible_exit_code(result) - - result = self.runner.invoke(app, ["control"]) - assert self._version_compatible_exit_code(result) - - result = self.runner.invoke(app, ["utils"]) - assert self._version_compatible_exit_code(result) - - def test_robot_method_with_mock(self, runner, mock_robot): - result = runner.invoke( - app, - [ - "connect", - "--ip", - "192.168.1.10", - "--robot-type", - "universalrobots", - ], - ) - assert result.exit_code == 0 - - result = runner.invoke(app, ["move", "joints", "0", "0", "0", "0", "0", "0"]) - assert result.exit_code == 0 - - def test_home_not_supported(self, runner, mock_robot): - result = runner.invoke( - app, - [ - "connect", - "--ip", - "192.168.1.10", - "--robot-type", - "universalrobots", - ], - ) - assert result.exit_code == 0 - - result = runner.invoke(app, ["move", "home"]) - assert result.exit_code == 1 - - -class TestCLISubprocess: - """Test CLI integration.""" - - def test_cli_help(self, runner): - result = runner.invoke(app, ["--help"]) - assert result.exit_code == 0 - assert "Agnostic Robotic Manipulation Controller" in result.stdout - - def test_cli_utils_list(self, runner): - result = runner.invoke(app, ["utils", "list"]) - assert result.exit_code == 0 - assert "universalrobots" in result.stdout or "UniversalRobots" in result.stdout - - -if __name__ == "__main__": - pytest.main([__file__, "-v"]) +EXPORTED_TYPES = sorted(get_robot_types()) + + +@pytest.mark.parametrize( + "args, expected", + [ + ([], ["connect", "disconnect", "move", "get", "control", "utils"]), + (["connect"], ["--ip", "--robot-type", "--port"]), + (["move"], ["joints", "cartesian", "home"]), + (["get"], ["joints", "cartesian", "state"]), + (["control"], ["stop", "sleep"]), + (["utils"], ["scan", "list"]), + ], +) +def test_help_lists_subcommands(runner, args, expected): + result = runner.invoke(app, [*args, "--help"]) + assert result.exit_code == 0 + output = click.unstyle(result.output) + for word in expected: + assert word in output + + +@pytest.mark.parametrize("group", ["move", "get", "control", "utils"]) +def test_group_without_subcommand_shows_usage(runner, group): + result = runner.invoke(app, [group]) + assert result.exit_code in (0, 2) # click < 8.2 exits 0 + assert "Usage" in click.unstyle(result.output) + + +def test_unknown_command(runner): + assert runner.invoke(app, ["invalid_command"]).exit_code == 2 + + +def test_utils_list(runner): + result = runner.invoke(app, ["utils", "list"]) + assert result.exit_code == 0 + for robot_type in EXPORTED_TYPES: + assert robot_type in result.output + + +def test_utils_scan(runner, monkeypatch): + monkeypatch.setattr(NetworkScanner, "scan_network", lambda: ["192.168.1.10", "192.168.1.20"]) + result = runner.invoke(app, ["utils", "scan"]) + assert result.exit_code == 0 + assert result.output.split() == ["192.168.1.10", "192.168.1.20"] + + +@pytest.mark.parametrize( + "args", + [ + ["connect", "--robot-type", "ur"], + ["connect", "--ip", "10.0.0.1"], + ["connect", "--ip", "10.0.0.1", "--robot-type", "invalid_robot"], + ], +) +def test_connect_rejects_bad_options(runner, args): + assert runner.invoke(app, args).exit_code == 2 + + +@pytest.mark.parametrize("robot_type", EXPORTED_TYPES) +def test_connect_every_robot_type(runner, cli_robot, robot_type): + result = runner.invoke(app, ["connect", "--ip", "10.0.0.1", "--robot-type", robot_type]) + assert result.exit_code == 0, result.output + assert cli_robot.init_args == ("10.0.0.1",) + cli_robot.connect.assert_called_once() + + +def test_connect_passes_port(runner, cli_robot): + args = ["connect", "--ip", "10.0.0.1", "--robot-type", "ur", "--port", "30001"] + assert runner.invoke(app, args).exit_code == 0 + assert cli_robot.init_args == ("10.0.0.1", 30001) + + +def test_connect_failure_reports_error(runner, cli_robot): + cli_robot.connect.side_effect = ConnectionError("Connection failed") + result = runner.invoke(app, ["connect", "--ip", "10.0.0.1", "--robot-type", "ur"]) + assert result.exit_code == 1 + assert "Connection failed" in result.output + + +def test_disconnect(runner, connected): + assert runner.invoke(app, ["disconnect"]).exit_code == 0 + connected.disconnect.assert_called_once() + assert runner.invoke(app, ["disconnect"]).exit_code == 1 + + +@pytest.mark.parametrize( + "args", + [ + ["disconnect"], + ["move", "joints", "0", "0", "0", "0", "0", "0"], + ["move", "cartesian", "0", "0", "0", "0", "0", "0"], + ["move", "home"], + ["get", "joints"], + ["get", "cartesian"], + ["get", "state"], + ["control", "stop"], + ["control", "sleep", "1.0"], + ], +) +def test_requires_connection(runner, args): + result = runner.invoke(app, args) + assert result.exit_code == 1 + assert "connect" in result.output.lower() + + +def test_move_joints_forwards_positions(runner, connected): + result = runner.invoke(app, ["move", "joints", "1", "2", "3"]) + assert result.exit_code == 0 + connected.move_joints.assert_called_once_with([1.0, 2.0, 3.0]) + + +def test_move_cartesian_forwards_pose(runner, connected): + result = runner.invoke(app, ["move", "cartesian", "1", "2", "3", "4", "5", "6"]) + assert result.exit_code == 0 + connected.move_cartesian.assert_called_once_with([1.0, 2.0, 3.0, 4.0, 5.0, 6.0]) + + +def test_move_cartesian_requires_six_values(runner, connected): + assert runner.invoke(app, ["move", "cartesian", "0", "0", "0", "0", "0"]).exit_code == 2 + connected.move_cartesian.assert_not_called() + + +def test_move_failure_reports_error(runner, connected): + connected.move_joints.side_effect = ValueError("out of range") + result = runner.invoke(app, ["move", "joints", "9"]) + assert result.exit_code == 1 + assert "out of range" in result.output + + +def test_move_home_unsupported(runner, connected): + result = runner.invoke(app, ["move", "home"]) + assert result.exit_code == 1 + assert "not supported" in result.output + + +@pytest.mark.parametrize( + "args, method", + [ + (["get", "joints"], "get_joint_positions"), + (["get", "cartesian"], "get_cartesian_position"), + ], +) +def test_get_prints_positions(runner, connected, args, method): + result = runner.invoke(app, args) + assert result.exit_code == 0 + values = getattr(connected, method).return_value + assert result.output.split() == [str(v) for v in values] + + +def test_get_state(runner, connected): + result = runner.invoke(app, ["get", "state"]) + assert result.exit_code == 0 + assert result.output.strip() == "RUNNING" + + +def test_control_stop(runner, connected): + assert runner.invoke(app, ["control", "stop"]).exit_code == 0 + connected.stop_motion.assert_called_once_with() + + +def test_control_sleep(runner, connected): + assert runner.invoke(app, ["control", "sleep", "0.5"]).exit_code == 0 + connected.sleep.assert_called_once_with(0.5) + + +def test_control_sleep_rejects_non_number(runner, connected): + assert runner.invoke(app, ["control", "sleep", "invalid"]).exit_code == 2 + connected.sleep.assert_not_called() + + +def test_get_robot_types_aliases(): + types = get_robot_types() + assert types["ur"] is types["universalrobots"] + assert types["elephant"] is types["elephantrobotics"] From e9d6cd1216ba97d9a60da63b3b5985542f270c41 Mon Sep 17 00:00:00 2001 From: MGross21 Date: Thu, 15 Oct 2026 16:16:35 -0700 Subject: [PATCH 2/6] test: parametrize inheritance and name tests and stop swallowing failures --- tests/test_inheritance.py | 35 ++++++++++-------- tests/test_logger.py | 3 +- tests/test_name_property.py | 74 ++++++++----------------------------- 3 files changed, 35 insertions(+), 77 deletions(-) diff --git a/tests/test_inheritance.py b/tests/test_inheritance.py index 3a95d66..12d4d11 100644 --- a/tests/test_inheritance.py +++ b/tests/test_inheritance.py @@ -1,3 +1,5 @@ +import pytest + from armctl.dobot import Dobot from armctl.elephant_robotics import ElephantRobotics, Pro600 from armctl.fanuc import Fanuc @@ -11,8 +13,8 @@ ) from armctl.universal_robots import UR3, UR5, UR10, UR16, UniversalRobots, UR5e from armctl.vention import Vention +from armctl.zaber import Zaber -# All robot classes that should inherit from Commands and a controller ROBOTS = [ Dobot, ElephantRobotics, @@ -27,22 +29,23 @@ Vention, Fanuc, ] +CONTROLLERS = (SerialController, SocketController, PLCController) + -# Required base classes -REQUIRED_BASES = (Commands, Properties) -CONTROLLER_BASES = (SerialController, SocketController, PLCController) +INCOMPLETE = pytest.mark.xfail(reason="Fanuc driver does not implement sleep", strict=True) -def test_robot_subclass_inheritance(): - """Test that all robot classes inherit from Commands and at least one controller.""" - for robot in ROBOTS: - # Check required base class inheritance - for base in REQUIRED_BASES: - assert issubclass(robot, base), f"{robot.__name__} must inherit from {base.__name__}" +@pytest.mark.parametrize( + "robot", + [pytest.param(r, marks=INCOMPLETE) if r is Fanuc else r for r in [*ROBOTS, Zaber]], + ids=lambda r: r.__name__, +) +def test_implements_commands_and_properties(robot): + assert issubclass(robot, Commands) + assert issubclass(robot, Properties) + assert not getattr(robot, "__abstractmethods__", None) + - # Check controller inheritance - has_controller = any(issubclass(robot, c) for c in CONTROLLER_BASES) - assert has_controller, ( - f"{robot.__name__} must inherit from at least one controller: " - f"{', '.join(c.__name__ for c in CONTROLLER_BASES)}" - ) +@pytest.mark.parametrize("robot", ROBOTS, ids=lambda r: r.__name__) +def test_uses_a_controller(robot): + assert issubclass(robot, CONTROLLERS) diff --git a/tests/test_logger.py b/tests/test_logger.py index a1254cb..8133118 100644 --- a/tests/test_logger.py +++ b/tests/test_logger.py @@ -36,10 +36,9 @@ def test_logger_receive_logs_message(caplog): def test_logger_verbosity_and_enable_disable(caplog): with caplog.at_level(logging.INFO): - Logger.enable() # Make sure logging is enabled + Logger.enable() logging.info("This should appear") assert "This should appear" in caplog.text Logger.disable() logging.info("This should NOT appear") assert "This should NOT appear" not in caplog.text - Logger.enable() # Re-enable for other tests diff --git a/tests/test_name_property.py b/tests/test_name_property.py index 9bb56a2..362835d 100644 --- a/tests/test_name_property.py +++ b/tests/test_name_property.py @@ -1,7 +1,3 @@ -""" -Test suite for the __name__ property behavior across robot class hierarchy. -""" - import pytest from armctl.dobot import Dobot @@ -9,60 +5,20 @@ from armctl.universal_robots import UR3, UR5, UniversalRobots -class TestRobotNameProperty: - """Test the __name__ property for different inheritance levels.""" - - def test_manufacturer_classes_return_class_name(self): - """Test that manufacturer classes return just their class name.""" - # These require instantiation since __name__ is now a property - try: - assert ElephantRobotics("192.168.1.1", 5001).__name__ == "ElephantRobotics" - except (NotImplementedError, ImportError): - pass # ElephantRobotics might not be fully implemented - - try: - assert UniversalRobots("192.168.1.1").__name__ == "UniversalRobots" - except (NotImplementedError, ImportError): - pass # UniversalRobots might not be fully implemented - - try: - assert Dobot("192.168.1.1", 5001).__name__ == "Dobot" - except (NotImplementedError, ImportError): - pass # Dobot raises NotImplementedError - - def test_robot_series_classes_return_manufacturer_and_model(self): - """Test that robot series classes return 'Manufacturer Model' format.""" - try: - assert Pro600().__name__ == "ElephantRobotics Pro600" - except (NotImplementedError, ImportError): - pass - - try: - assert UR5("192.168.1.1").__name__ == "UniversalRobots UR5" - except (NotImplementedError, ImportError): - pass +@pytest.mark.parametrize( + "robot_class, init_args, expected", + [ + (ElephantRobotics, ("192.168.1.1", 5001), "ElephantRobotics"), + (UniversalRobots, ("192.168.1.1",), "UniversalRobots"), + (Pro600, (), "ElephantRobotics Pro600"), + (UR5, ("192.168.1.1",), "UniversalRobots UR5"), + (UR3, ("192.168.1.1", 30002), "UniversalRobots UR3"), + ], +) +def test_name_property(robot_class, init_args, expected): + assert robot_class(*init_args).__name__ == expected - try: - assert UR3("192.168.1.1").__name__ == "UniversalRobots UR3" - except (NotImplementedError, ImportError): - pass - @pytest.mark.parametrize( - "robot_class,expected_name,init_args", - [ - (ElephantRobotics, "ElephantRobotics", ("192.168.1.1", 5001)), - (UniversalRobots, "UniversalRobots", ("192.168.1.1", 30002)), - (Dobot, "Dobot", ("192.168.1.1", 5001)), - (Pro600, "ElephantRobotics Pro600", ()), - (UR5, "UniversalRobots UR5", ("192.168.1.1",)), - (UR3, "UniversalRobots UR3", ("192.168.1.1", 30002)), - ], - ) - def test_name_property_parametrized(self, robot_class, expected_name, init_args): - """Parametrized test for all robot classes.""" - try: - instance = robot_class(*init_args) - assert instance.__name__ == expected_name - except (NotImplementedError, ImportError) as e: - # Some classes might not be fully implemented or missing dependencies, skip them - pytest.skip(f"Cannot test {robot_class.__name__} - {str(e)}") +def test_dobot_not_supported(): + with pytest.raises(NotImplementedError): + Dobot("192.168.1.1", 5001) From e80b0d2ff2ccde3551b0bbda09229c1ef9ab57e9 Mon Sep 17 00:00:00 2001 From: MGross21 Date: Thu, 15 Oct 2026 16:16:35 -0700 Subject: [PATCH 3/6] test: send connect banner from echo server to cut socket tests from 16s to 0.02s --- tests/_mock_robot.py | 76 +++++++++++--------------------------- tests/test_socket_robot.py | 61 ++++++++++-------------------- 2 files changed, 41 insertions(+), 96 deletions(-) diff --git a/tests/_mock_robot.py b/tests/_mock_robot.py index 90e51c1..97ffbbf 100644 --- a/tests/_mock_robot.py +++ b/tests/_mock_robot.py @@ -1,12 +1,4 @@ -""" -This module provides a `MockSocketRobot` class that inherits from the real -SocketController and Commands classes, enabling realistic socket-based communication -during tests. To avoid the need for an external server, a lightweight TCP echo server -is started in a background thread on initialization and stopped on exit. This allows -all socket commands to be sent and received as if communicating with a real robot, -with the server simply echoing back any data received. The echo server is started -and stopped automatically for each test instance, ensuring isolation and reliability. -""" +"""SocketController-backed robot whose commands are echoed back by an in-process TCP server.""" import socket import threading @@ -18,67 +10,41 @@ TEST_STRING_PREFIX = "MOCK!!" -def _start_echo_server(host, port): - server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - server.bind((host, port)) - server.listen() +def _start_echo_server(host): + server = socket.create_server((host, 0)) + server.settimeout(0.5) stop_event = threading.Event() - def handle_client(conn, addr): + def handle_client(conn): with conn: + conn.sendall(b"READY") while not stop_event.is_set(): - try: - data = conn.recv(4096) - if not data: - break - conn.sendall(data) - except Exception: + data = conn.recv(4096) + if not data: break + conn.sendall(data) - def server_thread(): - while not stop_event.is_set(): - try: - server.settimeout(0.5) - conn, addr = server.accept() - threading.Thread(target=handle_client, args=(conn, addr), daemon=True).start() - except socket.timeout: - continue - except Exception: - break - server.close() - - thread = threading.Thread(target=server_thread, daemon=True) - thread.start() - return stop_event, server - + def serve(): + with server: + while not stop_event.is_set(): + try: + conn, _ = server.accept() + except socket.timeout: + continue + threading.Thread(target=handle_client, args=(conn,), daemon=True).start() -def _get_free_port(): - """Find a free port on localhost.""" - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.bind(("127.0.0.1", 0)) - return s.getsockname()[1] + threading.Thread(target=serve, daemon=True).start() + return stop_event, server.getsockname()[1] class MockSocketRobot(Socket, Commands): - """ - Mock robot using real SocketController logic with an integrated echo server. - Starts a background echo server on a random free port for the lifetime of the object. - All commands are sent over a real socket and echoed back for test verification. - """ - - def __init__(self, ip: str = "127.0.0.1", port: int = None): - if port is None: - port = _get_free_port() - self._echo_port = port - self._echo_stop_event, self._echo_server = _start_echo_server(ip, port) - time.sleep(0.1) # Give server time to start + def __init__(self, ip: str = "127.0.0.1"): + self._echo_stop_event, port = _start_echo_server(ip) super().__init__(ip, port) def __exit__(self, exc_type, exc_val, exc_tb): super().__exit__(exc_type, exc_val, exc_tb) self._echo_stop_event.set() - time.sleep(0.1) # Allow server thread to exit def _mock_command(self, name: str, arg=None) -> str: cmd = f"{TEST_STRING_PREFIX} {name.upper()}" + (f": {arg}" if arg is not None else "") diff --git a/tests/test_socket_robot.py b/tests/test_socket_robot.py index 7eec2e4..8073da4 100644 --- a/tests/test_socket_robot.py +++ b/tests/test_socket_robot.py @@ -1,51 +1,30 @@ -import time - import pytest from tests._mock_robot import TEST_STRING_PREFIX, MockSocketRobot @pytest.fixture -def mock_robot(): +def echo_robot(): with MockSocketRobot() as robot: yield robot -def test_move_joints(mock_robot): - pos = [1, 2, 3] - expected = f"{TEST_STRING_PREFIX} MOVE_JOINTS: {pos}" - assert mock_robot.move_joints(pos) == expected - - -def test_move_cartesian(mock_robot): - pose = [0.1, 0.2, 0.3] - expected = f"{TEST_STRING_PREFIX} MOVE_CARTESIAN: {pose}" - assert mock_robot.move_cartesian(pose) == expected - - -def test_get_joint_positions(mock_robot): - expected = f"{TEST_STRING_PREFIX} GET_JOINT_POSITIONS" - assert mock_robot.get_joint_positions() == expected - - -def test_get_cartesian_position(mock_robot): - expected = f"{TEST_STRING_PREFIX} GET_CARTESIAN_POSITION" - assert mock_robot.get_cartesian_position() == expected - - -def test_stop_motion(mock_robot): - expected = f"{TEST_STRING_PREFIX} STOP_MOTION" - assert mock_robot.stop_motion() == expected - - -def test_get_robot_state(mock_robot): - expected = f"{TEST_STRING_PREFIX} GET_ROBOT_STATE" - assert mock_robot.get_robot_state() == expected - - -def test_sleep_duration(mock_robot): - sleep_seconds = 0.2 - start = time.time() - mock_robot.sleep(sleep_seconds) - elapsed = time.time() - start - assert elapsed >= sleep_seconds +@pytest.mark.parametrize( + "method, args, expected", + [ + ("move_joints", ([1, 2, 3],), "MOVE_JOINTS: [1, 2, 3]"), + ("move_cartesian", ([0.1, 0.2, 0.3],), "MOVE_CARTESIAN: [0.1, 0.2, 0.3]"), + ("get_joint_positions", (), "GET_JOINT_POSITIONS"), + ("get_cartesian_position", (), "GET_CARTESIAN_POSITION"), + ("stop_motion", (), "STOP_MOTION"), + ("get_robot_state", (), "GET_ROBOT_STATE"), + ], +) +def test_command_round_trips_over_socket(echo_robot, method, args, expected): + assert getattr(echo_robot, method)(*args) == f"{TEST_STRING_PREFIX} {expected}" + + +def test_reconnect_after_disconnect(echo_robot): + echo_robot.disconnect() + echo_robot.connect() + assert echo_robot.stop_motion() == f"{TEST_STRING_PREFIX} STOP_MOTION" From c66106d0dbe418ea8f25c31d01ba0ad2b402322b Mon Sep 17 00:00:00 2001 From: MGross21 Date: Thu, 15 Oct 2026 16:16:35 -0700 Subject: [PATCH 4/6] test: cover ur, jaka and vention command building and reply parsing --- tests/test_drivers.py | 153 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 153 insertions(+) create mode 100644 tests/test_drivers.py diff --git a/tests/test_drivers.py b/tests/test_drivers.py new file mode 100644 index 0000000..1f52333 --- /dev/null +++ b/tests/test_drivers.py @@ -0,0 +1,153 @@ +import ast +import math +from unittest.mock import MagicMock + +import pytest + +from armctl import Jaka, UniversalRobots, Vention + +HOME = [0.0, -1.57, 1.57, 0.0, 0.0, 0.0] + + +@pytest.fixture +def ur(): + robot = UniversalRobots("10.0.0.1") + robot.send_command = MagicMock(return_value="") + robot.rtde = MagicMock() + return robot + + +@pytest.fixture +def jaka(): + robot = Jaka("10.0.0.1") + robot.send_command = MagicMock() + return robot + + +def jaka_reply(robot, cmd_name, **fields): + robot.send_command.return_value = str({"errorCode": "0", "cmdName": cmd_name, **fields}) + + +@pytest.fixture +def vention(monkeypatch): + monkeypatch.setattr("armctl.vention.vention.time.sleep", lambda _: None) + robot = Vention() + robot.send_command = MagicMock() + return robot + + +def test_ur_move_joints_script(ur): + ur.move_joints(HOME, speed=0.5, acceleration=0.2) + ur.send_command.assert_called_once() + assert ur.send_command.call_args.args[0] == ( + "write_output_integer_register(0,0)\n" + "movej([0.0,-1.57,1.57,0.0,0.0,0.0], a=0.2, v=0.5, t=0.0, r=0.0)\n" + "write_output_integer_register(0,1)\n" + ) + ur.rtde.wait_until_stopped.assert_called_once_with(timeout=120.0) + + +def test_ur_blended_move_does_not_block(ur): + ur.move_joints(HOME, radius=0.05) + ur.rtde.wait_until_stopped.assert_not_called() + + +def test_ur_movep_omits_time(ur): + ur.move_cartesian([0.1, 0.2, 0.3, 0.0, 3.14, 0.0], move_type="movep") + script = ur.send_command.call_args.args[0] + assert "movep(p[0.1,0.2,0.3,0.0,3.14,0.0], a=0.1, v=0.1, r=0.0)" in script + assert "t=" not in script + + +def test_ur_rejects_unknown_move_type(ur): + with pytest.raises(ValueError, match="Unsupported move type"): + ur.move_cartesian([0.0] * 6, move_type="movec") + ur.send_command.assert_not_called() + + +@pytest.mark.parametrize( + "pos, kwargs, error", + [ + (HOME[:5], {}, ValueError), + ([7.0, 0, 0, 0, 0, 0], {}, ValueError), + (HOME, {"speed": 5.0}, ValueError), + (HOME, {"acceleration": 50.0}, ValueError), + (tuple(HOME), {}, TypeError), + (["0", 0, 0, 0, 0, 0], {}, TypeError), + ], +) +def test_ur_validates_before_sending(ur, pos, kwargs, error): + with pytest.raises(error): + ur.move_joints(pos, **kwargs) + ur.send_command.assert_not_called() + + +def test_jaka_move_joints_sends_degrees(jaka): + jaka_reply(jaka, "joint_move") + jaka.move_joints([math.pi / 2, 0, 0, 0, 0, 0], speed=math.pi, acceleration=math.pi / 2) + sent = ast.literal_eval(jaka.send_command.call_args.args[0]) + assert sent["jointPosition"] == pytest.approx([90, 0, 0, 0, 0, 0]) + assert sent["speed"] == pytest.approx(180) + assert sent["accel"] == pytest.approx(90) + + +def test_jaka_get_joint_positions_returns_radians(jaka): + jaka_reply(jaka, "get_joint_pos", joint_pos=[90, 0, -90, 0, 180, 0]) + assert jaka.get_joint_positions() == pytest.approx( + [math.pi / 2, 0, -math.pi / 2, 0, math.pi, 0] + ) + + +def test_jaka_get_cartesian_position_converts_orientation(jaka): + jaka_reply(jaka, "get_tcp_pos", tcp_pos=[100, 200, 300, 90, 0, 180]) + assert jaka.get_cartesian_position() == pytest.approx([100, 200, 300, math.pi / 2, 0, math.pi]) + + +@pytest.mark.parametrize( + "reply", + [ + str({"errorCode": "2", "cmdName": "get_joint_pos", "errorMsg": "not enabled"}), + str({"errorCode": "0", "cmdName": "other_cmd"}), + "not a dict literal", + ], +) +def test_jaka_rejects_bad_replies(jaka, reply): + jaka.send_command.return_value = reply + with pytest.raises(RuntimeError): + jaka.get_joint_positions() + + +def test_vention_move_converts_to_mm_and_waits(vention): + vention.send_command.side_effect = ["Ack", "Ack", "Ack", "Ack", "Ack", "false", "true"] + vention.move_joints([0.1, 0.25, 1.0], speed=1.0, acceleration=0.5) + sent = [c.args[0] for c in vention.send_command.call_args_list] + assert sent[:5] == [ + "SET speed/1.0/;", + "SET acceleration/0.5/;", + "SET im_move_abs_1/100.0/;", + "SET im_move_abs_2/250.0/;", + "SET im_move_abs_3/1000.0/;", + ] + assert sent[5:] == ["isMotionCompleted;", "isMotionCompleted;"] + + +def test_vention_move_fails_without_ack(vention): + vention.send_command.side_effect = ["Ack", "Ack", "Nack"] + with pytest.raises(RuntimeError, match="axis 1"): + vention.move_joints([0.1, 0.1, 0.1]) + + +@pytest.mark.parametrize( + "reply, expected", + [("(125.5)", 0.1255), ("(undefined)", 0.0)], +) +def test_vention_parses_axis_position(vention, reply, expected): + vention.send_command.return_value = reply + assert vention.get_joint_positions(axis=1) == pytest.approx(expected) + + +@pytest.mark.parametrize("reply", ["(-1)", "(garbage)"]) +def test_vention_rejects_bad_axis_position(vention, reply): + vention.send_command.return_value = reply + with pytest.raises(RuntimeError): + vention.get_joint_positions(axis=1) From c008dec0813ded8005b375b1b078861e435819d8 Mon Sep 17 00:00:00 2001 From: MGross21 Date: Thu, 15 Oct 2026 16:11:54 -0700 Subject: [PATCH 5/6] fix(jaka): convert cartesian position with pose2rad --- armctl/jaka/jaka.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/armctl/jaka/jaka.py b/armctl/jaka/jaka.py index 42a5409..0d73bea 100644 --- a/armctl/jaka/jaka.py +++ b/armctl/jaka/jaka.py @@ -146,7 +146,7 @@ def get_cartesian_position(self) -> list[float]: cc.get_cartesian_position() cmd = {"cmdName": "get_tcp_pos"} response = self._send_and_check(cmd) - return self.to_radians_cartesian(response["tcp_pos"]) + return uu.pose2rad(response["tcp_pos"]) def stop_motion(self) -> None: cc.stop_motion() From f19b46cff0311867d4d37260a23b3e478204fcb8 Mon Sep 17 00:00:00 2001 From: MGross21 Date: Thu, 15 Oct 2026 16:16:35 -0700 Subject: [PATCH 6/6] test(ursim): fail instead of skip when ursim connection breaks --- tests/test_ursim_docker.py | 29 +++++++---------------------- 1 file changed, 7 insertions(+), 22 deletions(-) diff --git a/tests/test_ursim_docker.py b/tests/test_ursim_docker.py index 77b2286..ab72903 100644 --- a/tests/test_ursim_docker.py +++ b/tests/test_ursim_docker.py @@ -29,6 +29,9 @@ import pytest +pytestmark = pytest.mark.integration +pytest.importorskip("rtde", reason="requires the ur extra") + if TYPE_CHECKING: from armctl import UR5 @@ -247,14 +250,7 @@ def get_ursim_port_mappings() -> dict[str, str]: @pytest.fixture(scope="session") def ursim_container(): - """Start URSim Docker container using the official image. - - Yields: - str: Container ID. - - Raises: - pytest.skip: If Docker is not available or container fails to start. - """ + """Yield the ID of a running URSim container; skips if Docker or the image is unavailable.""" if not check_docker_available(): pytest.skip("Docker is not available or not running") @@ -279,22 +275,12 @@ def ursim_container(): error_msg = e.stderr.decode() if e.stderr else str(e) pytest.skip(f"Failed to start URSim container: {error_msg}") except TimeoutError as e: - pytest.skip(f"URSim container startup timeout: {e}") + pytest.fail(f"URSim container startup timeout: {e}") @pytest.fixture(scope="session") def ur5_robot(ursim_container): - """Create UR5 robot connection with enhanced retry logic. - - Args: - ursim_container: URSim container fixture. - - Yields: - UR5: Connected robot instance. - - Raises: - pytest.skip: If robot connection fails. - """ + """Yield a UR5 connected to URSim; fails if the connection cannot be verified.""" from armctl import UR5 @retry_on_failure(max_retries=5, delay=1) @@ -322,8 +308,7 @@ def connect_robot(): robot = connect_robot() yield robot except Exception as e: - logger.error(f"Failed to connect to URSim robot: {e}") - pytest.skip(f"Failed to connect to URSim robot: {str(e)}") + pytest.fail(f"Failed to connect to URSim robot: {e}") finally: if robot: try: