From a9c86ef65c8d6653f0ccf88828f3c49e5595ef06 Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Mon, 13 Apr 2026 18:25:23 +0530 Subject: [PATCH 1/4] feat: add Linux AT-SPI cursorless automation with ydotool fallback [#29] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements LinuxATSPIAutomation in operator_use/computer/linux/atapi.py: - click(app_name, element_name): traverses AT-SPI tree, invokes Action interface; falls back to ydotool click when pyatspi unavailable or element not found - type_text(app_name, element_name, text): sets text via EditableText/Value AT-SPI interfaces; falls back to ydotool type - is_available(): returns True when pyatspi D-Bus session reachable or ydotool binary present; returns False on non-Linux platforms All dependencies are optional — pyatspi import is wrapped in try/except, module never hard-fails at import time. ydotool is a system tool documented under the new [linux-automation] optional-dependency group in pyproject.toml. 27 tests added in tests/test_linux_atapi.py covering AT-SPI primary path, ydotool fallback, is_available() variants, and cross-platform import guard. --- operator_use/computer/linux/atapi.py | 291 +++++++++++++++++++ pyproject.toml | 18 ++ tests/test_linux_atapi.py | 410 +++++++++++++++++++++++++++ uv.lock | 2 +- 4 files changed, 720 insertions(+), 1 deletion(-) create mode 100644 operator_use/computer/linux/atapi.py create mode 100644 tests/test_linux_atapi.py diff --git a/operator_use/computer/linux/atapi.py b/operator_use/computer/linux/atapi.py new file mode 100644 index 0000000..d840a6a --- /dev/null +++ b/operator_use/computer/linux/atapi.py @@ -0,0 +1,291 @@ +"""Linux AT-SPI cursorless automation via the D-Bus accessibility API. + +Primary path: pyatspi (AT-SPI2) — traverses the accessibility tree to find +UI elements and invokes the Action interface directly, without simulating +pointer/keyboard events. + +Fallback path: ydotool — a Wayland-compatible input injection tool that +runs as a privileged daemon. Used when pyatspi is unavailable or when the +target element cannot be located in the AT-SPI tree. + +Installation +------------ +AT-SPI (primary): + pip install "operator-use[linux-automation]" + # requires the AT-SPI2 D-Bus accessibility service to be running + +ydotool (fallback system tool — NOT a Python package): + sudo apt install ydotool # Debian/Ubuntu + sudo dnf install ydotool # Fedora + sudo pacman -S ydotool # Arch + # then start the daemon: sudo systemctl enable --now ydotoold + +Platform guard +-------------- +This module can be *imported* on any OS — all Linux-specific imports are lazy. +LinuxATSPIAutomation.is_available() will simply return False on non-Linux +systems, and the action methods will raise RuntimeError immediately. + +Usage +----- + from operator_use.computer.linux.atapi import LinuxATSPIAutomation + automation = LinuxATSPIAutomation() + automation.click("gedit", "Open") + automation.type_text("gedit", "text-field", "Hello, world!") +""" + +from __future__ import annotations + +import shutil +import subprocess +import sys +from typing import Any + +# --------------------------------------------------------------------------- +# Optional pyatspi import — never fails at module load time +# --------------------------------------------------------------------------- + +_pyatspi: Any = None +_pyatspi_available: bool = False + +try: + import pyatspi as _pyatspi # type: ignore[import-untyped] + + _pyatspi_available = True +except Exception: # ImportError, OSError, D-Bus errors … + pass + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + + +def _is_linux() -> bool: + return sys.platform == "linux" + + +def _ydotool_available() -> bool: + """Return True if the *ydotool* binary is on PATH.""" + return shutil.which("ydotool") is not None + + +def _find_element(app_name: str, element_name: str) -> Any | None: + """Traverse the AT-SPI tree and return the first matching element. + + Returns *None* if pyatspi is unavailable, the named application cannot be + found in the registry, or no child element matches *element_name*. + """ + if not _pyatspi_available or _pyatspi is None: + return None + + try: + desktop = _pyatspi.Registry.getDesktop(0) + except Exception: + return None + + # Locate the application by name (case-insensitive). + app_node = None + for i in range(desktop.childCount): + try: + child = desktop.getChildAtIndex(i) + if child and child.name.lower() == app_name.lower(): + app_node = child + break + except Exception: + continue + + if app_node is None: + return None + + # BFS through all children looking for the element by name. + queue: list[Any] = [app_node] + while queue: + node = queue.pop(0) + try: + if node.name and node.name.lower() == element_name.lower(): + return node + for i in range(node.childCount): + try: + queue.append(node.getChildAtIndex(i)) + except Exception: + pass + except Exception: + continue + + return None + + +def _ydotool_click() -> None: + """Issue a left-click at the current pointer position via ydotool.""" + subprocess.run( + ["ydotool", "click", "0xC0"], + check=True, + capture_output=True, + ) + + +def _ydotool_type(text: str) -> None: + """Type *text* via ydotool.""" + subprocess.run( + ["ydotool", "type", "--", text], + check=True, + capture_output=True, + ) + + +# --------------------------------------------------------------------------- +# Public automation class +# --------------------------------------------------------------------------- + + +class LinuxATSPIAutomation: + """Cursorless desktop automation for Linux using AT-SPI with ydotool fallback. + + All public methods are safe to call on non-Linux systems — they raise + RuntimeError immediately rather than crashing at import time. + """ + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def click(self, app_name: str, element_name: str) -> None: + """Click *element_name* inside *app_name*. + + Strategy + -------- + 1. If pyatspi is available: find the element in the AT-SPI tree and + invoke its default Action (typically "click" / "press"). + 2. If pyatspi is unavailable or element is not found: fall back to + ``ydotool click 0xC0`` (left-button down+up at current cursor). + + Parameters + ---------- + app_name: + Accessible name of the target application (e.g. ``"gedit"``). + element_name: + Accessible name of the target element (e.g. ``"Open"``). + + Raises + ------ + RuntimeError + On non-Linux platforms, or when neither pyatspi nor ydotool is + available. + """ + if not _is_linux(): + raise RuntimeError("LinuxATSPIAutomation is only available on Linux.") + + # Primary: AT-SPI + if _pyatspi_available: + try: + element = _find_element(app_name, element_name) + if element is not None: + action = element.queryAction() + # Prefer an action named "click" or "press"; fall back to index 0. + target_index = 0 + for i in range(action.nActions): + name = action.getName(i).lower() + if name in ("click", "press", "activate"): + target_index = i + break + action.doAction(target_index) + return + except Exception: + pass # Fall through to ydotool + + # Fallback: ydotool + if not _ydotool_available(): + raise RuntimeError( + "Neither pyatspi nor ydotool is available. " + "Install pyatspi>=2.46.0 or the ydotool system package." + ) + _ydotool_click() + + def type_text(self, app_name: str, element_name: str, text: str) -> None: + """Set the text of *element_name* inside *app_name* to *text*. + + Strategy + -------- + 1. If pyatspi is available: locate the element and call + ``setText()`` via the AT-SPI Text interface (or ``queryEditableText`` + on older pyatspi versions). + 2. Fallback: ``ydotool type -- ``. + + Parameters + ---------- + app_name: + Accessible name of the target application. + element_name: + Accessible name of the target element (e.g. a text field). + text: + The string to insert. + + Raises + ------ + RuntimeError + On non-Linux platforms, or when neither pyatspi nor ydotool is + available. + """ + if not _is_linux(): + raise RuntimeError("LinuxATSPIAutomation is only available on Linux.") + + # Primary: AT-SPI + if _pyatspi_available: + try: + element = _find_element(app_name, element_name) + if element is not None: + # Try EditableText interface first (most editable widgets). + try: + editable = element.queryEditableText() + editable.setTextContents(text) + return + except Exception: + pass + # Try Value interface for spinners / sliders. + try: + value = element.queryValue() + value.currentValue = float(text) + return + except Exception: + pass + except Exception: + pass # Fall through to ydotool + + # Fallback: ydotool + if not _ydotool_available(): + raise RuntimeError( + "Neither pyatspi nor ydotool is available. " + "Install pyatspi>=2.46.0 or the ydotool system package." + ) + _ydotool_type(text) + + # ------------------------------------------------------------------ + # Availability check + # ------------------------------------------------------------------ + + @staticmethod + def is_available() -> bool: + """Return True if at least one automation backend is accessible. + + Checks + ------ + - pyatspi can be imported **and** ``Registry.getDesktop(0)`` succeeds + (i.e. the AT-SPI D-Bus service is running), *or* + - the ``ydotool`` binary is on PATH. + + Returns False on non-Linux platforms. + """ + if not _is_linux(): + return False + + # Check pyatspi with a live D-Bus probe. + if _pyatspi_available and _pyatspi is not None: + try: + _pyatspi.Registry.getDesktop(0) + return True + except Exception: + pass + + # Check ydotool binary. + return _ydotool_available() diff --git a/pyproject.toml b/pyproject.toml index 111b2a5..35102bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -65,6 +65,24 @@ exa = [ tavily = [ "tavily-python>=0.5.0", ] +linux-automation = [ + # AT-SPI D-Bus accessibility API — primary cursorless automation backend. + # + # pyatspi is distributed as a SYSTEM package only — it is NOT on PyPI. + # Install it with your Linux system package manager: + # sudo apt install python3-atspi # Debian / Ubuntu (provides pyatspi) + # sudo dnf install python3-atspi # Fedora / RHEL + # sudo pacman -S python-atspi # Arch Linux + # + # ydotool is the Wayland fallback — also a system tool, not a Python package: + # sudo apt install ydotool # Debian / Ubuntu + # sudo systemctl enable --now ydotoold # start the required ydotool daemon + # + # This group intentionally has no pip-installable entries because both + # dependencies must be provided by the system. The [linux-automation] + # extra exists purely as a documentation marker for `pip install + # operator-use[linux-automation]` to surface the install instructions. +] [tool.pytest.ini_options] asyncio_mode = "strict" diff --git a/tests/test_linux_atapi.py b/tests/test_linux_atapi.py new file mode 100644 index 0000000..a0a7d80 --- /dev/null +++ b/tests/test_linux_atapi.py @@ -0,0 +1,410 @@ +"""Tests for operator_use.computer.linux.atapi. + +The pyatspi library is NOT installed in CI — all AT-SPI interactions are +mocked at the module level using unittest.mock. Each test class patches the +relevant internals so we can verify exact code paths without a D-Bus session. +""" + +from __future__ import annotations + +import subprocess +from types import ModuleType +from unittest import mock + +import pytest + +# --------------------------------------------------------------------------- +# Helpers: create a minimal fake pyatspi hierarchy +# --------------------------------------------------------------------------- + + +def _make_action(names: list[str], *, raises: bool = False) -> mock.MagicMock: + """Return a fake AT-SPI Action proxy.""" + action = mock.MagicMock() + if raises: + action.doAction.side_effect = RuntimeError("AT-SPI action error") + action.nActions = len(names) + action.getName.side_effect = lambda i: names[i] + return action + + +def _make_element( + name: str, + children: list | None = None, + *, + action: mock.MagicMock | None = None, + editable_text: mock.MagicMock | None = None, + value_iface: mock.MagicMock | None = None, + no_action: bool = False, +) -> mock.MagicMock: + children = children or [] + elem = mock.MagicMock() + elem.name = name + elem.childCount = len(children) + elem.getChildAtIndex.side_effect = lambda i: children[i] + + if no_action: + elem.queryAction.side_effect = Exception("no action iface") + else: + elem.queryAction.return_value = action or _make_action(["click"]) + + if editable_text is not None: + elem.queryEditableText.return_value = editable_text + else: + elem.queryEditableText.side_effect = Exception("no editable text") + + if value_iface is not None: + elem.queryValue.return_value = value_iface + else: + elem.queryValue.side_effect = Exception("no value iface") + + return elem + + +def _make_desktop(apps: list[mock.MagicMock]) -> mock.MagicMock: + desktop = mock.MagicMock() + desktop.childCount = len(apps) + desktop.getChildAtIndex.side_effect = lambda i: apps[i] + return desktop + + +def _make_fake_pyatspi(desktop: mock.MagicMock) -> ModuleType: + """Return a minimal fake pyatspi module.""" + mod = ModuleType("pyatspi") + registry = mock.MagicMock() + registry.getDesktop.return_value = desktop + mod.Registry = registry + return mod + + +# --------------------------------------------------------------------------- +# 1. Import tests +# --------------------------------------------------------------------------- + + +class TestImportBehaviour: + """The module must be importable regardless of platform or pyatspi status.""" + + def test_import_succeeds_on_current_platform(self): + """Module imports cleanly; pyatspi absence is silently handled.""" + + import operator_use.computer.linux.atapi as m + + assert m.LinuxATSPIAutomation is not None + + def test_import_succeeds_when_pyatspi_missing(self, monkeypatch): + """Simulated missing pyatspi must not raise at import time.""" + import operator_use.computer.linux.atapi as m + + # Patch the module-level flag as if pyatspi was never importable. + monkeypatch.setattr(m, "_pyatspi_available", False) + monkeypatch.setattr(m, "_pyatspi", None) + # Re-instantiating the class must still work. + auto = m.LinuxATSPIAutomation() + assert auto is not None + + def test_linux_automation_class_exported(self): + from operator_use.computer.linux.atapi import LinuxATSPIAutomation + + assert callable(LinuxATSPIAutomation) + + def test_module_level_flag_types(self): + import operator_use.computer.linux.atapi as m + + assert isinstance(m._pyatspi_available, bool) + + +# --------------------------------------------------------------------------- +# 2. is_available() tests +# --------------------------------------------------------------------------- + + +class TestIsAvailable: + """is_available() must accurately reflect which backends are reachable.""" + + def test_returns_false_on_non_linux(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: False) + assert m.LinuxATSPIAutomation.is_available() is False + + def test_returns_true_when_pyatspi_desktop_ok(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + fake_pyatspi = mock.MagicMock() + fake_pyatspi.Registry.getDesktop.return_value = mock.MagicMock() + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", True) + monkeypatch.setattr(m, "_pyatspi", fake_pyatspi) + assert m.LinuxATSPIAutomation.is_available() is True + + def test_returns_false_when_pyatspi_dbus_fails_and_no_ydotool(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + fake_pyatspi = mock.MagicMock() + fake_pyatspi.Registry.getDesktop.side_effect = Exception("no D-Bus") + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", True) + monkeypatch.setattr(m, "_pyatspi", fake_pyatspi) + monkeypatch.setattr(m, "_ydotool_available", lambda: False) + assert m.LinuxATSPIAutomation.is_available() is False + + def test_returns_true_when_only_ydotool_present(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", False) + monkeypatch.setattr(m, "_pyatspi", None) + monkeypatch.setattr(m, "_ydotool_available", lambda: True) + assert m.LinuxATSPIAutomation.is_available() is True + + def test_returns_false_when_neither_present(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", False) + monkeypatch.setattr(m, "_pyatspi", None) + monkeypatch.setattr(m, "_ydotool_available", lambda: False) + assert m.LinuxATSPIAutomation.is_available() is False + + +# --------------------------------------------------------------------------- +# 3. click() — AT-SPI primary path +# --------------------------------------------------------------------------- + + +class TestClickATSPI: + """click() must traverse the tree and invoke the action interface.""" + + def _setup(self, monkeypatch, action: mock.MagicMock | None = None): + import operator_use.computer.linux.atapi as m + + act = action or _make_action(["click"]) + target = _make_element("Open", action=act) + app = _make_element("gedit", children=[target]) + desktop = _make_desktop([app]) + fake_pyatspi = _make_fake_pyatspi(desktop) + + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", True) + monkeypatch.setattr(m, "_pyatspi", fake_pyatspi) + return m, act + + def test_click_succeeds_via_atspi(self, monkeypatch): + m, act = self._setup(monkeypatch) + m.LinuxATSPIAutomation().click("gedit", "Open") + act.doAction.assert_called_once_with(0) + + def test_click_prefers_action_named_click(self, monkeypatch): + """When multiple actions exist, the one named 'click' is preferred.""" + act = _make_action(["focus", "click", "press"]) + m, _ = self._setup(monkeypatch, action=act) + m.LinuxATSPIAutomation().click("gedit", "Open") + act.doAction.assert_called_once_with(1) # index 1 == "click" + + def test_click_selects_press_action(self, monkeypatch): + """'press' is also a recognised action name.""" + act = _make_action(["focus", "press"]) + m, _ = self._setup(monkeypatch, action=act) + m.LinuxATSPIAutomation().click("gedit", "Open") + act.doAction.assert_called_once_with(1) + + def test_click_falls_back_to_index_zero_for_unknown_action(self, monkeypatch): + act = _make_action(["do-something-weird"]) + m, _ = self._setup(monkeypatch, action=act) + m.LinuxATSPIAutomation().click("gedit", "Open") + act.doAction.assert_called_once_with(0) + + def test_click_raises_on_non_linux(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: False) + with pytest.raises(RuntimeError, match="only available on Linux"): + m.LinuxATSPIAutomation().click("gedit", "Open") + + +# --------------------------------------------------------------------------- +# 4. click() — ydotool fallback path +# --------------------------------------------------------------------------- + + +class TestClickYdotoolFallback: + """click() must call ydotool when pyatspi is absent or element not found.""" + + def test_click_falls_back_to_ydotool_when_pyatspi_raises(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + # pyatspi present but Registry raises + fake_pyatspi = mock.MagicMock() + fake_pyatspi.Registry.getDesktop.side_effect = Exception("D-Bus error") + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", True) + monkeypatch.setattr(m, "_pyatspi", fake_pyatspi) + monkeypatch.setattr(m, "_ydotool_available", lambda: True) + + with mock.patch("subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess([], 0) + m.LinuxATSPIAutomation().click("gedit", "Open") + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + assert args[0] == "ydotool" + assert args[1] == "click" + + def test_click_falls_back_when_pyatspi_unavailable(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", False) + monkeypatch.setattr(m, "_pyatspi", None) + monkeypatch.setattr(m, "_ydotool_available", lambda: True) + + with mock.patch("subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess([], 0) + m.LinuxATSPIAutomation().click("gedit", "Open") + mock_run.assert_called_once() + + def test_click_falls_back_when_element_not_found(self, monkeypatch): + """Element lookup returns None → falls back to ydotool.""" + import operator_use.computer.linux.atapi as m + + app = _make_element("gedit", children=[]) # no children → element not found + desktop = _make_desktop([app]) + fake_pyatspi = _make_fake_pyatspi(desktop) + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", True) + monkeypatch.setattr(m, "_pyatspi", fake_pyatspi) + monkeypatch.setattr(m, "_ydotool_available", lambda: True) + + with mock.patch("subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess([], 0) + m.LinuxATSPIAutomation().click("gedit", "NonExistentButton") + mock_run.assert_called_once() + + def test_click_raises_when_no_backend_available(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", False) + monkeypatch.setattr(m, "_pyatspi", None) + monkeypatch.setattr(m, "_ydotool_available", lambda: False) + + with pytest.raises(RuntimeError, match="Neither pyatspi nor ydotool"): + m.LinuxATSPIAutomation().click("gedit", "Open") + + +# --------------------------------------------------------------------------- +# 5. type_text() — AT-SPI primary path +# --------------------------------------------------------------------------- + + +class TestTypeTextATSPI: + """type_text() must use EditableText / Value interfaces when AT-SPI available.""" + + def test_type_text_uses_editable_text(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + editable = mock.MagicMock() + target = _make_element("text-field", editable_text=editable) + app = _make_element("gedit", children=[target]) + desktop = _make_desktop([app]) + fake_pyatspi = _make_fake_pyatspi(desktop) + + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", True) + monkeypatch.setattr(m, "_pyatspi", fake_pyatspi) + + m.LinuxATSPIAutomation().type_text("gedit", "text-field", "hello") + editable.setTextContents.assert_called_once_with("hello") + + def test_type_text_raises_on_non_linux(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: False) + with pytest.raises(RuntimeError, match="only available on Linux"): + m.LinuxATSPIAutomation().type_text("gedit", "text-field", "hello") + + +# --------------------------------------------------------------------------- +# 6. type_text() — ydotool fallback path +# --------------------------------------------------------------------------- + + +class TestTypeTextYdotoolFallback: + def test_type_text_falls_back_to_ydotool_when_pyatspi_raises(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + fake_pyatspi = mock.MagicMock() + fake_pyatspi.Registry.getDesktop.side_effect = Exception("D-Bus error") + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", True) + monkeypatch.setattr(m, "_pyatspi", fake_pyatspi) + monkeypatch.setattr(m, "_ydotool_available", lambda: True) + + with mock.patch("subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess([], 0) + m.LinuxATSPIAutomation().type_text("gedit", "text-field", "world") + mock_run.assert_called_once() + args = mock_run.call_args[0][0] + assert args[0] == "ydotool" + assert args[1] == "type" + assert "world" in args + + def test_type_text_falls_back_when_pyatspi_unavailable(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", False) + monkeypatch.setattr(m, "_pyatspi", None) + monkeypatch.setattr(m, "_ydotool_available", lambda: True) + + with mock.patch("subprocess.run") as mock_run: + mock_run.return_value = subprocess.CompletedProcess([], 0) + m.LinuxATSPIAutomation().type_text("gedit", "text-field", "foo") + mock_run.assert_called_once() + + def test_type_text_raises_when_no_backend_available(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: True) + monkeypatch.setattr(m, "_pyatspi_available", False) + monkeypatch.setattr(m, "_pyatspi", None) + monkeypatch.setattr(m, "_ydotool_available", lambda: False) + + with pytest.raises(RuntimeError, match="Neither pyatspi nor ydotool"): + m.LinuxATSPIAutomation().type_text("gedit", "text-field", "bar") + + +# --------------------------------------------------------------------------- +# 7. Cross-platform import guard +# --------------------------------------------------------------------------- + + +class TestNonLinuxPlatform: + """On non-Linux platforms the module must import fine but do nothing.""" + + def test_import_does_not_crash_on_non_linux(self): + """Import must succeed regardless of sys.platform value.""" + # This test always passes — we already imported successfully above. + import operator_use.computer.linux.atapi # noqa: F401 + + def test_is_available_false_on_non_linux(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: False) + assert m.LinuxATSPIAutomation.is_available() is False + + def test_click_raises_runtime_on_non_linux(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: False) + auto = m.LinuxATSPIAutomation() + with pytest.raises(RuntimeError): + auto.click("app", "elem") + + def test_type_text_raises_runtime_on_non_linux(self, monkeypatch): + import operator_use.computer.linux.atapi as m + + monkeypatch.setattr(m, "_is_linux", lambda: False) + auto = m.LinuxATSPIAutomation() + with pytest.raises(RuntimeError): + auto.type_text("app", "elem", "text") diff --git a/uv.lock b/uv.lock index 6449804..35abe88 100644 --- a/uv.lock +++ b/uv.lock @@ -1901,7 +1901,7 @@ requires-dist = [ { name = "twitchio", specifier = ">=2.0.0,<3.0.0" }, { name = "typer", specifier = ">=0.24.1" }, ] -provides-extras = ["dev", "fal", "exa", "tavily"] +provides-extras = ["dev", "fal", "exa", "tavily", "linux-automation"] [[package]] name = "packaging" From b218dbaa0b01be62b2238c87c27baea06eda83cd Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Sun, 19 Apr 2026 21:48:07 +0530 Subject: [PATCH 2/4] fix: rebase onto main with plugin fixes included [ci] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - test_control_center.py: update import from operator_use.agent.tools.builtin (deleted in main) to operator_use.tools.control_center; add side-effect import of operator_use.agent.tools to resolve circular import; update all patch paths. - test_computer_plugin.py: remove BEFORE_LLM_CALL assertions — 9f5d002 removed _state_hook registration from BEFORE_LLM_CALL by design (state captured inside computer_task loop, not on every main-agent LLM call). - test_browser_plugin.py: same BEFORE_LLM_CALL removal; register_hooks now only stores the hooks reference without registering any handler. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- tests/test_browser_plugin.py | 17 ++++++++++----- tests/test_computer_plugin.py | 8 +++---- tests/test_control_center.py | 39 ++++++++++++++++++----------------- 3 files changed, 35 insertions(+), 29 deletions(-) diff --git a/tests/test_browser_plugin.py b/tests/test_browser_plugin.py index 9089d95..41e4c8b 100644 --- a/tests/test_browser_plugin.py +++ b/tests/test_browser_plugin.py @@ -61,7 +61,9 @@ def test_unregister_tools_removes_browser_tool(): # --------------------------------------------------------------------------- -# register_hooks — BEFORE_LLM_CALL gated on _enabled +# register_hooks — no hooks registered on BEFORE_LLM_CALL +# (BEFORE_LLM_CALL registration was removed in 9f5d002: browser state is +# now captured inside the browser_task loop, not on every main-agent LLM call) # --------------------------------------------------------------------------- @@ -69,7 +71,8 @@ def test_disabled_plugin_registers_no_hooks(): plugin = BrowserPlugin(enabled=False) hooks = Hooks() plugin.register_hooks(hooks) - assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] + # register_hooks stores the hooks reference but does not register any handler + assert plugin._hooks is hooks def test_enabled_plugin_registers_state_hook(): @@ -77,7 +80,8 @@ def test_enabled_plugin_registers_state_hook(): plugin._enabled = True hooks = Hooks() plugin.register_hooks(hooks) - assert plugin._state_hook in hooks._handlers[HookEvent.BEFORE_LLM_CALL] + # No hook is registered on BEFORE_LLM_CALL; hook reference is stored only + assert plugin._hooks is hooks def test_unregister_hooks_removes_state_hook(): @@ -86,7 +90,8 @@ def test_unregister_hooks_removes_state_hook(): hooks = Hooks() plugin.register_hooks(hooks) plugin.unregister_hooks(hooks) - assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] + # unregister_hooks is a no-op; _hooks reference is preserved + assert plugin._hooks is hooks # --------------------------------------------------------------------------- @@ -135,7 +140,8 @@ async def test_enable_registers_hooks_and_injects_prompt(): await plugin.enable() assert plugin._enabled is True - assert plugin._state_hook in hooks._handlers[HookEvent.BEFORE_LLM_CALL] + # _state_hook is not registered on BEFORE_LLM_CALL (see 9f5d002) + assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] context.register_plugin_prompt.assert_called_once_with(SYSTEM_PROMPT) @@ -151,6 +157,7 @@ async def test_disable_unregisters_hooks_and_removes_prompt(): await plugin.disable() assert plugin._enabled is False + # _state_hook was never in BEFORE_LLM_CALL (see 9f5d002) assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] context.unregister_plugin_prompt.assert_called_once_with(SYSTEM_PROMPT) diff --git a/tests/test_computer_plugin.py b/tests/test_computer_plugin.py index 47e4480..68eb277 100644 --- a/tests/test_computer_plugin.py +++ b/tests/test_computer_plugin.py @@ -34,7 +34,9 @@ def test_enabled_plugin_returns_system_prompt(): # --------------------------------------------------------------------------- -# register_hooks — BEFORE_LLM_CALL + AFTER_TOOL_CALL, gated on _enabled +# register_hooks — AFTER_TOOL_CALL only, gated on _enabled +# (BEFORE_LLM_CALL was removed in 9f5d002: state captured inside +# computer_task loop, not on every main-agent LLM call) # --------------------------------------------------------------------------- @@ -42,7 +44,6 @@ def test_disabled_plugin_registers_no_hooks(): plugin = ComputerPlugin(enabled=False) hooks = Hooks() plugin.register_hooks(hooks) - assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] assert plugin._wait_for_ui_hook not in hooks._handlers[HookEvent.AFTER_TOOL_CALL] @@ -51,7 +52,6 @@ def test_enabled_plugin_registers_both_hooks(): plugin._enabled = True hooks = Hooks() plugin.register_hooks(hooks) - assert plugin._state_hook in hooks._handlers[HookEvent.BEFORE_LLM_CALL] assert plugin._wait_for_ui_hook in hooks._handlers[HookEvent.AFTER_TOOL_CALL] @@ -61,7 +61,6 @@ def test_unregister_hooks_removes_both(): hooks = Hooks() plugin.register_hooks(hooks) plugin.unregister_hooks(hooks) - assert plugin._state_hook not in hooks._handlers[HookEvent.BEFORE_LLM_CALL] assert plugin._wait_for_ui_hook not in hooks._handlers[HookEvent.AFTER_TOOL_CALL] @@ -111,7 +110,6 @@ async def test_enable_registers_both_hooks_and_prompt(): await plugin.enable() assert plugin._enabled is True - assert plugin._state_hook in hooks._handlers[HookEvent.BEFORE_LLM_CALL] assert plugin._wait_for_ui_hook in hooks._handlers[HookEvent.AFTER_TOOL_CALL] context.register_plugin_prompt.assert_called_once_with(SYSTEM_PROMPT) diff --git a/tests/test_control_center.py b/tests/test_control_center.py index f3a2e5b..158d84c 100644 --- a/tests/test_control_center.py +++ b/tests/test_control_center.py @@ -4,12 +4,17 @@ import pytest from unittest.mock import AsyncMock, MagicMock, patch -from operator_use.agent.tools.builtin.control_center import ( +# Import via operator_use.agent.tools so the circular-import chain is resolved +# in the correct order before we access the module directly. +import operator_use.agent.tools # noqa: F401 — side-effect import resolves cycle +from operator_use.tools.control_center import ( control_center, _set_plugin_enabled, _get_plugin_enabled, ) +_MODULE = "operator_use.tools.control_center" + # --------------------------------------------------------------------------- # Helpers @@ -73,7 +78,7 @@ async def test_enable_browser_use_calls_agent(tmp_path): mock_agent = MagicMock() mock_agent.enable_browser_use = AsyncMock() - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): result = await _call_cc(browser_use=True, _agent=mock_agent) mock_agent.enable_browser_use.assert_awaited_once() @@ -90,7 +95,7 @@ async def test_enable_both_computer_use_and_browser_use_independently(tmp_path): mock_agent.enable_computer_use = AsyncMock() mock_agent.enable_browser_use = AsyncMock() - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): result = await _call_cc(computer_use=True, browser_use=True, _agent=mock_agent) saved = json.loads(cfg_file.read_text()) @@ -111,7 +116,7 @@ async def test_disable_browser_use_calls_agent(tmp_path): mock_agent = MagicMock() mock_agent.disable_browser_use = AsyncMock() - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): result = await _call_cc(browser_use=False, _agent=mock_agent) mock_agent.disable_browser_use.assert_awaited_once() @@ -124,7 +129,7 @@ async def test_status_only_returns_current_state(tmp_path): cfg_file = tmp_path / "config.json" cfg_file.write_text(json.dumps(cfg)) - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): result = await _call_cc() assert result.success @@ -148,10 +153,8 @@ async def test_audit_log_emitted_on_plugin_change(tmp_path, caplog): mock_agent = MagicMock() mock_agent.enable_browser_use = AsyncMock() - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): - with caplog.at_level( - logging.WARNING, logger="operator_use.agent.tools.builtin.control_center" - ): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): + with caplog.at_level(logging.WARNING, logger=_MODULE): await _call_cc( browser_use=True, _agent=mock_agent, @@ -172,10 +175,8 @@ async def test_audit_log_emitted_on_status_check(tmp_path, caplog): cfg_file = tmp_path / "config.json" cfg_file.write_text(json.dumps(cfg)) - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): - with caplog.at_level( - logging.WARNING, logger="operator_use.agent.tools.builtin.control_center" - ): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): + with caplog.at_level(logging.WARNING, logger=_MODULE): await _call_cc(_channel="discord", _chat_id="999", _agent_id="op") assert any("control_center" in r.message for r in caplog.records) @@ -197,8 +198,8 @@ async def test_restart_calls_graceful_fn_not_os_exit(tmp_path): async def mock_graceful(): graceful_called.append(True) - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): - with patch("operator_use.agent.tools.builtin.control_center._do_restart") as mock_restart: + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}._do_restart") as mock_restart: mock_restart.return_value = None result = await _call_cc(restart=True, _graceful_restart_fn=mock_graceful) @@ -217,8 +218,8 @@ async def test_restart_without_graceful_fn_still_works(tmp_path): cfg_file = tmp_path / "config.json" cfg_file.write_text(json.dumps(cfg)) - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): - with patch("operator_use.agent.tools.builtin.control_center._do_restart"): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}._do_restart"): result = await _call_cc(restart=True) assert result.success @@ -234,7 +235,7 @@ async def test_returns_error_when_no_agents(tmp_path): cfg_file = tmp_path / "config.json" cfg_file.write_text(json.dumps({"agents": {"list": []}})) - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", cfg_file): + with patch(f"{_MODULE}.CONFIG_PATH", cfg_file): result = await _call_cc(browser_use=True) assert not result.success @@ -244,7 +245,7 @@ async def test_returns_error_when_no_agents(tmp_path): async def test_returns_error_when_config_missing(tmp_path): missing = tmp_path / "no_config.json" - with patch("operator_use.agent.tools.builtin.control_center.CONFIG_PATH", missing): + with patch(f"{_MODULE}.CONFIG_PATH", missing): result = await _call_cc(browser_use=True) assert not result.success From 75e2daaef5c23bfa6b8b4ab023709c1d9179245c Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Sun, 19 Apr 2026 22:01:53 +0530 Subject: [PATCH 3/4] fix: update test imports for refactored tools paths [ci] --- tests/test_local_agents.py | 2 +- tests/test_plugins.py | 2 +- tests/test_tool_registry.py | 2 +- tests/test_tools.py | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_local_agents.py b/tests/test_local_agents.py index 8fd831b..a1b5168 100644 --- a/tests/test_local_agents.py +++ b/tests/test_local_agents.py @@ -2,7 +2,7 @@ import pytest -from operator_use.agent.tools.builtin.local_agents import LOCAL_AGENT_DELEGATION_CHAIN, localagents +from operator_use.tools.local_agents import LOCAL_AGENT_DELEGATION_CHAIN, localagents from operator_use.messages.service import AIMessage diff --git a/tests/test_plugins.py b/tests/test_plugins.py index f6ba6d4..5d9f8b9 100644 --- a/tests/test_plugins.py +++ b/tests/test_plugins.py @@ -7,7 +7,7 @@ from operator_use.agent.tools.registry import ToolRegistry from operator_use.agent.hooks.service import Hooks from operator_use.agent.hooks.events import HookEvent -from operator_use.tools.service import Tool +from operator_use.agent.tools.service import Tool from pydantic import BaseModel diff --git a/tests/test_tool_registry.py b/tests/test_tool_registry.py index ca6ed75..77c70b9 100644 --- a/tests/test_tool_registry.py +++ b/tests/test_tool_registry.py @@ -4,7 +4,7 @@ from pydantic import BaseModel from operator_use.agent.tools.registry import ToolRegistry -from operator_use.tools.service import Tool +from operator_use.agent.tools.service import Tool # --- Helpers --- diff --git a/tests/test_tools.py b/tests/test_tools.py index 8cbf913..de572ab 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -4,7 +4,7 @@ from pydantic import BaseModel from typing import Literal -from operator_use.tools.service import Tool, ToolResult +from operator_use.agent.tools.service import Tool, ToolResult # --- ToolResult --- From 2b33468dd513e589e04a55cd65dd19c2dd001477 Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Sun, 19 Apr 2026 22:10:31 +0530 Subject: [PATCH 4/4] fix: fix remaining test_agent.py and e2e imports for refactored tools [ci] --- tests/test_agent.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_agent.py b/tests/test_agent.py index 4fb6c3f..13db174 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -186,7 +186,7 @@ async def test_agent_run_with_tool_call_then_text(tmp_path): # Register a simple echo tool from pydantic import BaseModel - from operator_use.tools.service import Tool + from operator_use.agent.tools.service import Tool class EchoParams(BaseModel): message: str