diff --git a/src/askui/models/shared/tools.py b/src/askui/models/shared/tools.py index 2d72b513..63d3e4eb 100644 --- a/src/askui/models/shared/tools.py +++ b/src/askui/models/shared/tools.py @@ -1,5 +1,5 @@ from abc import ABC, abstractmethod -from typing import Any, cast +from typing import Any from anthropic.types.beta import BetaToolParam, BetaToolUnionParam from anthropic.types.beta.beta_tool_param import InputSchema @@ -147,10 +147,7 @@ def _run_tool( tool_use_id=tool_use_block_param.id, ) try: - tool_result: ToolCallResult = cast( - "ToolCallResult", - tool(**tool_use_block_param.input), # type: ignore - ) + tool_result: ToolCallResult = tool(**tool_use_block_param.input) # type: ignore return ToolResultBlockParam( content=_convert_to_content(tool_result), tool_use_id=tool_use_block_param.id, diff --git a/src/askui/tools/askui/askui_controller.py b/src/askui/tools/askui/askui_controller.py index 2ff1f2d8..a08937ea 100644 --- a/src/askui/tools/askui/askui_controller.py +++ b/src/askui/tools/askui/askui_controller.py @@ -8,14 +8,13 @@ import grpc from PIL import Image -from pydantic import BaseModel, Field, model_validator -from pydantic_settings import BaseSettings, SettingsConfigDict from typing_extensions import Self, override from askui.container import telemetry from askui.logger import logger from askui.reporting import Reporter from askui.tools.agent_os import AgentOs, Coordinate, ModifierKey, PcKey +from askui.tools.askui.askui_controller_settings import AskUiControllerSettings from askui.tools.askui.askui_ui_controller_grpc.generated import ( Controller_V1_pb2 as controller_v1_pbs, ) @@ -48,137 +47,19 @@ ) -class RemoteDeviceController(BaseModel): - askui_remote_device_controller: pathlib.Path = Field( - alias="AskUIRemoteDeviceController" - ) - - -class Executables(BaseModel): - executables: RemoteDeviceController = Field(alias="Executables") - - -class InstalledPackages(BaseModel): - remote_device_controller_uuid: Executables = Field( - alias="{aed1b543-e856-43ad-b1bc-19365d35c33e}" - ) - - -class AskUiComponentRegistry(BaseModel): - definition_version: int = Field(alias="DefinitionVersion") - installed_packages: InstalledPackages = Field(alias="InstalledPackages") - - -class AskUiControllerSettings(BaseSettings): - model_config = SettingsConfigDict( - env_prefix="ASKUI_", - ) - - component_registry_file: pathlib.Path | None = None - installation_directory: pathlib.Path | None = None - - @model_validator(mode="after") - def validate_either_component_registry_or_installation_directory_is_set( - self, - ) -> "AskUiControllerSettings": - if self.component_registry_file is None and self.installation_directory is None: - error_msg = ( - "Either ASKUI_COMPONENT_REGISTRY_FILE or " - "ASKUI_INSTALLATION_DIRECTORY environment variable must be set" - ) - raise ValueError(error_msg) - return self - - class AskUiControllerServer: """ Concrete implementation of `ControllerServer` for managing the AskUI Remote Device Controller process. Handles process discovery, startup, and shutdown for the native controller binary. + + Args: + settings (AskUiControllerSettings | None, optional): Settings for the AskUI. """ - def __init__(self) -> None: + def __init__(self, settings: AskUiControllerSettings | None = None) -> None: self._process: subprocess.Popen[bytes] | None = None - self._settings = AskUiControllerSettings() - - def _find_remote_device_controller(self) -> pathlib.Path: - if ( - self._settings.installation_directory is not None - and self._settings.component_registry_file is None - ): - logger.warning( - "Outdated AskUI Suite detected. Please update to the latest version." - ) - askui_remote_device_controller_path = ( - self._find_remote_device_controller_by_legacy_path() - ) - if not askui_remote_device_controller_path.is_file(): - error_msg = ( - "AskUIRemoteDeviceController executable does not exist under " - f"'{askui_remote_device_controller_path}'" - ) - raise FileNotFoundError(error_msg) - return askui_remote_device_controller_path - return self._find_remote_device_controller_by_component_registry() - - def _find_remote_device_controller_by_component_registry(self) -> pathlib.Path: - assert self._settings.component_registry_file is not None, ( - "Component registry file is not set" - ) - component_registry = AskUiComponentRegistry.model_validate_json( - self._settings.component_registry_file.read_text() - ) - askui_remote_device_controller_path = ( - component_registry.installed_packages.remote_device_controller_uuid.executables.askui_remote_device_controller # noqa: E501 - ) - if not askui_remote_device_controller_path.is_file(): - error_msg = ( - "AskUIRemoteDeviceController executable does not exist under " - f"'{askui_remote_device_controller_path}'" - ) - raise FileNotFoundError(error_msg) - return askui_remote_device_controller_path - - def _find_remote_device_controller_by_legacy_path(self) -> pathlib.Path: - assert self._settings.installation_directory is not None, ( - "Installation directory is not set" - ) - match sys.platform: - case "win32": - return ( - self._settings.installation_directory - / "Binaries" - / "resources" - / "assets" - / "binaries" - / "AskuiRemoteDeviceController.exe" - ) - case "darwin": - return ( - self._settings.installation_directory - / "Binaries" - / "askui-ui-controller.app" - / "Contents" - / "Resources" - / "assets" - / "binaries" - / "AskuiRemoteDeviceController" - ) - case "linux": - return ( - self._settings.installation_directory - / "Binaries" - / "resources" - / "assets" - / "binaries" - / "AskuiRemoteDeviceController" - ) - case _: - error_msg = ( - f"Platform {sys.platform} not supported by " - "AskUI Remote Device Controller" - ) - raise NotImplementedError(error_msg) + self._settings = settings or AskUiControllerSettings() def _start_process(self, path: pathlib.Path) -> None: self._process = subprocess.Popen(path) @@ -198,11 +79,11 @@ def start(self, clean_up: bool = False) -> None: and process_exists("AskuiRemoteDeviceController.exe") ): self.clean_up() - remote_device_controller_path = self._find_remote_device_controller() logger.debug( - "Starting AskUI Remote Device Controller: %s", remote_device_controller_path + "Starting AskUI Remote Device Controller: %s", + self._settings.controller_path, ) - self._start_process(remote_device_controller_path) + self._start_process(self._settings.controller_path) time.sleep(0.5) def clean_up(self) -> None: diff --git a/src/askui/tools/askui/askui_controller_settings.py b/src/askui/tools/askui/askui_controller_settings.py new file mode 100644 index 00000000..da0ee983 --- /dev/null +++ b/src/askui/tools/askui/askui_controller_settings.py @@ -0,0 +1,150 @@ +import pathlib +import sys +from functools import cached_property + +from pydantic import BaseModel, Field, model_validator +from pydantic_settings import BaseSettings, SettingsConfigDict +from typing_extensions import Self + + +class RemoteDeviceController(BaseModel): + askui_remote_device_controller: pathlib.Path = Field( + alias="AskUIRemoteDeviceController" + ) + + +class Executables(BaseModel): + executables: RemoteDeviceController = Field(alias="Executables") + + +class InstalledPackages(BaseModel): + remote_device_controller_uuid: Executables = Field( + alias="{aed1b543-e856-43ad-b1bc-19365d35c33e}" + ) + + +class AskUiComponentRegistry(BaseModel): + definition_version: int = Field(alias="DefinitionVersion") + installed_packages: InstalledPackages = Field(alias="InstalledPackages") + + +class AskUiControllerSettings(BaseSettings): + model_config = SettingsConfigDict( + env_prefix="ASKUI_", + ) + + component_registry_file: pathlib.Path | None = None + installation_directory: pathlib.Path | None = Field( + None, + deprecated="ASKUI_INSTALLATION_DIRECTORY has been deprecated in favor of " + "ASKUI_COMPONENT_REGISTRY_FILE and ASKUI_CONTROLLER_PATH. You may be using an " + "outdated AskUI Suite. If you think so, reinstall to upgrade the AskUI Suite " + "(see https://docs.askui.com/01-tutorials/00-installation).", + ) + controller_path_setting: pathlib.Path | None = Field( + None, + validation_alias="ASKUI_CONTROLLER_PATH", + description="Path to the AskUI Remote Device Controller executable. Takes " + "precedence over ASKUI_COMPONENT_REGISTRY_FILE and ASKUI_INSTALLATION_DIRECTORY" + ".", + ) + + @model_validator(mode="after") + def validate_either_component_registry_or_installation_directory_is_set( + self, + ) -> "Self": + if ( + self.component_registry_file is None + and self.installation_directory is None + and self.controller_path_setting is None + ): + error_msg = ( + "Either ASKUI_COMPONENT_REGISTRY_FILE, ASKUI_INSTALLATION_DIRECTORY, " + "or ASKUI_CONTROLLER_PATH environment variable must be set" + ) + raise ValueError(error_msg) + return self + + def _find_remote_device_controller_by_installation_directory( + self, + ) -> pathlib.Path | None: + if self.installation_directory is None: + return None + + return self._build_controller_path(self.installation_directory) + + def _build_controller_path( + self, installation_directory: pathlib.Path + ) -> pathlib.Path: + match sys.platform: + case "win32": + return ( + installation_directory + / "Binaries" + / "resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController.exe" + ) + case "darwin": + return ( + installation_directory + / "Binaries" + / "askui-ui-controller.app" + / "Contents" + / "Resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController" + ) + case "linux": + return ( + installation_directory + / "Binaries" + / "resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController" + ) + case _: + error_msg = ( + f'Platform "{sys.platform}" not supported by ' + "AskUI Remote Device Controller" + ) + raise NotImplementedError(error_msg) + + def _find_remote_device_controller_by_component_registry_file( + self, + ) -> pathlib.Path | None: + if self.component_registry_file is None: + return None + + component_registry = AskUiComponentRegistry.model_validate_json( + self.component_registry_file.read_text() + ) + return ( + component_registry.installed_packages.remote_device_controller_uuid.executables.askui_remote_device_controller # noqa: E501 + ) + + @cached_property + def controller_path(self) -> pathlib.Path: + result = ( + self.controller_path_setting + or self._find_remote_device_controller_by_component_registry_file() + or self._find_remote_device_controller_by_installation_directory() + ) + assert result is not None, ( + "No AskUI Remote Device Controller found. Please set the " + "ASKUI_COMPONENT_REGISTRY_FILE, ASKUI_INSTALLATION_DIRECTORY, or " + "ASKUI_CONTROLLER_PATH environment variable." + ) + if not result.is_file(): + error_msg = ( + "AskUIRemoteDeviceController executable does not exist under " + f"`{result}`" + ) + raise FileNotFoundError(error_msg) + return result + + +__all__ = ["AskUiControllerSettings"] diff --git a/tests/e2e/tools/askui/test_askui_controller.py b/tests/e2e/tools/askui/test_askui_controller.py index 982a553f..d9c3a097 100644 --- a/tests/e2e/tools/askui/test_askui_controller.py +++ b/tests/e2e/tools/askui/test_askui_controller.py @@ -1,6 +1,5 @@ import base64 import io -from pathlib import Path from typing import Literal import pytest @@ -31,15 +30,6 @@ def controller_client( ) -def test_find_remote_device_controller_by_component_registry( - controller_server: AskUiControllerServer, -) -> None: - remote_device_controller_path = Path( - controller_server._find_remote_device_controller_by_component_registry() - ) - assert "AskuiRemoteDeviceController" == remote_device_controller_path.stem - - def test_actions(controller_client: AskUiControllerClient) -> None: with controller_client: controller_client.screenshot() diff --git a/tests/unit/tools/__init__.py b/tests/unit/tools/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unit/tools/askui/__init__.py b/tests/unit/tools/askui/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unit/tools/askui/test_askui_controller_settings.py b/tests/unit/tools/askui/test_askui_controller_settings.py new file mode 100644 index 00000000..7b984333 --- /dev/null +++ b/tests/unit/tools/askui/test_askui_controller_settings.py @@ -0,0 +1,398 @@ +import pathlib +import sys +from unittest.mock import patch + +import pytest +from pydantic import ValidationError + +from askui.tools.askui.askui_controller_settings import AskUiControllerSettings + + +class TestAskUiControllerSettings: + """Test suite for AskUiControllerSettings.""" + + def test_controller_path_setting_takes_precedence( + self, tmp_path: pathlib.Path + ) -> None: + """Test that ASKUI_CONTROLLER_PATH takes precedence over other settings.""" + controller_path = tmp_path / "controller.exe" + controller_path.touch() + + with patch.dict( + "os.environ", {"ASKUI_CONTROLLER_PATH": str(controller_path)}, clear=True + ): + settings = AskUiControllerSettings() + assert settings.controller_path == controller_path + + def test_component_registry_file_resolution(self, tmp_path: pathlib.Path) -> None: + """Test resolution via component registry file.""" + registry_file = tmp_path / "registry.json" + controller_path = tmp_path / "controller.exe" + controller_path.touch() + + registry_content = { + "DefinitionVersion": 1, + "InstalledPackages": { + "{aed1b543-e856-43ad-b1bc-19365d35c33e}": { + "Executables": {"AskUIRemoteDeviceController": str(controller_path)} + } + }, + } + + import json + + registry_file.write_text(json.dumps(registry_content)) + + with patch.dict( + "os.environ", + {"ASKUI_COMPONENT_REGISTRY_FILE": str(registry_file)}, + clear=True, + ): + settings = AskUiControllerSettings() + assert settings.controller_path == controller_path + + def test_installation_directory_resolution(self, tmp_path: pathlib.Path) -> None: + """Test resolution via installation directory.""" + installation_dir = tmp_path / "installation" + installation_dir.mkdir() + + # Create the expected path structure based on platform + if sys.platform == "win32": + controller_path = ( + installation_dir + / "Binaries" + / "resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController.exe" + ) + elif sys.platform == "darwin": + controller_path = ( + installation_dir + / "Binaries" + / "askui-ui-controller.app" + / "Contents" + / "Resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController" + ) + else: # linux + controller_path = ( + installation_dir + / "Binaries" + / "resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController" + ) + + controller_path.parent.mkdir(parents=True, exist_ok=True) + controller_path.touch() + + with patch.dict( + "os.environ", + {"ASKUI_INSTALLATION_DIRECTORY": str(installation_dir)}, + clear=True, + ): + settings = AskUiControllerSettings() + assert settings.controller_path == controller_path + + def test_no_environment_variables_raises_error(self) -> None: + """Test that ValueError is raised when no environment variables are set.""" + with patch.dict("os.environ", {}, clear=True): + with pytest.raises( + ValueError, match="Either ASKUI_COMPONENT_REGISTRY_FILE" + ): + AskUiControllerSettings() + + def test_build_controller_path_windows(self) -> None: + """Test _build_controller_path for Windows platform.""" + with patch("sys.platform", "win32"): + with patch.dict( + "os.environ", {"ASKUI_CONTROLLER_PATH": "/tmp/test"}, clear=True + ): + settings = AskUiControllerSettings() + installation_dir = pathlib.Path("/test/installation") + expected_path = ( + installation_dir + / "Binaries" + / "resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController.exe" + ) + assert ( + settings._build_controller_path(installation_dir) == expected_path + ) + + def test_build_controller_path_darwin(self) -> None: + """Test _build_controller_path for macOS platform.""" + with patch("sys.platform", "darwin"): + with patch.dict( + "os.environ", {"ASKUI_CONTROLLER_PATH": "/tmp/test"}, clear=True + ): + settings = AskUiControllerSettings() + installation_dir = pathlib.Path("/test/installation") + expected_path = ( + installation_dir + / "Binaries" + / "askui-ui-controller.app" + / "Contents" + / "Resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController" + ) + assert ( + settings._build_controller_path(installation_dir) == expected_path + ) + + def test_build_controller_path_linux(self) -> None: + """Test _build_controller_path for Linux platform.""" + with patch("sys.platform", "linux"): + with patch.dict( + "os.environ", {"ASKUI_CONTROLLER_PATH": "/tmp/test"}, clear=True + ): + settings = AskUiControllerSettings() + installation_dir = pathlib.Path("/test/installation") + expected_path = ( + installation_dir + / "Binaries" + / "resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController" + ) + assert ( + settings._build_controller_path(installation_dir) == expected_path + ) + + def test_build_controller_path_unsupported_platform(self) -> None: + """Test _build_controller_path for unsupported platform.""" + with patch("sys.platform", "unsupported"): + with patch.dict( + "os.environ", {"ASKUI_CONTROLLER_PATH": "/tmp/test"}, clear=True + ): + settings = AskUiControllerSettings() + installation_dir = pathlib.Path("/test/installation") + with pytest.raises( + NotImplementedError, match='Platform "unsupported" not supported' + ): + settings._build_controller_path(installation_dir) + + def test_invalid_component_registry_file(self, tmp_path: pathlib.Path) -> None: + """Test handling of invalid component registry file.""" + registry_file = tmp_path / "invalid.json" + registry_file.write_text("invalid json") + + with patch.dict( + "os.environ", + {"ASKUI_COMPONENT_REGISTRY_FILE": str(registry_file)}, + clear=True, + ): + settings = AskUiControllerSettings() + # Should return None when registry file is invalid + with pytest.raises(ValidationError): + settings._find_remote_device_controller_by_component_registry_file() + + def test_missing_component_registry_file(self, tmp_path: pathlib.Path) -> None: + """Test handling of missing component registry file.""" + registry_file = tmp_path / "missing.json" + + with patch.dict( + "os.environ", + {"ASKUI_COMPONENT_REGISTRY_FILE": str(registry_file)}, + clear=True, + ): + settings = AskUiControllerSettings() + # Should raise FileNotFoundError when registry file doesn't exist + with pytest.raises(FileNotFoundError): + settings._find_remote_device_controller_by_component_registry_file() + + def test_controller_executable_not_found(self, tmp_path: pathlib.Path) -> None: + """Test error when controller executable doesn't exist.""" + controller_path = tmp_path / "nonexistent.exe" + + with patch.dict( + "os.environ", {"ASKUI_CONTROLLER_PATH": str(controller_path)}, clear=True + ): + settings = AskUiControllerSettings() + with pytest.raises( + FileNotFoundError, + match="AskUIRemoteDeviceController executable does not exist", + ): + _ = settings.controller_path + + def test_controller_path_cached_property(self, tmp_path: pathlib.Path) -> None: + """Test that controller_path is cached.""" + controller_path = tmp_path / "controller.exe" + controller_path.touch() + + with patch.dict( + "os.environ", {"ASKUI_CONTROLLER_PATH": str(controller_path)}, clear=True + ): + settings = AskUiControllerSettings() + + # First call should resolve the path + first_result = settings.controller_path + assert first_result == controller_path + + # Second call should use cached result + second_result = settings.controller_path + assert second_result == controller_path + assert first_result is second_result + + def test_priority_order_controller_path_first(self, tmp_path: pathlib.Path) -> None: + """Test that controller_path_setting takes priority over other methods.""" + controller_path = tmp_path / "controller.exe" + controller_path.touch() + + registry_file = tmp_path / "registry.json" + registry_file.write_text('{"DefinitionVersion": 1, "InstalledPackages": {}}') + + installation_dir = tmp_path / "installation" + installation_dir.mkdir() + + with patch.dict( + "os.environ", + { + "ASKUI_CONTROLLER_PATH": str(controller_path), + "ASKUI_COMPONENT_REGISTRY_FILE": str(registry_file), + "ASKUI_INSTALLATION_DIRECTORY": str(installation_dir), + }, + clear=True, + ): + settings = AskUiControllerSettings() + assert settings.controller_path == controller_path + + def test_priority_order_component_registry_second( + self, tmp_path: pathlib.Path + ) -> None: + """Test that component registry takes priority over installation directory.""" + registry_file = tmp_path / "registry.json" + controller_path = tmp_path / "controller.exe" + controller_path.touch() + + registry_content = { + "DefinitionVersion": 1, + "InstalledPackages": { + "{aed1b543-e856-43ad-b1bc-19365d35c33e}": { + "Executables": {"AskUIRemoteDeviceController": str(controller_path)} + } + }, + } + + import json + + registry_file.write_text(json.dumps(registry_content)) + + installation_dir = tmp_path / "installation" + installation_dir.mkdir() + + with patch.dict( + "os.environ", + { + "ASKUI_COMPONENT_REGISTRY_FILE": str(registry_file), + "ASKUI_INSTALLATION_DIRECTORY": str(installation_dir), + }, + clear=True, + ): + settings = AskUiControllerSettings() + assert settings.controller_path == controller_path + + def test_installation_directory_fallback(self, tmp_path: pathlib.Path) -> None: + """Test that installation directory is used as fallback.""" + installation_dir = tmp_path / "installation" + installation_dir.mkdir() + + # Create the expected path structure + if sys.platform == "win32": + controller_path = ( + installation_dir + / "Binaries" + / "resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController.exe" + ) + elif sys.platform == "darwin": + controller_path = ( + installation_dir + / "Binaries" + / "askui-ui-controller.app" + / "Contents" + / "Resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController" + ) + else: # linux + controller_path = ( + installation_dir + / "Binaries" + / "resources" + / "assets" + / "binaries" + / "AskuiRemoteDeviceController" + ) + + controller_path.parent.mkdir(parents=True, exist_ok=True) + controller_path.touch() + + with patch.dict( + "os.environ", + {"ASKUI_INSTALLATION_DIRECTORY": str(installation_dir)}, + clear=True, + ): + settings = AskUiControllerSettings() + assert settings.controller_path == controller_path + + def test_none_values_return_none(self) -> None: + """Test that None values are handled correctly.""" + # Create a settings instance with mocked environment + with patch.dict( + "os.environ", {"ASKUI_CONTROLLER_PATH": "/tmp/test"}, clear=True + ): + settings = AskUiControllerSettings() + + # Mock the methods to return None + with ( + patch.object(settings, "component_registry_file", None), + patch.object(settings, "installation_directory", None), + ): + assert ( + settings._find_remote_device_controller_by_installation_directory() + is None + ) + # Note: _find_remote_device_controller_by_component_registry_file will + # raise + # FileNotFoundError when component_registry_file is None, so we test + # that separately + + def test_assertion_error_when_no_controller_found(self) -> None: + """Test that assertion error is raised when no controller is found.""" + # Create a settings instance with mocked environment + with patch.dict( + "os.environ", {"ASKUI_CONTROLLER_PATH": "/tmp/test"}, clear=True + ): + settings = AskUiControllerSettings() + + # Mock all resolution methods to return None + with ( + patch.object( + settings, + "_find_remote_device_controller_by_component_registry_file", + return_value=None, + ), + patch.object( + settings, + "_find_remote_device_controller_by_installation_directory", + return_value=None, + ), + patch.object(settings, "controller_path_setting", None), + ): + with pytest.raises( + AssertionError, match="No AskUI Remote Device Controller found" + ): + _ = settings.controller_path