test: overhaul test suite

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()

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)"]

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 "")

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