From 3614ba8f8f6f1dda65a6abdfaf989ff35d2576e7 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Wed, 11 Feb 2026 03:57:25 -0700 Subject: [PATCH] Replace dict returns with typed Pydantic response models across all 65 tools Every tool now returns a structured BaseModel instead of dict[str, Any], giving callers attribute access, IDE autocomplete, and schema validation. Adds ~30 model classes to models.py and updates all test assertions. --- src/mixins/apps.py | 252 ++++++++++++----------- src/mixins/connectivity.py | 118 ++++++----- src/mixins/devices.py | 216 ++++++++++++-------- src/mixins/files.py | 156 ++++++++------- src/mixins/input.py | 206 ++++++++++--------- src/mixins/screenshot.py | 165 +++++++-------- src/mixins/settings.py | 328 +++++++++++++++--------------- src/mixins/ui.py | 157 ++++++++------- src/models.py | 400 ++++++++++++++++++++++++++++++++++++- src/server.py | 37 ++-- tests/test_apps.py | 68 +++---- tests/test_connectivity.py | 56 +++--- tests/test_devices.py | 55 ++--- tests/test_files.py | 50 ++--- tests/test_input.py | 96 ++++----- tests/test_screenshot.py | 42 ++-- tests/test_server.py | 20 +- tests/test_settings.py | 148 +++++++------- tests/test_ui.py | 42 ++-- 19 files changed, 1558 insertions(+), 1054 deletions(-) diff --git a/src/mixins/apps.py b/src/mixins/apps.py index e9a2717..65061d4 100644 --- a/src/mixins/apps.py +++ b/src/mixins/apps.py @@ -4,12 +4,17 @@ Provides tools for app management and launching. """ import re -from typing import Any from fastmcp import Context from fastmcp.contrib.mcp_mixin import mcp_resource, mcp_tool from ..config import is_developer_mode +from ..models import ( + AppActionResult, + AppCurrentResult, + IntentResult, + PackageListResult, +) from .base import ADBBaseMixin # Common Android intent flags (hex values for am start -f) @@ -40,7 +45,7 @@ class AppsMixin(ADBBaseMixin): self, package_name: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> AppActionResult: """Launch an app by package name. Starts the main activity of the specified application. @@ -70,19 +75,19 @@ class AppsMixin(ADBBaseMixin): ], device_id, ) - return { - "success": result.success, - "action": "launch", - "package": package_name, - "error": result.stderr if not result.success else None, - } + return AppActionResult( + success=result.success, + action="launch", + package=package_name, + error=result.stderr if not result.success else None, + ) @mcp_tool() async def app_open_url( self, url: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> AppActionResult: """Open a URL in the default browser. Launches the default browser and navigates to the URL. @@ -106,19 +111,19 @@ class AppsMixin(ADBBaseMixin): ], device_id, ) - return { - "success": result.success, - "action": "open_url", - "url": url, - "error": result.stderr if not result.success else None, - } + return AppActionResult( + success=result.success, + action="open_url", + url=url, + error=result.stderr if not result.success else None, + ) @mcp_tool() async def app_close( self, package_name: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> AppActionResult: """Force stop an app. Stops the application and all its background services. @@ -133,18 +138,18 @@ class AppsMixin(ADBBaseMixin): result = await self.run_shell_args( ["am", "force-stop", package_name], device_id ) - return { - "success": result.success, - "action": "close", - "package": package_name, - "error": result.stderr if not result.success else None, - } + return AppActionResult( + success=result.success, + action="close", + package=package_name, + error=result.stderr if not result.success else None, + ) @mcp_tool() async def app_current( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> AppCurrentResult: """Get the currently focused app. Returns the package name of the app currently in foreground. @@ -172,17 +177,17 @@ class AppsMixin(ADBBaseMixin): package = match.group(1) activity = match.group(2) break - return { - "success": True, - "package": package, - "activity": activity, - "raw": result.stdout[:500] if not package else None, - } + return AppCurrentResult( + success=True, + package=package, + activity=activity, + raw=result.stdout[:500] if not package else None, + ) - return { - "success": False, - "error": result.stderr, - } + return AppCurrentResult( + success=False, + error=result.stderr, + ) # === Developer Mode Tools === @@ -196,7 +201,7 @@ class AppsMixin(ADBBaseMixin): system_only: bool = False, third_party_only: bool = False, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> PackageListResult: """List installed packages. [DEVELOPER MODE] Retrieves all installed application packages. @@ -211,10 +216,10 @@ class AppsMixin(ADBBaseMixin): List of package names """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return PackageListResult( + success=False, + error="Developer mode required", + ) cmd = ["pm", "list", "packages"] if system_only: @@ -232,16 +237,16 @@ class AppsMixin(ADBBaseMixin): if filter_text is None or filter_text.lower() in pkg.lower(): packages.append(pkg) - return { - "success": True, - "packages": sorted(packages), - "count": len(packages), - } + return PackageListResult( + success=True, + packages=sorted(packages), + count=len(packages), + ) - return { - "success": False, - "error": result.stderr, - } + return PackageListResult( + success=False, + error=result.stderr, + ) @mcp_tool( tags={"developer"}, @@ -251,7 +256,7 @@ class AppsMixin(ADBBaseMixin): self, apk_path: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> AppActionResult: """Install an APK file. [DEVELOPER MODE] Installs an APK from the host machine to the device. @@ -264,19 +269,20 @@ class AppsMixin(ADBBaseMixin): Installation result """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return AppActionResult( + success=False, + action="install", + error="Developer mode required", + ) result = await self.run_adb(["install", "-r", apk_path], device_id) - return { - "success": result.success, - "action": "install", - "apk": apk_path, - "output": result.stdout, - "error": result.stderr if not result.success else None, - } + return AppActionResult( + success=result.success, + action="install", + apk=apk_path, + output=result.stdout, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -288,7 +294,7 @@ class AppsMixin(ADBBaseMixin): package_name: str, keep_data: bool = False, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> AppActionResult: """Uninstall an app. [DEVELOPER MODE] Removes an application from the device. @@ -304,10 +310,11 @@ class AppsMixin(ADBBaseMixin): Uninstall result """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return AppActionResult( + success=False, + action="uninstall", + error="Developer mode required", + ) # Elicit confirmation await ctx.warning(f"Uninstall requested: {package_name}") @@ -320,11 +327,12 @@ class AppsMixin(ADBBaseMixin): if confirmation.action != "accept" or confirmation.content == "Cancel": await ctx.info("Uninstall cancelled by user") - return { - "success": False, - "cancelled": True, - "message": "Uninstall cancelled by user", - } + return AppActionResult( + success=False, + action="uninstall", + cancelled=True, + message="Uninstall cancelled by user", + ) await ctx.info(f"Uninstalling {package_name}...") @@ -340,13 +348,13 @@ class AppsMixin(ADBBaseMixin): else: await ctx.error(f"Uninstall failed: {result.stderr}") - return { - "success": result.success, - "action": "uninstall", - "package": package_name, - "kept_data": keep_data, - "error": result.stderr if not result.success else None, - } + return AppActionResult( + success=result.success, + action="uninstall", + package=package_name, + kept_data=keep_data, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -357,7 +365,7 @@ class AppsMixin(ADBBaseMixin): ctx: Context, package_name: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> AppActionResult: """Clear app data and cache. [DEVELOPER MODE] Clears all data for an application (like a fresh @@ -372,10 +380,11 @@ class AppsMixin(ADBBaseMixin): Clear result """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return AppActionResult( + success=False, + action="clear_data", + error="Developer mode required", + ) # Elicit confirmation await ctx.warning(f"Clear data requested: {package_name}") @@ -390,11 +399,12 @@ class AppsMixin(ADBBaseMixin): if confirmation.action != "accept" or confirmation.content == "Cancel": await ctx.info("Clear data cancelled by user") - return { - "success": False, - "cancelled": True, - "message": "Clear data cancelled by user", - } + return AppActionResult( + success=False, + action="clear_data", + cancelled=True, + message="Clear data cancelled by user", + ) await ctx.info(f"Clearing data for {package_name}...") @@ -405,12 +415,12 @@ class AppsMixin(ADBBaseMixin): else: await ctx.error(f"Clear data failed: {result.stderr}") - return { - "success": result.success, - "action": "clear_data", - "package": package_name, - "error": result.stderr if not result.success else None, - } + return AppActionResult( + success=result.success, + action="clear_data", + package=package_name, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -424,7 +434,7 @@ class AppsMixin(ADBBaseMixin): extras: dict[str, str] | None = None, flags: list[str] | None = None, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> IntentResult: """Start a specific activity with intent. [DEVELOPER MODE] Launch an activity with full intent control. @@ -450,10 +460,11 @@ class AppsMixin(ADBBaseMixin): data_uri="myapp://product/123" """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return IntentResult( + success=False, + action="activity_start", + error="Developer mode required", + ) cmd_args = ["am", "start"] @@ -490,15 +501,15 @@ class AppsMixin(ADBBaseMixin): result = await self.run_shell_args(cmd_args, device_id) - return { - "success": result.success, - "action": "activity_start", - "component": component, - "intent_action": action, - "data_uri": data_uri, - "output": result.stdout, - "error": result.stderr if not result.success else None, - } + return IntentResult( + success=result.success, + action="activity_start", + component=component, + intent_action=action, + data_uri=data_uri, + output=result.stdout, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -510,7 +521,7 @@ class AppsMixin(ADBBaseMixin): extras: dict[str, str] | None = None, package: str | None = None, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> IntentResult: """Send a broadcast intent. [DEVELOPER MODE] Sends a broadcast that can be received by @@ -531,10 +542,11 @@ class AppsMixin(ADBBaseMixin): - android.net.conn.CONNECTIVITY_CHANGE """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return IntentResult( + success=False, + action="broadcast_send", + error="Developer mode required", + ) cmd_args = ["am", "broadcast", "-a", action] @@ -552,18 +564,18 @@ class AppsMixin(ADBBaseMixin): result = await self.run_shell_args(cmd_args, device_id) - return { - "success": result.success, - "action": "broadcast_send", - "broadcast_action": action, - "package": package, - "output": result.stdout, - "error": result.stderr if not result.success else None, - } + return IntentResult( + success=result.success, + action="broadcast_send", + broadcast_action=action, + package=package, + output=result.stdout, + error=result.stderr if not result.success else None, + ) # === Resources === @mcp_resource(uri="adb://apps/current") - async def resource_current_app(self) -> dict[str, Any]: + async def resource_current_app(self) -> AppCurrentResult: """Resource: Get currently focused app.""" return await self.app_current() diff --git a/src/mixins/connectivity.py b/src/mixins/connectivity.py index 40fa912..c14c1ae 100644 --- a/src/mixins/connectivity.py +++ b/src/mixins/connectivity.py @@ -4,12 +4,12 @@ Provides tools for managing ADB network connections and device properties. """ import re -from typing import Any from fastmcp import Context from fastmcp.contrib.mcp_mixin import mcp_tool from ..config import is_developer_mode +from ..models import ConnectResult, DevicePropertiesResult, TcpipResult from .base import ADBBaseMixin @@ -28,7 +28,7 @@ class ConnectivityMixin(ADBBaseMixin): self, host: str, port: int = 5555, - ) -> dict[str, Any]: + ) -> ConnectResult: """Connect to a device over TCP/IP. Establishes an ADB connection to a device on the network. @@ -49,20 +49,20 @@ class ConnectivityMixin(ADBBaseMixin): connected = result.success and "connected" in result.stdout.lower() already = "already connected" in result.stdout.lower() - return { - "success": connected, - "already_connected": already, - "address": target, - "output": result.stdout, - "error": result.stderr if not connected else None, - } + return ConnectResult( + success=connected, + already_connected=already, + address=target, + output=result.stdout, + error=result.stderr if not connected else None, + ) @mcp_tool() async def adb_disconnect( self, host: str, port: int = 5555, - ) -> dict[str, Any]: + ) -> ConnectResult: """Disconnect a network-connected device. Drops the ADB TCP/IP connection to the specified device. @@ -79,12 +79,12 @@ class ConnectivityMixin(ADBBaseMixin): disconnected = result.success and "disconnected" in result.stdout.lower() - return { - "success": disconnected, - "address": target, - "output": result.stdout, - "error": result.stderr if not disconnected else None, - } + return ConnectResult( + success=disconnected, + address=target, + output=result.stdout, + error=result.stderr if not disconnected else None, + ) @mcp_tool( tags={"developer"}, @@ -95,7 +95,7 @@ class ConnectivityMixin(ADBBaseMixin): ctx: Context, port: int = 5555, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> TcpipResult: """Switch a USB-connected device to TCP/IP mode. [DEVELOPER MODE] Restarts ADB on the device in TCP/IP mode, @@ -115,21 +115,21 @@ class ConnectivityMixin(ADBBaseMixin): Result with device IP address for subsequent adb_connect """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return TcpipResult( + success=False, + error="Developer mode required", + ) # Reject if device_id looks like a network device (IP:port format) target = device_id or self.get_current_device() if target and re.match(r"\d+\.\d+\.\d+\.\d+:\d+", target): - return { - "success": False, - "error": ( + return TcpipResult( + success=False, + error=( f"Device '{target}' is already a network device. " "adb_tcpip only works on USB-connected devices." ), - } + ) # Get device IP before switching (wlan0) ip_result = await self.run_shell_args( @@ -142,13 +142,13 @@ class ConnectivityMixin(ADBBaseMixin): device_ip = match.group(1) if not device_ip: - return { - "success": False, - "error": ( + return TcpipResult( + success=False, + error=( "Could not determine device IP address. " "Ensure the device is connected to WiFi." ), - } + ) await ctx.info(f"Switching device to TCP/IP mode on port {port}...") @@ -156,26 +156,26 @@ class ConnectivityMixin(ADBBaseMixin): result = await self.run_adb(["tcpip", str(port)], device_id) if not result.success: - return { - "success": False, - "error": result.stderr or result.stdout, - } + return TcpipResult( + success=False, + error=result.stderr or result.stdout, + ) await ctx.info( f"Device switched to TCP/IP on port {port}. " f"Connect with: adb_connect('{device_ip}', {port})" ) - return { - "success": True, - "port": port, - "device_ip": device_ip, - "connect_address": f"{device_ip}:{port}", - "message": ( + return TcpipResult( + success=True, + port=port, + device_ip=device_ip, + connect_address=f"{device_ip}:{port}", + message=( f"Device now listening on {device_ip}:{port}. " "USB connection will drop. Use adb_connect() to reconnect." ), - } + ) @mcp_tool() async def adb_pair( @@ -183,7 +183,7 @@ class ConnectivityMixin(ADBBaseMixin): host: str, port: int, pairing_code: str, - ) -> dict[str, Any]: + ) -> ConnectResult: """Pair with a device for wireless debugging (Android 11+). Pairs with a device using the wireless debugging pairing code @@ -205,18 +205,18 @@ class ConnectivityMixin(ADBBaseMixin): paired = result.success and "successfully paired" in result.stdout.lower() - return { - "success": paired, - "address": target, - "output": result.stdout, - "error": result.stderr if not paired else None, - } + return ConnectResult( + success=paired, + address=target, + output=result.stdout, + error=result.stderr if not paired else None, + ) @mcp_tool() async def device_properties( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> DevicePropertiesResult: """Get detailed device properties via getprop. Fetches a comprehensive batch of system properties including @@ -226,7 +226,7 @@ class ConnectivityMixin(ADBBaseMixin): device_id: Target device Returns: - Dictionary of device properties grouped by category + Device properties grouped by category """ props_to_fetch = { "identity": [ @@ -257,7 +257,7 @@ class ConnectivityMixin(ADBBaseMixin): ], } - result: dict[str, Any] = {"success": True} + categories: dict[str, dict[str, str]] = {} for category, prop_list in props_to_fetch.items(): category_data: dict[str, str] = {} @@ -266,11 +266,19 @@ class ConnectivityMixin(ADBBaseMixin): if value: category_data[friendly_name] = value if category_data: - result[category] = category_data + categories[category] = category_data # Check if we got anything at all - if len(result) == 1: # only "success" key - result["success"] = False - result["error"] = "No properties returned. Is the device connected?" + if not categories: + return DevicePropertiesResult( + success=False, + error="No properties returned. Is the device connected?", + ) - return result + return DevicePropertiesResult( + success=True, + identity=categories.get("identity"), + software=categories.get("software"), + hardware=categories.get("hardware"), + system=categories.get("system"), + ) diff --git a/src/mixins/devices.py b/src/mixins/devices.py index 21c4464..f78baa5 100644 --- a/src/mixins/devices.py +++ b/src/mixins/devices.py @@ -11,7 +11,13 @@ from fastmcp import Context from fastmcp.contrib.mcp_mixin import mcp_resource, mcp_tool from ..config import is_developer_mode -from ..models import DeviceInfo +from ..models import ( + DeviceInfo, + DeviceInfoResult, + DeviceSelectResult, + LogcatResult, + RebootResult, +) from .base import ADBBaseMixin @@ -84,7 +90,7 @@ class DevicesMixin(ADBBaseMixin): return await self._refresh_devices() @mcp_tool() - async def devices_use(self, device_id: str) -> dict[str, Any]: + async def devices_use(self, device_id: str) -> DeviceSelectResult: """Set the current working device. All subsequent commands will target this device by default. @@ -101,28 +107,28 @@ class DevicesMixin(ADBBaseMixin): device = next((d for d in devices if d.device_id == device_id), None) if not device: - return { - "success": False, - "error": f"Device {device_id} not found", - "available": [d.device_id for d in devices], - } + return DeviceSelectResult( + success=False, + error=f"Device {device_id} not found", + available=[d.device_id for d in devices], + ) if device.status != "device": - return { - "success": False, - "error": f"Device {device_id} is {device.status}, not ready", - } + return DeviceSelectResult( + success=False, + error=f"Device {device_id} is {device.status}, not ready", + ) self.set_current_device(device_id) - return { - "success": True, - "message": f"Now using device {device_id}", - "device": device.model_dump(), - } + return DeviceSelectResult( + success=True, + message=f"Now using device {device_id}", + device=device.model_dump(), + ) @mcp_tool() - async def devices_current(self) -> dict[str, Any]: + async def devices_current(self) -> DeviceSelectResult: """Get information about the current working device. Returns: @@ -134,22 +140,31 @@ class DevicesMixin(ADBBaseMixin): devices = await self._refresh_devices() if len(devices) == 1: # Auto-select if only one device - return { - "device": None, - "message": "No device set, but only one available", - "available": devices[0].model_dump(), - } - return { - "device": None, - "error": "No current device set. Use devices_use() first.", - "available": [d.device_id for d in devices], - } + return DeviceSelectResult( + success=True, + device=None, + message="No device set, but only one available", + available=devices[0].model_dump(), + ) + return DeviceSelectResult( + success=False, + device=None, + error="No current device set. Use devices_use() first.", + available=[d.device_id for d in devices], + ) device = self._devices_cache.get(current) if device: - return {"device": device.model_dump()} + return DeviceSelectResult( + success=True, + device=device.model_dump(), + ) - return {"device": current, "cached_info": None} + return DeviceSelectResult( + success=True, + device=current, + cached_info=None, + ) @mcp_resource(uri="adb://devices") async def resource_devices_list(self) -> dict[str, Any]: @@ -203,7 +218,7 @@ class DevicesMixin(ADBBaseMixin): async def device_info( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> DeviceInfoResult: """Get comprehensive device information. Returns device state including battery, wifi, storage, and system info. @@ -215,17 +230,13 @@ class DevicesMixin(ADBBaseMixin): Returns: Device information including battery, wifi, storage, etc. """ - info: dict[str, Any] = {} - # Battery info — also serves as connectivity check battery = await self.run_shell_args(["dumpsys", "battery"], device_id) if not battery.success: - return { - "success": False, - "error": battery.stderr or "No device connected", - } - - info["success"] = True + return DeviceInfoResult( + success=False, + error=battery.stderr or "No device connected", + ) battery_info: dict[str, Any] = {} for line in battery.stdout.split("\n"): @@ -251,7 +262,16 @@ class DevicesMixin(ADBBaseMixin): } plugged = line.split(":")[1].strip() battery_info["plugged"] = plugged_map.get(plugged, plugged) - info["battery"] = battery_info + + # Collect fields progressively + ip_address: str | None = None + wifi_ssid: str | None = None + model: str | None = None + manufacturer: str | None = None + device_name: str | None = None + android_version: str | None = None + sdk_version: str | None = None + storage: dict[str, int] | None = None # Get IP address — parse ip addr output in Python (no pipes) ip_result = await self.run_shell_args( @@ -260,7 +280,7 @@ class DevicesMixin(ADBBaseMixin): if ip_result.success: inet_match = re.search(r"inet (\d+\.\d+\.\d+\.\d+)/", ip_result.stdout) if inet_match: - info["ip_address"] = inet_match.group(1) + ip_address = inet_match.group(1) # WiFi connection info — parse dumpsys in Python (no pipes) wifi = await self.run_shell_args(["dumpsys", "wifi"], device_id) @@ -269,39 +289,57 @@ class DevicesMixin(ADBBaseMixin): if "mWifiInfo" in wifi_line and "SSID:" in wifi_line: try: ssid_part = wifi_line.split("SSID:")[1].split(",")[0].strip() - info["wifi_ssid"] = ssid_part.strip('"') + wifi_ssid = ssid_part.strip('"') except IndexError: pass break # System properties props_to_fetch = [ - ("android_version", "ro.build.version.release"), - ("sdk_version", "ro.build.version.sdk"), ("model", "ro.product.model"), ("manufacturer", "ro.product.manufacturer"), ("device_name", "ro.product.device"), + ("android_version", "ro.build.version.release"), + ("sdk_version", "ro.build.version.sdk"), ] + prop_values: dict[str, str] = {} for key, prop in props_to_fetch: value = await self.get_device_property(prop, device_id) if value: - info[key] = value + prop_values[key] = value + + model = prop_values.get("model") + manufacturer = prop_values.get("manufacturer") + device_name = prop_values.get("device_name") + android_version = prop_values.get("android_version") + sdk_version = prop_values.get("sdk_version") # Storage info — parse df output in Python (no pipes) - storage = await self.run_shell_args(["df", "/data"], device_id) - if storage.success: - lines = storage.stdout.strip().split("\n") + storage_result = await self.run_shell_args(["df", "/data"], device_id) + if storage_result.success: + lines = storage_result.stdout.strip().split("\n") if len(lines) >= 2: parts = lines[-1].split() if len(parts) >= 4: with contextlib.suppress(ValueError): - info["storage"] = { + storage = { "total_kb": int(parts[1]), "used_kb": int(parts[2]), "available_kb": int(parts[3]), } - return info + return DeviceInfoResult( + success=True, + battery=battery_info or None, + ip_address=ip_address, + wifi_ssid=wifi_ssid, + model=model, + manufacturer=manufacturer, + device_name=device_name, + android_version=android_version, + sdk_version=sdk_version, + storage=storage, + ) @mcp_tool( tags={"developer"}, @@ -312,7 +350,7 @@ class DevicesMixin(ADBBaseMixin): ctx: Context, mode: str | None = None, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> RebootResult: """Reboot the device. [DEVELOPER MODE] Reboots the Android device. @@ -330,10 +368,12 @@ class DevicesMixin(ADBBaseMixin): Reboot command result """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return RebootResult( + success=False, + action="reboot", + mode=mode or "normal", + error="Developer mode required", + ) # Elicit confirmation for this dangerous action mode_desc = mode or "normal" @@ -348,11 +388,13 @@ class DevicesMixin(ADBBaseMixin): if confirmation.action != "accept" or confirmation.content == "Cancel": await ctx.info("Reboot cancelled by user") - return { - "success": False, - "cancelled": True, - "message": "Reboot cancelled by user", - } + return RebootResult( + success=False, + action="reboot", + mode=mode_desc, + cancelled=True, + message="Reboot cancelled by user", + ) await ctx.info(f"Initiating {mode_desc} reboot...") @@ -367,12 +409,12 @@ class DevicesMixin(ADBBaseMixin): else: await ctx.error(f"Reboot failed: {result.stderr}") - return { - "success": result.success, - "action": "reboot", - "mode": mode_desc, - "error": result.stderr if not result.success else None, - } + return RebootResult( + success=result.success, + action="reboot", + mode=mode_desc, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -384,7 +426,7 @@ class DevicesMixin(ADBBaseMixin): filter_spec: str | None = None, clear_first: bool = False, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> LogcatResult: """Capture logcat output. [DEVELOPER MODE] Retrieves Android system logs. @@ -400,10 +442,10 @@ class DevicesMixin(ADBBaseMixin): Logcat output """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return LogcatResult( + success=False, + error="Developer mode required", + ) # Clear first if requested if clear_first: @@ -418,13 +460,13 @@ class DevicesMixin(ADBBaseMixin): result = await self.run_shell_args(cmd, device_id) - return { - "success": result.success, - "lines_requested": lines, - "filter": filter_spec, - "output": result.stdout, - "error": result.stderr if not result.success else None, - } + return LogcatResult( + success=result.success, + lines_requested=lines, + filter=filter_spec, + output=result.stdout, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -433,7 +475,7 @@ class DevicesMixin(ADBBaseMixin): async def logcat_clear( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> LogcatResult: """Clear the logcat buffer. [DEVELOPER MODE] Clears all logs from the device log buffer. @@ -446,15 +488,15 @@ class DevicesMixin(ADBBaseMixin): Success status """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return LogcatResult( + success=False, + error="Developer mode required", + ) result = await self.run_shell_args(["logcat", "-c"], device_id) - return { - "success": result.success, - "action": "logcat_clear", - "error": result.stderr if not result.success else None, - } + return LogcatResult( + success=result.success, + action="logcat_clear", + error=result.stderr if not result.success else None, + ) diff --git a/src/mixins/files.py b/src/mixins/files.py index fb4e530..c93dee3 100644 --- a/src/mixins/files.py +++ b/src/mixins/files.py @@ -4,12 +4,17 @@ Provides tools for file transfer between host and device. """ from pathlib import Path -from typing import Any from fastmcp import Context from fastmcp.contrib.mcp_mixin import mcp_tool from ..config import is_developer_mode +from ..models import ( + FileDeleteResult, + FileExistsResult, + FileListResult, + FileTransferResult, +) from .base import ADBBaseMixin @@ -32,7 +37,7 @@ class FilesMixin(ADBBaseMixin): local_path: str, device_path: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> FileTransferResult: """Push a file from host to device. [DEVELOPER MODE] Transfers a file from the local machine to the @@ -53,18 +58,20 @@ class FilesMixin(ADBBaseMixin): Transfer result with bytes transferred """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return FileTransferResult( + success=False, + error="Developer mode required", + action="push", + ) # Verify local file exists local = Path(local_path) if not local.exists(): - return { - "success": False, - "error": f"Local file not found: {local_path}", - } + return FileTransferResult( + success=False, + error=f"Local file not found: {local_path}", + action="push", + ) file_size = local.stat().st_size await ctx.info(f"Pushing {local.name} ({file_size:,} bytes) to {device_path}") @@ -78,14 +85,14 @@ class FilesMixin(ADBBaseMixin): else: await ctx.error(f"Push failed: {result.stderr}") - return { - "success": result.success, - "action": "push", - "local_path": str(local.absolute()), - "device_path": device_path, - "output": result.stdout, - "error": result.stderr if not result.success else None, - } + return FileTransferResult( + success=result.success, + action="push", + local_path=str(local.absolute()), + device_path=device_path, + output=result.stdout, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -97,7 +104,7 @@ class FilesMixin(ADBBaseMixin): device_path: str, local_path: str | None = None, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> FileTransferResult: """Pull a file from device to host. [DEVELOPER MODE] Transfers a file from the Android device to the @@ -118,10 +125,11 @@ class FilesMixin(ADBBaseMixin): Transfer result with local file path """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return FileTransferResult( + success=False, + error="Developer mode required", + action="pull", + ) # Default local path to current directory with same filename if not local_path: @@ -139,14 +147,14 @@ class FilesMixin(ADBBaseMixin): else: await ctx.error(f"Pull failed: {result.stderr}") - return { - "success": result.success, - "action": "pull", - "device_path": device_path, - "local_path": str(local), - "output": result.stdout, - "error": result.stderr if not result.success else None, - } + return FileTransferResult( + success=result.success, + action="pull", + device_path=device_path, + local_path=str(local), + output=result.stdout, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -156,7 +164,7 @@ class FilesMixin(ADBBaseMixin): self, device_path: str = "/sdcard/", device_id: str | None = None, - ) -> dict[str, Any]: + ) -> FileListResult: """List files in a directory on the device. [DEVELOPER MODE] Lists files and directories at the specified path. @@ -169,18 +177,18 @@ class FilesMixin(ADBBaseMixin): List of files and directories """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return FileListResult( + success=False, + error="Developer mode required", + ) result = await self.run_shell_args(["ls", "-la", device_path], device_id) if not result.success: - return { - "success": False, - "error": result.stderr or "Failed to list directory", - } + return FileListResult( + success=False, + error=result.stderr or "Failed to list directory", + ) # Parse ls output — Android uses ISO dates (YYYY-MM-DD HH:MM) # while traditional ls uses (Mon DD HH:MM), so date takes 2 or 3 fields @@ -217,12 +225,12 @@ class FilesMixin(ADBBaseMixin): } ) - return { - "success": True, - "path": device_path, - "files": files, - "count": len(files), - } + return FileListResult( + success=True, + path=device_path, + files=files, + count=len(files), + ) @mcp_tool( tags={"developer"}, @@ -233,7 +241,7 @@ class FilesMixin(ADBBaseMixin): ctx: Context, device_path: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> FileDeleteResult: """Delete a file on the device. [DEVELOPER MODE] Removes a file from the device storage. @@ -248,10 +256,11 @@ class FilesMixin(ADBBaseMixin): Deletion result """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return FileDeleteResult( + success=False, + error="Developer mode required", + action="delete", + ) # Elicit confirmation await ctx.warning(f"Delete requested: {device_path}") @@ -263,11 +272,12 @@ class FilesMixin(ADBBaseMixin): if confirmation.action != "accept" or confirmation.content == "Cancel": await ctx.info("Delete cancelled by user") - return { - "success": False, - "cancelled": True, - "message": "Delete cancelled by user", - } + return FileDeleteResult( + success=False, + action="delete", + cancelled=True, + message="Delete cancelled by user", + ) await ctx.info(f"Deleting {device_path}...") @@ -278,12 +288,12 @@ class FilesMixin(ADBBaseMixin): else: await ctx.error(f"Delete failed: {result.stderr}") - return { - "success": result.success, - "action": "delete", - "path": device_path, - "error": result.stderr if not result.success else None, - } + return FileDeleteResult( + success=result.success, + action="delete", + path=device_path, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -293,7 +303,7 @@ class FilesMixin(ADBBaseMixin): self, device_path: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> FileExistsResult: """Check if a file exists on the device. [DEVELOPER MODE] Tests for file existence. @@ -306,16 +316,18 @@ class FilesMixin(ADBBaseMixin): Existence check result """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return FileExistsResult( + success=False, + error="Developer mode required", + path=device_path, + exists=False, + ) # Use test -e and check returncode (injection-safe via run_shell_args) result = await self.run_shell_args(["test", "-e", device_path], device_id) - return { - "success": True, - "path": device_path, - "exists": result.success, - } + return FileExistsResult( + success=True, + path=device_path, + exists=result.success, + ) diff --git a/src/mixins/input.py b/src/mixins/input.py index b56a7b4..8b7946c 100644 --- a/src/mixins/input.py +++ b/src/mixins/input.py @@ -4,11 +4,16 @@ Provides tools for simulating user input on Android devices. """ import re -from typing import Any from fastmcp.contrib.mcp_mixin import mcp_tool from ..config import is_developer_mode +from ..models import ( + ClipboardSetResult, + InputResult, + ShellResult, + SwipeResult, +) from .base import ADBBaseMixin # Characters that ADB's input text command cannot handle — suggest clipboard @@ -45,7 +50,7 @@ class InputMixin(ADBBaseMixin): x: int, y: int, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> InputResult: """Tap at screen coordinates. Simulates a finger tap at the specified position. @@ -59,12 +64,12 @@ class InputMixin(ADBBaseMixin): Success status """ result = await self.run_shell_args(["input", "tap", str(x), str(y)], device_id) - return { - "success": result.success, - "action": "tap", - "coordinates": {"x": x, "y": y}, - "error": result.stderr if not result.success else None, - } + return InputResult( + success=result.success, + action="tap", + coordinates={"x": x, "y": y}, + error=result.stderr if not result.success else None, + ) @mcp_tool() async def input_swipe( @@ -75,7 +80,7 @@ class InputMixin(ADBBaseMixin): y2: int, duration_ms: int = 300, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> SwipeResult: """Swipe between two points. Simulates a finger swipe gesture. Use for scrolling, dragging, etc. @@ -109,20 +114,20 @@ class InputMixin(ADBBaseMixin): ], device_id, ) - return { - "success": result.success, - "action": "swipe", - "from": {"x": x1, "y": y1}, - "to": {"x": x2, "y": y2}, - "duration_ms": duration_ms, - "error": result.stderr if not result.success else None, - } + return SwipeResult( + success=result.success, + action="swipe", + start={"x": x1, "y": y1}, + end={"x": x2, "y": y2}, + duration_ms=duration_ms, + error=result.stderr if not result.success else None, + ) @mcp_tool() async def input_scroll_down( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> InputResult: """Scroll down one page. Convenience method for common scroll-down gesture. @@ -151,17 +156,17 @@ class InputMixin(ADBBaseMixin): ], device_id, ) - return { - "success": result.success, - "action": "scroll_down", - "error": result.stderr if not result.success else None, - } + return InputResult( + success=result.success, + action="scroll_down", + error=result.stderr if not result.success else None, + ) @mcp_tool() async def input_scroll_up( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> InputResult: """Scroll up one page. Convenience method for common scroll-up gesture. @@ -190,17 +195,17 @@ class InputMixin(ADBBaseMixin): ], device_id, ) - return { - "success": result.success, - "action": "scroll_up", - "error": result.stderr if not result.success else None, - } + return InputResult( + success=result.success, + action="scroll_up", + error=result.stderr if not result.success else None, + ) @mcp_tool() async def input_back( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> InputResult: """Press the Back button. Simulates pressing the Android back button. @@ -214,17 +219,17 @@ class InputMixin(ADBBaseMixin): result = await self.run_shell_args( ["input", "keyevent", "KEYCODE_BACK"], device_id ) - return { - "success": result.success, - "action": "back", - "error": result.stderr if not result.success else None, - } + return InputResult( + success=result.success, + action="back", + error=result.stderr if not result.success else None, + ) @mcp_tool() async def input_home( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> InputResult: """Press the Home button. Returns to the home screen. @@ -238,17 +243,17 @@ class InputMixin(ADBBaseMixin): result = await self.run_shell_args( ["input", "keyevent", "KEYCODE_HOME"], device_id ) - return { - "success": result.success, - "action": "home", - "error": result.stderr if not result.success else None, - } + return InputResult( + success=result.success, + action="home", + error=result.stderr if not result.success else None, + ) @mcp_tool() async def input_recent_apps( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> InputResult: """Open recent apps / app switcher. Shows the recent applications overview. @@ -262,18 +267,18 @@ class InputMixin(ADBBaseMixin): result = await self.run_shell_args( ["input", "keyevent", "KEYCODE_APP_SWITCH"], device_id ) - return { - "success": result.success, - "action": "recent_apps", - "error": result.stderr if not result.success else None, - } + return InputResult( + success=result.success, + action="recent_apps", + error=result.stderr if not result.success else None, + ) @mcp_tool() async def input_key( self, key_code: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> InputResult: """Send a key event. Send any Android key event by code name. @@ -298,19 +303,19 @@ class InputMixin(ADBBaseMixin): clean = f"KEYCODE_{clean.upper()}" result = await self.run_shell_args(["input", "keyevent", clean], device_id) - return { - "success": result.success, - "action": "key", - "key_code": clean, - "error": result.stderr if not result.success else None, - } + return InputResult( + success=result.success, + action="key", + key_code=clean, + error=result.stderr if not result.success else None, + ) @mcp_tool() async def input_text( self, text: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> InputResult: """Type text into the focused input field. Types the specified text as if entered via keyboard. @@ -330,25 +335,26 @@ class InputMixin(ADBBaseMixin): # Check for characters that ADB input text can't handle has_unsafe = any(c in _INPUT_TEXT_UNSAFE for c in text) if has_unsafe: - return { - "success": False, - "error": ( + return InputResult( + success=False, + action="text", + error=( "Text contains special characters that ADB input " "text cannot handle reliably. Use " "clipboard_set(text, paste=True) instead." ), - "text": text, - } + text=text, + ) # ADB input text: spaces must be %s, no shell metacharacters escaped = text.replace(" ", "%s") result = await self.run_shell_args(["input", "text", escaped], device_id) - return { - "success": result.success, - "action": "text", - "text": text, - "error": result.stderr if not result.success else None, - } + return InputResult( + success=result.success, + action="text", + text=text, + error=result.stderr if not result.success else None, + ) # === Developer Mode Tools === @@ -360,7 +366,7 @@ class InputMixin(ADBBaseMixin): self, command: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ShellResult: """Execute arbitrary shell command on device. [DEVELOPER MODE] Run any shell command on the Android device. @@ -381,24 +387,25 @@ class InputMixin(ADBBaseMixin): Command output with stdout, stderr, and return code """ if not is_developer_mode(): - return { - "success": False, - "error": ( + return ShellResult( + success=False, + command=command, + error=( "Developer mode required. " "Enable with config_set_developer_mode(True)" ), - } + ) # Developer shell_command intentionally uses run_shell (string form) # since the user explicitly provides the command string result = await self.run_shell(command, device_id) - return { - "success": result.success, - "command": command, - "stdout": result.stdout, - "stderr": result.stderr, - "returncode": result.returncode, - } + return ShellResult( + success=result.success, + command=command, + stdout=result.stdout, + stderr=result.stderr, + returncode=result.returncode, + ) @mcp_tool( tags={"developer"}, @@ -410,7 +417,7 @@ class InputMixin(ADBBaseMixin): y: int, duration_ms: int = 1000, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> InputResult: """Long press at screen coordinates. [DEVELOPER MODE] Simulates a long press / press-and-hold gesture. @@ -425,10 +432,11 @@ class InputMixin(ADBBaseMixin): Success status """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return InputResult( + success=False, + action="long_press", + error="Developer mode required", + ) # Long press is a swipe with no movement result = await self.run_shell_args( @@ -443,13 +451,13 @@ class InputMixin(ADBBaseMixin): ], device_id, ) - return { - "success": result.success, - "action": "long_press", - "coordinates": {"x": x, "y": y}, - "duration_ms": duration_ms, - "error": result.stderr if not result.success else None, - } + return InputResult( + success=result.success, + action="long_press", + coordinates={"x": x, "y": y}, + duration_ms=duration_ms, + error=result.stderr if not result.success else None, + ) @mcp_tool() async def clipboard_set( @@ -457,7 +465,7 @@ class InputMixin(ADBBaseMixin): text: str, paste: bool = False, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ClipboardSetResult: """Set clipboard text and optionally paste. Sets the device clipboard to the specified text. Unlike input_text, @@ -510,20 +518,20 @@ class InputMixin(ADBBaseMixin): ) preview = text[:100] + "..." if len(text) > 100 else text - response: dict[str, Any] = { - "success": result.success, - "action": "clipboard_set", - "text": preview, - "error": result.stderr if not result.success else None, - } + response = ClipboardSetResult( + success=result.success, + action="clipboard_set", + text=preview, + error=result.stderr if not result.success else None, + ) # Paste if requested if paste and result.success: paste_result = await self.run_shell_args( ["input", "keyevent", "KEYCODE_PASTE"], device_id ) - response["pasted"] = paste_result.success + response.pasted = paste_result.success if not paste_result.success: - response["paste_error"] = paste_result.stderr + response.paste_error = paste_result.stderr return response diff --git a/src/mixins/screenshot.py b/src/mixins/screenshot.py index c1b1dc8..0f4d368 100644 --- a/src/mixins/screenshot.py +++ b/src/mixins/screenshot.py @@ -11,7 +11,14 @@ from fastmcp import Context from fastmcp.contrib.mcp_mixin import mcp_resource, mcp_tool from ..config import get_config, is_developer_mode -from ..models import ScreenshotResult +from ..models import ( + ActionResult, + RecordingResult, + ScreenDensityResult, + ScreenSetResult, + ScreenshotResult, + ScreenSizeResult, +) from .base import ADBBaseMixin @@ -99,7 +106,7 @@ class ScreenshotMixin(ADBBaseMixin): async def screen_size( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ScreenSizeResult: """Get the screen dimensions. Returns the physical screen resolution in pixels. @@ -121,24 +128,24 @@ class ScreenshotMixin(ADBBaseMixin): size = parts[1].strip() if "x" in size: w, h = size.split("x") - return { - "success": True, - "width": int(w), - "height": int(h), - "raw": result.stdout, - } + return ScreenSizeResult( + success=True, + width=int(w), + height=int(h), + raw=result.stdout, + ) - return { - "success": False, - "error": result.stderr or "Could not parse screen size", - "raw": result.stdout, - } + return ScreenSizeResult( + success=False, + error=result.stderr or "Could not parse screen size", + raw=result.stdout, + ) @mcp_tool() async def screen_density( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ScreenDensityResult: """Get the screen density (DPI). Args: @@ -156,24 +163,24 @@ class ScreenshotMixin(ADBBaseMixin): if len(parts) == 2: try: dpi = int(parts[1].strip()) - return { - "success": True, - "dpi": dpi, - } + return ScreenDensityResult( + success=True, + dpi=dpi, + ) except ValueError: pass - return { - "success": False, - "error": result.stderr or "Could not parse density", - "raw": result.stdout, - } + return ScreenDensityResult( + success=False, + error=result.stderr or "Could not parse density", + raw=result.stdout, + ) @mcp_tool() async def screen_on( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ActionResult: """Turn the screen on. Wakes up the device display. Does not unlock. @@ -187,17 +194,17 @@ class ScreenshotMixin(ADBBaseMixin): result = await self.run_shell_args( ["input", "keyevent", "KEYCODE_WAKEUP"], device_id ) - return { - "success": result.success, - "action": "screen_on", - "error": result.stderr if not result.success else None, - } + return ActionResult( + success=result.success, + action="screen_on", + error=result.stderr if not result.success else None, + ) @mcp_tool() async def screen_off( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ActionResult: """Turn the screen off. Puts the device display to sleep. @@ -211,11 +218,11 @@ class ScreenshotMixin(ADBBaseMixin): result = await self.run_shell_args( ["input", "keyevent", "KEYCODE_SLEEP"], device_id ) - return { - "success": result.success, - "action": "screen_off", - "error": result.stderr if not result.success else None, - } + return ActionResult( + success=result.success, + action="screen_off", + error=result.stderr if not result.success else None, + ) # === Developer Mode Tools === @@ -229,7 +236,7 @@ class ScreenshotMixin(ADBBaseMixin): filename: str | None = None, duration_seconds: int = 10, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> RecordingResult: """Record the screen. [DEVELOPER MODE] Records the device screen to a video file. @@ -245,10 +252,10 @@ class ScreenshotMixin(ADBBaseMixin): Recording result with file path """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return RecordingResult( + success=False, + error="Developer mode required", + ) # Generate default filename if not filename: @@ -283,10 +290,10 @@ class ScreenshotMixin(ADBBaseMixin): ) if not result.success: - return { - "success": False, - "error": f"Failed to record: {result.stderr}", - } + return RecordingResult( + success=False, + error=f"Failed to record: {result.stderr}", + ) await ctx.info("Transferring recording to host...") @@ -299,18 +306,18 @@ class ScreenshotMixin(ADBBaseMixin): await self.run_shell_args(["rm", device_temp], device_id) if not pull_result.success: - return { - "success": False, - "error": (f"Failed to pull recording: {pull_result.stderr}"), - } + return RecordingResult( + success=False, + error=f"Failed to pull recording: {pull_result.stderr}", + ) await ctx.info(f"Recording saved: {output_path}") - return { - "success": True, - "local_path": str(output_path), - "duration_seconds": duration, - } + return RecordingResult( + success=True, + local_path=str(output_path), + duration_seconds=duration, + ) @mcp_tool( tags={"developer"}, @@ -321,7 +328,7 @@ class ScreenshotMixin(ADBBaseMixin): width: int, height: int, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ScreenSetResult: """Override screen resolution. [DEVELOPER MODE] Changes the display resolution. @@ -336,21 +343,22 @@ class ScreenshotMixin(ADBBaseMixin): Success status """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return ScreenSetResult( + success=False, + action="set_size", + error="Developer mode required", + ) result = await self.run_shell_args( ["wm", "size", f"{width}x{height}"], device_id ) - return { - "success": result.success, - "action": "set_size", - "width": width, - "height": height, - "error": result.stderr if not result.success else None, - } + return ScreenSetResult( + success=result.success, + action="set_size", + width=width, + height=height, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -359,7 +367,7 @@ class ScreenshotMixin(ADBBaseMixin): async def screen_reset_size( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ActionResult: """Reset screen to physical resolution. [DEVELOPER MODE] Restores the original display resolution. @@ -371,17 +379,18 @@ class ScreenshotMixin(ADBBaseMixin): Success status """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return ActionResult( + success=False, + action="reset_size", + error="Developer mode required", + ) result = await self.run_shell_args(["wm", "size", "reset"], device_id) - return { - "success": result.success, - "action": "reset_size", - "error": result.stderr if not result.success else None, - } + return ActionResult( + success=result.success, + action="reset_size", + error=result.stderr if not result.success else None, + ) # === Resources === @@ -392,7 +401,7 @@ class ScreenshotMixin(ADBBaseMixin): density = await self.screen_density() return { - "width": size.get("width"), - "height": size.get("height"), - "dpi": density.get("dpi"), + "width": size.width, + "height": size.height, + "dpi": density.dpi, } diff --git a/src/mixins/settings.py b/src/mixins/settings.py index 848a94b..a04a910 100644 --- a/src/mixins/settings.py +++ b/src/mixins/settings.py @@ -5,12 +5,21 @@ display configuration, notification access, clipboard, and media control. """ import re -from typing import Any from fastmcp import Context from fastmcp.contrib.mcp_mixin import mcp_tool from ..config import is_developer_mode +from ..models import ( + BrightnessResult, + ClipboardGetResult, + MediaControlResult, + NotificationListResult, + SettingGetResult, + SettingPutResult, + TimeoutResult, + ToggleResult, +) from .base import ADBBaseMixin _VALID_NAMESPACES = {"system", "global", "secure"} @@ -47,7 +56,7 @@ class SettingsMixin(ADBBaseMixin): namespace: str, key: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> SettingGetResult: """Read an Android system setting. Reads a value from the device's settings database. @@ -65,43 +74,43 @@ class SettingsMixin(ADBBaseMixin): The setting value """ if namespace not in _VALID_NAMESPACES: - return { - "success": False, - "error": ( + return SettingGetResult( + success=False, + error=( f"Invalid namespace '{namespace}'. " f"Must be one of: {', '.join(sorted(_VALID_NAMESPACES))}" ), - } + ) if not _SETTING_KEY_PATTERN.match(key): - return { - "success": False, - "error": ( + return SettingGetResult( + success=False, + error=( f"Invalid key '{key}'. Keys must contain " "only letters, digits, underscores, and dots." ), - } + ) result = await self.run_shell_args( ["settings", "get", namespace, key], device_id ) if not result.success: - return { - "success": False, - "error": result.stderr, - } + return SettingGetResult( + success=False, + error=result.stderr, + ) value = result.stdout.strip() is_null = value == "null" - return { - "success": True, - "namespace": namespace, - "key": key, - "value": None if is_null else value, - "exists": not is_null, - } + return SettingGetResult( + success=True, + namespace=namespace, + key=key, + value=None if is_null else value, + exists=not is_null, + ) @mcp_tool( tags={"developer"}, @@ -114,7 +123,7 @@ class SettingsMixin(ADBBaseMixin): key: str, value: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> SettingPutResult: """Write an Android system setting. [DEVELOPER MODE] Writes a value to the device's settings database. @@ -132,28 +141,28 @@ class SettingsMixin(ADBBaseMixin): Result with read-back verification """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return SettingPutResult( + success=False, + error="Developer mode required", + ) if namespace not in _VALID_NAMESPACES: - return { - "success": False, - "error": ( + return SettingPutResult( + success=False, + error=( f"Invalid namespace '{namespace}'. " f"Must be one of: {', '.join(sorted(_VALID_NAMESPACES))}" ), - } + ) if not _SETTING_KEY_PATTERN.match(key): - return { - "success": False, - "error": ( + return SettingPutResult( + success=False, + error=( f"Invalid key '{key}'. Keys must contain " "only letters, digits, underscores, and dots." ), - } + ) # Extra confirmation for secure namespace if namespace == "secure": @@ -165,21 +174,21 @@ class SettingsMixin(ADBBaseMixin): ["Yes, write setting", "Cancel"], ) if confirmation.action != "accept" or confirmation.content == "Cancel": - return { - "success": False, - "cancelled": True, - "message": "Settings write cancelled by user", - } + return SettingPutResult( + success=False, + cancelled=True, + message="Settings write cancelled by user", + ) result = await self.run_shell_args( ["settings", "put", namespace, key, value], device_id ) if not result.success: - return { - "success": False, - "error": result.stderr, - } + return SettingPutResult( + success=False, + error=result.stderr, + ) # Read back to verify verify = await self.run_shell_args( @@ -189,14 +198,14 @@ class SettingsMixin(ADBBaseMixin): await ctx.info(f"Set {namespace}/{key} = {value}") - return { - "success": True, - "namespace": namespace, - "key": key, - "value": value, - "readback": readback, - "verified": readback == value, - } + return SettingPutResult( + success=True, + namespace=namespace, + key=key, + value=value, + readback=readback, + verified=readback == value, + ) @mcp_tool( tags={"developer"}, @@ -206,7 +215,7 @@ class SettingsMixin(ADBBaseMixin): self, enabled: bool, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ToggleResult: """Toggle WiFi on or off. [DEVELOPER MODE] Enables or disables WiFi using the svc command. @@ -220,19 +229,21 @@ class SettingsMixin(ADBBaseMixin): Result with verified WiFi state """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return ToggleResult( + success=False, + action="enable" if enabled else "disable", + error="Developer mode required", + ) action = "enable" if enabled else "disable" result = await self.run_shell_args(["svc", "wifi", action], device_id) if not result.success: - return { - "success": False, - "error": result.stderr, - } + return ToggleResult( + success=False, + action=action, + error=result.stderr, + ) # Verify state change verify = await self.run_shell_args( @@ -240,12 +251,12 @@ class SettingsMixin(ADBBaseMixin): ) current = verify.stdout.strip() if verify.success else "unknown" - return { - "success": True, - "action": action, - "wifi_on": current, - "verified": current == ("1" if enabled else "0"), - } + return ToggleResult( + success=True, + action=action, + wifi_on=current, + verified=current == ("1" if enabled else "0"), + ) @mcp_tool( tags={"developer"}, @@ -255,7 +266,7 @@ class SettingsMixin(ADBBaseMixin): self, enabled: bool, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ToggleResult: """Toggle Bluetooth on or off. [DEVELOPER MODE] Enables or disables Bluetooth using the svc command. @@ -268,19 +279,20 @@ class SettingsMixin(ADBBaseMixin): Result with action taken """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return ToggleResult( + success=False, + action="enable" if enabled else "disable", + error="Developer mode required", + ) action = "enable" if enabled else "disable" result = await self.run_shell_args(["svc", "bluetooth", action], device_id) - return { - "success": result.success, - "action": action, - "error": result.stderr if not result.success else None, - } + return ToggleResult( + success=result.success, + action=action, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -291,7 +303,7 @@ class SettingsMixin(ADBBaseMixin): ctx: Context, enabled: bool, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ToggleResult: """Toggle airplane mode on or off. [DEVELOPER MODE] Enables or disables airplane mode. @@ -307,10 +319,11 @@ class SettingsMixin(ADBBaseMixin): Result with airplane mode state """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return ToggleResult( + success=False, + action="enabled" if enabled else "disabled", + error="Developer mode required", + ) # Warn about network disconnection risk if enabled: @@ -331,11 +344,12 @@ class SettingsMixin(ADBBaseMixin): ["Yes, enable airplane mode", "Cancel"], ) if confirmation.action != "accept" or confirmation.content == "Cancel": - return { - "success": False, - "cancelled": True, - "message": "Airplane mode toggle cancelled by user", - } + return ToggleResult( + success=False, + action="enabled" if enabled else "disabled", + cancelled=True, + message="Airplane mode toggle cancelled by user", + ) # Set the setting value = "1" if enabled else "0" @@ -344,10 +358,11 @@ class SettingsMixin(ADBBaseMixin): ) if not put_result.success: - return { - "success": False, - "error": put_result.stderr, - } + return ToggleResult( + success=False, + action="enabled" if enabled else "disabled", + error=put_result.stderr, + ) # Broadcast the change so the system acts on it await self.run_shell_args( @@ -366,11 +381,11 @@ class SettingsMixin(ADBBaseMixin): action = "enabled" if enabled else "disabled" await ctx.info(f"Airplane mode {action}") - return { - "success": True, - "airplane_mode": enabled, - "action": action, - } + return ToggleResult( + success=True, + action=action, + airplane_mode=enabled, + ) @mcp_tool( tags={"developer"}, @@ -380,7 +395,7 @@ class SettingsMixin(ADBBaseMixin): self, level: int, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> BrightnessResult: """Set screen brightness level. [DEVELOPER MODE] Sets the screen brightness to a specific level. @@ -394,16 +409,16 @@ class SettingsMixin(ADBBaseMixin): Result with brightness level set """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return BrightnessResult( + success=False, + error="Developer mode required", + ) if not 0 <= level <= 255: - return { - "success": False, - "error": f"Brightness level must be 0-255, got {level}", - } + return BrightnessResult( + success=False, + error=f"Brightness level must be 0-255, got {level}", + ) # Disable auto-brightness first await self.run_shell_args( @@ -417,12 +432,12 @@ class SettingsMixin(ADBBaseMixin): device_id, ) - return { - "success": result.success, - "brightness": level, - "auto_brightness": False, - "error": result.stderr if not result.success else None, - } + return BrightnessResult( + success=result.success, + brightness=level, + auto_brightness=False, + error=result.stderr if not result.success else None, + ) @mcp_tool( tags={"developer"}, @@ -432,7 +447,7 @@ class SettingsMixin(ADBBaseMixin): self, seconds: int, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> TimeoutResult: """Set screen timeout duration. [DEVELOPER MODE] Sets how long the screen stays on before @@ -446,16 +461,16 @@ class SettingsMixin(ADBBaseMixin): Result with timeout value set """ if not is_developer_mode(): - return { - "success": False, - "error": "Developer mode required", - } + return TimeoutResult( + success=False, + error="Developer mode required", + ) if seconds < 1 or seconds > 1800: - return { - "success": False, - "error": f"Timeout must be 1-1800 seconds, got {seconds}", - } + return TimeoutResult( + success=False, + error=f"Timeout must be 1-1800 seconds, got {seconds}", + ) # Android stores timeout in milliseconds ms = seconds * 1000 @@ -464,19 +479,19 @@ class SettingsMixin(ADBBaseMixin): device_id, ) - return { - "success": result.success, - "timeout_seconds": seconds, - "timeout_ms": ms, - "error": result.stderr if not result.success else None, - } + return TimeoutResult( + success=result.success, + timeout_seconds=seconds, + timeout_ms=ms, + error=result.stderr if not result.success else None, + ) @mcp_tool() async def notification_list( self, limit: int = 50, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> NotificationListResult: """List recent notifications. Retrieves notifications from the notification shade. @@ -494,10 +509,10 @@ class SettingsMixin(ADBBaseMixin): ) if not result.success: - return { - "success": False, - "error": result.stderr, - } + return NotificationListResult( + success=False, + error=result.stderr, + ) notifications: list[dict[str, str | None]] = [] current: dict[str, str | None] = {} @@ -537,17 +552,17 @@ class SettingsMixin(ADBBaseMixin): if current and len(notifications) < limit: notifications.append(current) - return { - "success": True, - "notifications": notifications, - "count": len(notifications), - } + return NotificationListResult( + success=True, + notifications=notifications, + count=len(notifications), + ) @mcp_tool() async def clipboard_get( self, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> ClipboardGetResult: """Read the device clipboard contents. Retrieves the current text from the device clipboard. @@ -577,19 +592,19 @@ class SettingsMixin(ADBBaseMixin): if result.success and "Parcel(" in result.stdout: text = self._parse_clipboard_parcel(result.stdout) if text is not None: - return { - "success": True, - "text": text, - "method": "service_call", - } + return ClipboardGetResult( + success=True, + text=text, + method="service_call", + ) - return { - "success": False, - "error": ( + return ClipboardGetResult( + success=False, + error=( "Could not read clipboard. The device may have " "an empty clipboard or use an unsupported format." ), - } + ) @staticmethod def _parse_clipboard_parcel(raw: str) -> str | None: @@ -656,7 +671,7 @@ class SettingsMixin(ADBBaseMixin): self, action: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> MediaControlResult: """Control media playback. Sends media key events to control the active media player. @@ -683,19 +698,20 @@ class SettingsMixin(ADBBaseMixin): keycode = _MEDIA_KEYCODES.get(action_lower) if not keycode: - return { - "success": False, - "error": ( + return MediaControlResult( + success=False, + action=action_lower, + error=( f"Unknown action '{action}'. " f"Available: {', '.join(sorted(_MEDIA_KEYCODES))}" ), - } + ) result = await self.run_shell_args(["input", "keyevent", keycode], device_id) - return { - "success": result.success, - "action": action_lower, - "keycode": keycode, - "error": result.stderr if not result.success else None, - } + return MediaControlResult( + success=result.success, + action=action_lower, + keycode=keycode, + error=result.stderr if not result.success else None, + ) diff --git a/src/mixins/ui.py b/src/mixins/ui.py index 8f32dea..81d7f91 100644 --- a/src/mixins/ui.py +++ b/src/mixins/ui.py @@ -11,6 +11,7 @@ from typing import Any from fastmcp import Context from fastmcp.contrib.mcp_mixin import mcp_tool +from ..models import TapTextResult, UIDumpResult, UIFindResult, WaitResult from .base import ADBBaseMixin @@ -28,7 +29,7 @@ class UIMixin(ADBBaseMixin): self, ctx: Context | None = None, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> UIDumpResult: """Dump the current UI hierarchy. Returns the accessibility tree as XML, showing all visible elements @@ -62,10 +63,10 @@ class UIMixin(ADBBaseMixin): if not result.success: if ctx: await ctx.error(f"UI dump failed: {result.stderr}") - return { - "success": False, - "error": f"Failed to dump UI: {result.stderr}", - } + return UIDumpResult( + success=False, + error=f"Failed to dump UI: {result.stderr}", + ) # Read the dump cat_result = await self.run_shell_args(["cat", device_path], device_id) @@ -73,10 +74,10 @@ class UIMixin(ADBBaseMixin): if not cat_result.success: if ctx: await ctx.error(f"Failed to read dump: {cat_result.stderr}") - return { - "success": False, - "error": f"Failed to read UI dump: {cat_result.stderr}", - } + return UIDumpResult( + success=False, + error=f"Failed to read UI dump: {cat_result.stderr}", + ) # Clean up await self.run_shell_args(["rm", device_path], device_id) @@ -89,12 +90,12 @@ class UIMixin(ADBBaseMixin): if ctx: await ctx.info(f"Found {len(clickable_elements)} interactive elements") - return { - "success": True, - "xml": xml_content, - "clickable_elements": clickable_elements, - "element_count": len(clickable_elements), - } + return UIDumpResult( + success=True, + xml=xml_content, + clickable_elements=clickable_elements, + element_count=len(clickable_elements), + ) def _parse_ui_elements(self, xml_content: str) -> list[dict[str, Any]]: """Parse UI XML to extract clickable/important elements.""" @@ -147,7 +148,7 @@ class UIMixin(ADBBaseMixin): resource_id: str | None = None, class_name: str | None = None, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> UIFindResult: """Find UI elements matching criteria. Searches the current UI for elements matching the specified @@ -167,10 +168,10 @@ class UIMixin(ADBBaseMixin): # Get UI dump (internal call, no ctx) dump = await self.ui_dump(device_id=device_id) - if not dump.get("success"): - return dump + if not dump.success: + return UIFindResult(success=False, error=dump.error) - elements = dump["clickable_elements"] + elements = dump.clickable_elements matches = [] for elem in elements: @@ -190,11 +191,11 @@ class UIMixin(ADBBaseMixin): if match: matches.append(elem) - return { - "success": True, - "matches": matches, - "count": len(matches), - } + return UIFindResult( + success=True, + matches=matches, + count=len(matches), + ) @mcp_tool() async def wait_for_text( @@ -203,7 +204,7 @@ class UIMixin(ADBBaseMixin): timeout_seconds: float = 10.0, poll_interval: float = 0.5, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> WaitResult: """Wait for text to appear on screen. Polls the UI hierarchy until the specified text is found @@ -227,27 +228,27 @@ class UIMixin(ADBBaseMixin): # Internal call, no ctx dump = await self.ui_dump(device_id=device_id) - if dump.get("success"): - for elem in dump.get("clickable_elements", []): + if dump.success: + for elem in dump.clickable_elements: if text in elem.get("text", "") or text in elem.get( "content_desc", "" ): - return { - "success": True, - "found": True, - "element": elem, - "wait_time": round(time.time() - start_time, 2), - "attempts": attempts, - } + return WaitResult( + success=True, + found=True, + element=elem, + wait_time=round(time.time() - start_time, 2), + attempts=attempts, + ) await asyncio.sleep(poll_interval) - return { - "success": False, - "found": False, - "error": (f"Text '{text}' not found after {timeout_seconds}s"), - "attempts": attempts, - } + return WaitResult( + success=False, + found=False, + error=f"Text '{text}' not found after {timeout_seconds}s", + attempts=attempts, + ) @mcp_tool() async def wait_for_text_gone( @@ -256,7 +257,7 @@ class UIMixin(ADBBaseMixin): timeout_seconds: float = 10.0, poll_interval: float = 0.5, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> WaitResult: """Wait for text to disappear from screen. Useful for waiting for loading indicators to finish, @@ -279,9 +280,9 @@ class UIMixin(ADBBaseMixin): dump = await self.ui_dump(device_id=device_id) - if dump.get("success"): + if dump.success: found = False - for elem in dump.get("clickable_elements", []): + for elem in dump.clickable_elements: if text in elem.get("text", "") or text in elem.get( "content_desc", "" ): @@ -289,28 +290,28 @@ class UIMixin(ADBBaseMixin): break if not found: - return { - "success": True, - "gone": True, - "wait_time": round(time.time() - start_time, 2), - "attempts": attempts, - } + return WaitResult( + success=True, + gone=True, + wait_time=round(time.time() - start_time, 2), + attempts=attempts, + ) await asyncio.sleep(poll_interval) - return { - "success": False, - "gone": False, - "error": (f"Text '{text}' still present after {timeout_seconds}s"), - "attempts": attempts, - } + return WaitResult( + success=False, + gone=False, + error=f"Text '{text}' still present after {timeout_seconds}s", + attempts=attempts, + ) @mcp_tool() async def tap_text( self, text: str, device_id: str | None = None, - ) -> dict[str, Any]: + ) -> TapTextResult: """Find element by text and tap it. Convenience method that combines ui_find_element + input_tap. @@ -326,30 +327,32 @@ class UIMixin(ADBBaseMixin): # Find element result = await self.ui_find_element(text=text, device_id=device_id) - if not result.get("success"): - return result + if not result.success: + return TapTextResult(success=False, error=result.error, action="tap_text") - matches = result.get("matches", []) + matches = result.matches if not matches: # Try content-desc as fallback result = await self.ui_find_element(content_desc=text, device_id=device_id) - matches = result.get("matches", []) + matches = result.matches if not matches: - return { - "success": False, - "error": f"No element found with text '{text}'", - } + return TapTextResult( + success=False, + error=f"No element found with text '{text}'", + action="tap_text", + ) element = matches[0] center = element.get("center") if not center: - return { - "success": False, - "error": "Element found but could not determine coordinates", - "element": element, - } + return TapTextResult( + success=False, + error="Element found but could not determine coordinates", + action="tap_text", + element=element, + ) # Tap the center tap_result = await self.run_shell_args( @@ -357,11 +360,11 @@ class UIMixin(ADBBaseMixin): device_id, ) - return { - "success": tap_result.success, - "action": "tap_text", - "text": text, - "coordinates": center, - "element": element, - "error": tap_result.stderr if not tap_result.success else None, - } + return TapTextResult( + success=tap_result.success, + action="tap_text", + text=text, + coordinates=center, + element=element, + error=tap_result.stderr if not tap_result.success else None, + ) diff --git a/src/models.py b/src/models.py index c3d8e7a..5475285 100644 --- a/src/models.py +++ b/src/models.py @@ -1,7 +1,11 @@ """Pydantic models for Android ADB MCP Server.""" +from typing import Any + from pydantic import BaseModel, Field +# ── Data Models (not tool results) ────────────────────────────────── + class DeviceInfo(BaseModel): """Android device information returned by ADB.""" @@ -18,7 +22,7 @@ class DeviceInfo(BaseModel): class CommandResult(BaseModel): - """Result of an ADB command execution.""" + """Result of an ADB command execution (internal).""" success: bool = Field(description="Whether the command succeeded") stdout: str = Field(default="", description="Standard output from command") @@ -26,11 +30,399 @@ class CommandResult(BaseModel): returncode: int = Field(description="Command exit code") -class ScreenshotResult(BaseModel): +# ── Base Result ───────────────────────────────────────────────────── + + +class ADBResult(BaseModel): + """Base result for all ADB tool operations.""" + + success: bool = Field(description="Whether the operation succeeded") + error: str | None = Field(None, description="Error message if operation failed") + + +class ActionResult(ADBResult): + """Result of a simple action (tap, press, toggle, etc.).""" + + action: str = Field(description="Action that was performed") + + +# ── Screenshot / Screen ───────────────────────────────────────────── + + +class ScreenshotResult(ADBResult): """Screenshot capture operation result.""" - success: bool = Field(description="Whether screenshot was captured successfully") local_path: str | None = Field( None, description="Absolute path to the saved screenshot file" ) - error: str | None = Field(None, description="Error message if operation failed") + + +class ScreenSizeResult(ADBResult): + """Screen resolution information.""" + + width: int | None = Field(None, description="Screen width in pixels") + height: int | None = Field(None, description="Screen height in pixels") + raw: str | None = Field(None, description="Raw wm size output") + + +class ScreenDensityResult(ADBResult): + """Screen density information.""" + + dpi: int | None = Field(None, description="Screen density in DPI") + raw: str | None = Field(None, description="Raw wm density output") + + +class RecordingResult(ADBResult): + """Screen recording result.""" + + local_path: str | None = Field(None, description="Path to saved recording file") + duration_seconds: int | None = Field(None, description="Actual recording duration") + + +class ScreenSetResult(ActionResult): + """Screen size override result.""" + + width: int | None = Field(None, description="New width in pixels") + height: int | None = Field(None, description="New height in pixels") + + +# ── Input ──────────────────────────────────────────────────────────── + + +class InputResult(ActionResult): + """Result of an input simulation action.""" + + coordinates: dict[str, int] | None = Field( + None, description="Tap/press coordinates {x, y}" + ) + key_code: str | None = Field(None, description="Key code sent") + text: str | None = Field(None, description="Text typed or on clipboard") + duration_ms: int | None = Field(None, description="Duration in ms") + + +class SwipeResult(ActionResult): + """Result of a swipe gesture.""" + + start: dict[str, int] = Field(description="Start coordinates {x, y}") + end: dict[str, int] = Field(description="End coordinates {x, y}") + duration_ms: int = Field(description="Swipe duration in ms") + + +class ClipboardSetResult(ActionResult): + """Result of a clipboard set operation.""" + + text: str = Field(description="Text placed on clipboard (preview)") + pasted: bool | None = Field(None, description="Whether paste was performed") + paste_error: str | None = Field(None, description="Paste error if any") + + +class ShellResult(ADBResult): + """Result of a shell command execution.""" + + command: str = Field(description="Command that was executed") + stdout: str = Field(default="", description="Standard output") + stderr: str = Field(default="", description="Standard error") + returncode: int = Field(default=0, description="Exit code") + + +# ── Apps ───────────────────────────────────────────────────────────── + + +class AppActionResult(ActionResult): + """Result of an app management action.""" + + package: str | None = Field(None, description="Target package name") + url: str | None = Field(None, description="URL opened") + output: str | None = Field(None, description="Command output") + apk: str | None = Field(None, description="APK path") + kept_data: bool | None = Field(None, description="Whether data was preserved") + cancelled: bool | None = Field(None, description="Whether operation was cancelled") + message: str | None = Field(None, description="Status message") + + +class AppCurrentResult(ADBResult): + """Currently focused app information.""" + + package: str | None = Field(None, description="Foreground package name") + activity: str | None = Field(None, description="Current activity class") + raw: str | None = Field( + None, description="Raw dumpsys output (if no package found)" + ) + + +class PackageListResult(ADBResult): + """List of installed packages.""" + + packages: list[str] = Field(default_factory=list, description="Package names") + count: int = Field(default=0, description="Number of packages") + + +class IntentResult(ActionResult): + """Result of an intent-based operation.""" + + component: str | None = Field(None, description="Activity component") + intent_action: str | None = Field(None, description="Intent action") + data_uri: str | None = Field(None, description="Data URI") + broadcast_action: str | None = Field(None, description="Broadcast action sent") + package: str | None = Field(None, description="Target package") + output: str | None = Field(None, description="Command output") + + +# ── Devices ────────────────────────────────────────────────────────── + + +class DeviceSelectResult(ADBResult): + """Device selection/status result.""" + + device: dict[str, Any] | str | None = Field( + None, description="Selected device info" + ) + message: str | None = Field(None, description="Status message") + available: list[str] | dict[str, Any] | None = Field( + None, description="Available devices" + ) + cached_info: Any | None = Field(None, description="Cached device info (if any)") + + +class DeviceInfoResult(ADBResult): + """Comprehensive device information.""" + + battery: dict[str, Any] | None = Field( + None, description="Battery state (level, status, plugged)" + ) + ip_address: str | None = Field(None, description="Device IP address") + wifi_ssid: str | None = Field(None, description="Connected WiFi SSID") + model: str | None = Field(None, description="Device model") + manufacturer: str | None = Field(None, description="Device manufacturer") + device_name: str | None = Field(None, description="Device codename") + android_version: str | None = Field(None, description="Android version") + sdk_version: str | None = Field(None, description="SDK API level") + storage: dict[str, int] | None = Field( + None, description="Storage info (total_kb, used_kb, available_kb)" + ) + + +class RebootResult(ActionResult): + """Device reboot result.""" + + mode: str = Field(description="Reboot mode (normal, recovery, bootloader)") + cancelled: bool | None = Field(None, description="Whether cancelled") + message: str | None = Field(None, description="Status message") + + +class LogcatResult(ADBResult): + """Logcat capture result.""" + + action: str | None = Field(None, description="Action (logcat_clear)") + lines_requested: int | None = Field(None, description="Lines requested") + filter: str | None = Field(None, description="Filter spec applied") + output: str | None = Field(None, description="Log output") + + +# ── Connectivity ───────────────────────────────────────────────────── + + +class ConnectResult(ADBResult): + """ADB network connection result.""" + + address: str = Field(description="Device address (host:port)") + output: str | None = Field(None, description="ADB command output") + already_connected: bool | None = Field( + None, description="Whether already connected" + ) + + +class TcpipResult(ADBResult): + """TCP/IP mode switch result.""" + + port: int | None = Field(None, description="ADB TCP port") + device_ip: str | None = Field(None, description="Device IP address") + connect_address: str | None = Field( + None, description="Address for adb_connect (ip:port)" + ) + message: str | None = Field(None, description="Status message") + + +class DevicePropertiesResult(ADBResult): + """Batch device properties result.""" + + identity: dict[str, str] | None = Field( + None, description="Identity props (model, manufacturer, serial)" + ) + software: dict[str, str] | None = Field( + None, description="Software props (android version, sdk, build)" + ) + hardware: dict[str, str] | None = Field( + None, description="Hardware props (chipset, ABI)" + ) + system: dict[str, str] | None = Field( + None, description="System props (timezone, locale)" + ) + + +# ── UI ─────────────────────────────────────────────────────────────── + + +class UIDumpResult(ADBResult): + """UI hierarchy dump result.""" + + xml: str | None = Field(None, description="Raw XML hierarchy") + clickable_elements: list[dict[str, Any]] = Field( + default_factory=list, description="Interactive elements" + ) + element_count: int = Field(default=0, description="Number of interactive elements") + + +class UIFindResult(ADBResult): + """UI element search result.""" + + matches: list[dict[str, Any]] = Field( + default_factory=list, description="Matching elements" + ) + count: int = Field(default=0, description="Number of matches") + + +class WaitResult(ADBResult): + """Wait/poll operation result.""" + + found: bool | None = Field(None, description="Whether text was found") + gone: bool | None = Field(None, description="Whether text disappeared") + element: dict[str, Any] | None = Field(None, description="Found element info") + wait_time: float | None = Field(None, description="Time waited in seconds") + attempts: int = Field(default=0, description="Poll attempts made") + + +class TapTextResult(ActionResult): + """Tap-by-text result.""" + + text: str | None = Field(None, description="Text searched for") + coordinates: dict[str, int] | None = Field(None, description="Tapped coordinates") + element: dict[str, Any] | None = Field(None, description="Element that was tapped") + + +# ── Files ──────────────────────────────────────────────────────────── + + +class FileTransferResult(ActionResult): + """File push/pull result.""" + + local_path: str | None = Field(None, description="Host file path") + device_path: str | None = Field(None, description="Device file path") + output: str | None = Field(None, description="ADB output") + + +class FileListResult(ADBResult): + """Directory listing result.""" + + path: str | None = Field(None, description="Listed directory path") + files: list[dict[str, Any]] = Field( + default_factory=list, description="File entries" + ) + count: int = Field(default=0, description="Number of files") + + +class FileDeleteResult(ActionResult): + """File deletion result.""" + + path: str | None = Field(None, description="Deleted file path") + cancelled: bool | None = Field(None, description="Whether cancelled") + message: str | None = Field(None, description="Status message") + + +class FileExistsResult(ADBResult): + """File existence check result.""" + + path: str | None = Field(None, description="Checked path") + exists: bool = Field(description="Whether the file exists") + + +# ── Settings ───────────────────────────────────────────────────────── + + +class SettingGetResult(ADBResult): + """Settings read result.""" + + namespace: str | None = Field(None, description="Settings namespace") + key: str | None = Field(None, description="Setting key") + value: str | None = Field(None, description="Setting value") + exists: bool | None = Field(None, description="Whether key exists") + + +class SettingPutResult(ADBResult): + """Settings write result.""" + + namespace: str | None = Field(None, description="Settings namespace") + key: str | None = Field(None, description="Setting key") + value: str | None = Field(None, description="Value written") + readback: str | None = Field(None, description="Read-back verification") + verified: bool | None = Field(None, description="Whether verified") + cancelled: bool | None = Field(None, description="Whether cancelled") + message: str | None = Field(None, description="Status message") + + +class ToggleResult(ActionResult): + """Radio/toggle result (wifi, bluetooth, airplane).""" + + verified: bool | None = Field(None, description="Whether state verified") + wifi_on: str | None = Field(None, description="WiFi state after toggle") + airplane_mode: bool | None = Field(None, description="Airplane mode state") + cancelled: bool | None = Field(None, description="Whether cancelled") + message: str | None = Field(None, description="Status message") + + +class BrightnessResult(ADBResult): + """Screen brightness result.""" + + brightness: int | None = Field(None, description="Brightness level 0-255") + auto_brightness: bool | None = Field(None, description="Auto-brightness state") + + +class TimeoutResult(ADBResult): + """Screen timeout result.""" + + timeout_seconds: int | None = Field(None, description="Timeout in seconds") + timeout_ms: int | None = Field(None, description="Timeout in milliseconds") + + +class NotificationListResult(ADBResult): + """Notification list result.""" + + notifications: list[dict[str, str | None]] = Field( + default_factory=list, description="Notification entries" + ) + count: int = Field(default=0, description="Number of notifications") + + +class ClipboardGetResult(ADBResult): + """Clipboard read result.""" + + text: str | None = Field(None, description="Clipboard text content") + method: str | None = Field(None, description="Method used to read") + + +class MediaControlResult(ActionResult): + """Media control result.""" + + keycode: str | None = Field(None, description="Android keycode sent") + + +# ── Config ─────────────────────────────────────────────────────────── + + +class ConfigStatusResult(BaseModel): + """Server configuration status.""" + + developer_mode: bool = Field(description="Developer mode enabled") + auto_select_single_device: bool = Field(description="Auto-select when one device") + default_screenshot_dir: str | None = Field( + None, description="Screenshot output directory" + ) + current_device: str | None = Field(None, description="Currently selected device") + + +class ConfigResult(ADBResult): + """Configuration change result.""" + + developer_mode: bool | None = Field(None, description="Developer mode state") + screenshot_dir: str | None = Field(None, description="Screenshot directory") + message: str | None = Field(None, description="Status message") diff --git a/src/server.py b/src/server.py index 5e1a8b7..9d66a91 100644 --- a/src/server.py +++ b/src/server.py @@ -29,6 +29,7 @@ from .mixins import ( SettingsMixin, UIMixin, ) +from .models import ConfigResult, ConfigStatusResult class ADBServer( @@ -59,7 +60,7 @@ class ADBServer( # === Configuration Tools === @mcp_tool() - async def config_status(self) -> dict[str, Any]: + async def config_status(self) -> ConfigStatusResult: """Get current server configuration. Shows developer mode status and other settings. @@ -68,15 +69,15 @@ class ADBServer( Current configuration values """ config = get_config() - return { - "developer_mode": config.developer_mode, - "auto_select_single_device": config.auto_select_single_device, - "default_screenshot_dir": config.default_screenshot_dir, - "current_device": self.get_current_device(), - } + return ConfigStatusResult( + developer_mode=config.developer_mode, + auto_select_single_device=config.auto_select_single_device, + default_screenshot_dir=config.default_screenshot_dir, + current_device=self.get_current_device(), + ) @mcp_tool() - async def config_set_developer_mode(self, enabled: bool) -> dict[str, Any]: + async def config_set_developer_mode(self, enabled: bool) -> ConfigResult: """Enable or disable developer mode. Developer mode unlocks advanced tools: @@ -97,18 +98,18 @@ class ADBServer( config = get_config() config.developer_mode = enabled - return { - "success": True, - "developer_mode": enabled, - "message": ( + return ConfigResult( + success=True, + developer_mode=enabled, + message=( "Developer mode enabled. Advanced tools are now available." if enabled else "Developer mode disabled. Using standard tools only." ), - } + ) @mcp_tool() - async def config_set_screenshot_dir(self, directory: str | None) -> dict[str, Any]: + async def config_set_screenshot_dir(self, directory: str | None) -> ConfigResult: """Set default directory for screenshots. Screenshots will be saved to this directory by default. @@ -123,10 +124,10 @@ class ADBServer( config = get_config() config.default_screenshot_dir = directory - return { - "success": True, - "screenshot_dir": directory, - } + return ConfigResult( + success=True, + screenshot_dir=directory, + ) # === Help / Discovery === diff --git a/tests/test_apps.py b/tests/test_apps.py index 652ee9c..d5401fc 100644 --- a/tests/test_apps.py +++ b/tests/test_apps.py @@ -9,8 +9,8 @@ class TestAppLaunch: async def test_launch(self, server): server.run_shell_args.return_value = ok() result = await server.app_launch("com.android.chrome") - assert result["success"] is True - assert result["package"] == "com.android.chrome" + assert result.success is True + assert result.package == "com.android.chrome" args = server.run_shell_args.call_args[0][0] assert "monkey" in args assert "com.android.chrome" in args @@ -18,15 +18,15 @@ class TestAppLaunch: async def test_failure(self, server): server.run_shell_args.return_value = fail("not found") result = await server.app_launch("com.missing.app") - assert result["success"] is False + assert result.success is False class TestAppOpenUrl: async def test_open(self, server): server.run_shell_args.return_value = ok() result = await server.app_open_url("https://example.com") - assert result["success"] is True - assert result["url"] == "https://example.com" + assert result.success is True + assert result.url == "https://example.com" args = server.run_shell_args.call_args[0][0] assert "am" in args assert "android.intent.action.VIEW" in args @@ -36,8 +36,8 @@ class TestAppClose: async def test_close(self, server): server.run_shell_args.return_value = ok() result = await server.app_close("com.example.app") - assert result["success"] is True - assert result["package"] == "com.example.app" + assert result.success is True + assert result.package == "com.example.app" args = server.run_shell_args.call_args[0][0] assert "am" in args assert "force-stop" in args @@ -51,22 +51,22 @@ class TestAppCurrent: ) server.run_shell_args.return_value = ok(stdout=focused) result = await server.app_current() - assert result["success"] is True - assert result["package"] == "com.android.chrome" + assert result.success is True + assert result.package == "com.android.chrome" async def test_focused_app_format(self, server): server.run_shell_args.return_value = ok( stdout=" mFocusedApp=ActivityRecord{abc com.example/.MainActivity t123}" ) result = await server.app_current() - assert result["success"] is True - assert result["package"] == "com.example" + assert result.success is True + assert result.package == "com.example" async def test_no_focus(self, server): server.run_shell_args.return_value = ok(stdout="no focus info") result = await server.app_current() - assert result["success"] is True - assert result["package"] is None + assert result.success is True + assert result.package is None class TestAppListPackages: @@ -76,9 +76,9 @@ class TestAppListPackages: stdout="package:com.android.chrome\npackage:com.example.app\n" ) result = await server.app_list_packages() - assert result["success"] is True - assert result["count"] == 2 - assert "com.android.chrome" in result["packages"] + assert result.success is True + assert result.count == 2 + assert "com.android.chrome" in result.packages @pytest.mark.usefixtures("_dev_mode") async def test_filter(self, server): @@ -86,8 +86,8 @@ class TestAppListPackages: stdout="package:com.android.chrome\npackage:com.example.app\n" ) result = await server.app_list_packages(filter_text="chrome") - assert result["count"] == 1 - assert "com.android.chrome" in result["packages"] + assert result.count == 1 + assert "com.android.chrome" in result.packages @pytest.mark.usefixtures("_dev_mode") async def test_third_party(self, server): @@ -99,7 +99,7 @@ class TestAppListPackages: @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.app_list_packages() - assert result["success"] is False + assert result.success is False class TestAppInstall: @@ -107,7 +107,7 @@ class TestAppInstall: async def test_install(self, server): server.run_adb.return_value = ok(stdout="Success") result = await server.app_install("/tmp/app.apk") - assert result["success"] is True + assert result.success is True args = server.run_adb.call_args[0][0] assert "install" in args assert "-r" in args @@ -115,7 +115,7 @@ class TestAppInstall: @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.app_install("/tmp/app.apk") - assert result["success"] is False + assert result.success is False class TestAppUninstall: @@ -124,15 +124,15 @@ class TestAppUninstall: ctx.set_elicit("accept", "Yes, uninstall") server.run_adb.return_value = ok() result = await server.app_uninstall(ctx, "com.example.app") - assert result["success"] is True - assert result["package"] == "com.example.app" + assert result.success is True + assert result.package == "com.example.app" @pytest.mark.usefixtures("_dev_mode") async def test_keep_data(self, server, ctx): ctx.set_elicit("accept", "Yes, uninstall") server.run_adb.return_value = ok() result = await server.app_uninstall(ctx, "com.example.app", keep_data=True) - assert result["kept_data"] is True + assert result.kept_data is True args = server.run_adb.call_args[0][0] assert "-k" in args @@ -140,8 +140,8 @@ class TestAppUninstall: async def test_cancelled(self, server, ctx): ctx.set_elicit("accept", "Cancel") result = await server.app_uninstall(ctx, "com.example.app") - assert result["success"] is False - assert result.get("cancelled") is True + assert result.success is False + assert result.cancelled is True class TestAppClearData: @@ -150,13 +150,13 @@ class TestAppClearData: ctx.set_elicit("accept", "Yes, clear all data") server.run_shell_args.return_value = ok() result = await server.app_clear_data(ctx, "com.example.app") - assert result["success"] is True + assert result.success is True @pytest.mark.usefixtures("_dev_mode") async def test_cancelled(self, server, ctx): ctx.set_elicit("accept", "Cancel") result = await server.app_clear_data(ctx, "com.example.app") - assert result.get("cancelled") is True + assert result.cancelled is True class TestActivityStart: @@ -164,8 +164,8 @@ class TestActivityStart: async def test_basic(self, server): server.run_shell_args.return_value = ok() result = await server.activity_start("com.example/.MainActivity") - assert result["success"] is True - assert result["component"] == "com.example/.MainActivity" + assert result.success is True + assert result.component == "com.example/.MainActivity" args = server.run_shell_args.call_args[0][0] assert "am" in args assert "start" in args @@ -208,7 +208,7 @@ class TestActivityStart: @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.activity_start("com.example/.Act") - assert result["success"] is False + assert result.success is False class TestBroadcastSend: @@ -216,8 +216,8 @@ class TestBroadcastSend: async def test_basic(self, server): server.run_shell_args.return_value = ok() result = await server.broadcast_send("com.example.ACTION") - assert result["success"] is True - assert result["broadcast_action"] == "com.example.ACTION" + assert result.success is True + assert result.broadcast_action == "com.example.ACTION" @pytest.mark.usefixtures("_dev_mode") async def test_with_package(self, server): @@ -230,4 +230,4 @@ class TestBroadcastSend: @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.broadcast_send("ACTION") - assert result["success"] is False + assert result.success is False diff --git a/tests/test_connectivity.py b/tests/test_connectivity.py index 607f840..0aa2d5a 100644 --- a/tests/test_connectivity.py +++ b/tests/test_connectivity.py @@ -9,38 +9,38 @@ class TestAdbConnect: async def test_success(self, server): server.run_adb.return_value = ok(stdout="connected to 10.0.0.1:5555") result = await server.adb_connect("10.0.0.1") - assert result["success"] is True - assert result["address"] == "10.0.0.1:5555" + assert result.success is True + assert result.address == "10.0.0.1:5555" server.run_adb.assert_called_once_with(["connect", "10.0.0.1:5555"]) async def test_custom_port(self, server): server.run_adb.return_value = ok(stdout="connected to 10.0.0.1:5556") result = await server.adb_connect("10.0.0.1", port=5556) - assert result["address"] == "10.0.0.1:5556" + assert result.address == "10.0.0.1:5556" async def test_already_connected(self, server): server.run_adb.return_value = ok(stdout="already connected to 10.0.0.1:5555") result = await server.adb_connect("10.0.0.1") - assert result["success"] is True - assert result["already_connected"] is True + assert result.success is True + assert result.already_connected is True async def test_failure(self, server): server.run_adb.return_value = ok(stdout="failed to connect") result = await server.adb_connect("10.0.0.1") - assert result["success"] is False + assert result.success is False class TestAdbDisconnect: async def test_success(self, server): server.run_adb.return_value = ok(stdout="disconnected 10.0.0.1:5555") result = await server.adb_disconnect("10.0.0.1") - assert result["success"] is True - assert result["address"] == "10.0.0.1:5555" + assert result.success is True + assert result.address == "10.0.0.1:5555" async def test_failure(self, server): server.run_adb.return_value = ok(stdout="error: no such device") result = await server.adb_disconnect("10.0.0.1") - assert result["success"] is False + assert result.success is False class TestAdbTcpip: @@ -51,23 +51,23 @@ class TestAdbTcpip: ) server.run_adb.return_value = ok(stdout="restarting in TCP mode port: 5555") result = await server.adb_tcpip(ctx) - assert result["success"] is True - assert result["device_ip"] == "192.168.1.100" - assert result["connect_address"] == "192.168.1.100:5555" + assert result.success is True + assert result.device_ip == "192.168.1.100" + assert result.connect_address == "192.168.1.100:5555" @pytest.mark.usefixtures("_dev_mode") async def test_rejects_network_device(self, server, ctx): server.set_current_device("10.20.0.25:5555") result = await server.adb_tcpip(ctx) - assert result["success"] is False - assert "already a network device" in result["error"] + assert result.success is False + assert "already a network device" in result.error @pytest.mark.usefixtures("_dev_mode") async def test_no_wifi_ip(self, server, ctx): server.run_shell_args.return_value = ok(stdout="wlan0: no ip") result = await server.adb_tcpip(ctx) - assert result["success"] is False - assert "WiFi" in result["error"] + assert result.success is False + assert "WiFi" in result.error @pytest.mark.usefixtures("_dev_mode") async def test_custom_port(self, server, ctx): @@ -76,27 +76,27 @@ class TestAdbTcpip: ) server.run_adb.return_value = ok() result = await server.adb_tcpip(ctx, port=5556) - assert result["port"] == 5556 - assert result["connect_address"] == "192.168.1.50:5556" + assert result.port == 5556 + assert result.connect_address == "192.168.1.50:5556" @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server, ctx): result = await server.adb_tcpip(ctx) - assert result["success"] is False - assert "developer mode" in result["error"].lower() + assert result.success is False + assert "developer mode" in result.error.lower() class TestAdbPair: async def test_success(self, server): server.run_adb.return_value = ok(stdout="Successfully paired to 10.0.0.1:37000") result = await server.adb_pair("10.0.0.1", 37000, "123456") - assert result["success"] is True + assert result.success is True server.run_adb.assert_called_once_with(["pair", "10.0.0.1:37000", "123456"]) async def test_failure(self, server): server.run_adb.return_value = fail("Failed: wrong code") result = await server.adb_pair("10.0.0.1", 37000, "000000") - assert result["success"] is False + assert result.success is False class TestDeviceProperties: @@ -110,13 +110,13 @@ class TestDeviceProperties: } server.get_device_property.side_effect = lambda p, d=None: props.get(p) result = await server.device_properties() - assert result["success"] is True - assert result["identity"]["model"] == "Pixel 6" - assert result["software"]["android_version"] == "14" - assert result["hardware"]["chipset"] == "gs101" + assert result.success is True + assert result.identity["model"] == "Pixel 6" + assert result.software["android_version"] == "14" + assert result.hardware["chipset"] == "gs101" async def test_no_properties(self, server): server.get_device_property.return_value = None result = await server.device_properties() - assert result["success"] is False - assert "No properties" in result["error"] + assert result.success is False + assert "No properties" in result.error diff --git a/tests/test_devices.py b/tests/test_devices.py index 0e45999..b18fa7d 100644 --- a/tests/test_devices.py +++ b/tests/test_devices.py @@ -37,7 +37,7 @@ class TestDevicesUse: stdout="List of devices attached\nABC123\tdevice\n" ) result = await server.devices_use("ABC123") - assert result["success"] is True + assert result.success is True assert server.get_current_device() == "ABC123" async def test_not_found(self, server): @@ -45,30 +45,30 @@ class TestDevicesUse: stdout="List of devices attached\nOTHER\tdevice\n" ) result = await server.devices_use("MISSING") - assert result["success"] is False - assert "not found" in result["error"] + assert result.success is False + assert "not found" in result.error async def test_offline_device(self, server): server.run_adb.return_value = ok( stdout="List of devices attached\nABC123\toffline\n" ) result = await server.devices_use("ABC123") - assert result["success"] is False - assert "offline" in result["error"] + assert result.success is False + assert "offline" in result.error class TestDevicesCurrent: async def test_no_device_set(self, server): server.run_adb.return_value = ok(stdout="List of devices attached\n") result = await server.devices_current() - assert result["device"] is None + assert result.device is None async def test_auto_detect_single(self, server): server.run_adb.return_value = ok( stdout="List of devices attached\nABC123\tdevice\n" ) result = await server.devices_current() - assert result.get("available") is not None + assert result.available is not None async def test_device_set(self, server): # Pre-populate cache and set device @@ -78,7 +78,8 @@ class TestDevicesCurrent: await server.devices_list() server.set_current_device("ABC123") result = await server.devices_current() - assert result["device"]["device_id"] == "ABC123" + # device is a dict from model_dump() + assert result.device["device_id"] == "ABC123" class TestDeviceInfo: @@ -105,16 +106,16 @@ class TestDeviceInfo: }.get(p) result = await server.device_info() - assert result["success"] is True - assert result["battery"]["level"] == 85 - assert result["ip_address"] == "192.168.1.100" - assert result["wifi_ssid"] == "MyNetwork" - assert result["model"] == "Pixel 6" + assert result.success is True + assert result.battery["level"] == 85 + assert result.ip_address == "192.168.1.100" + assert result.wifi_ssid == "MyNetwork" + assert result.model == "Pixel 6" async def test_device_offline(self, server): server.run_shell_args.return_value = fail("device offline") result = await server.device_info() - assert result["success"] is False + assert result.success is False class TestDeviceReboot: @@ -123,27 +124,27 @@ class TestDeviceReboot: ctx.set_elicit("accept", "Yes, reboot now") server.run_adb.return_value = ok() result = await server.device_reboot(ctx) - assert result["success"] is True - assert result["mode"] == "normal" + assert result.success is True + assert result.mode == "normal" @pytest.mark.usefixtures("_dev_mode") async def test_reboot_recovery(self, server, ctx): ctx.set_elicit("accept", "Yes, reboot now") server.run_adb.return_value = ok() result = await server.device_reboot(ctx, mode="recovery") - assert result["mode"] == "recovery" + assert result.mode == "recovery" @pytest.mark.usefixtures("_dev_mode") async def test_cancelled(self, server, ctx): ctx.set_elicit("accept", "Cancel") result = await server.device_reboot(ctx) - assert result["success"] is False - assert result.get("cancelled") is True + assert result.success is False + assert result.cancelled is True @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server, ctx): result = await server.device_reboot(ctx) - assert result["success"] is False + assert result.success is False class TestLogcat: @@ -152,30 +153,30 @@ class TestLogcat: logline = "01-01 00:00:00.000 I/TAG: message" server.run_shell_args.return_value = ok(stdout=logline) result = await server.logcat_capture() - assert result["success"] is True - assert result["output"].startswith("01-01") + assert result.success is True + assert result.output.startswith("01-01") @pytest.mark.usefixtures("_dev_mode") async def test_with_filter(self, server): server.run_shell_args.return_value = ok(stdout="filtered output") result = await server.logcat_capture(filter_spec="MyApp:D *:S") - assert result["filter"] == "MyApp:D *:S" + assert result.filter == "MyApp:D *:S" @pytest.mark.usefixtures("_dev_mode") async def test_clear_first(self, server): server.run_shell_args.side_effect = [ok(), ok(stdout="fresh logs")] result = await server.logcat_capture(clear_first=True) - assert result["success"] is True + assert result.success is True assert server.run_shell_args.call_count == 2 @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.logcat_capture() - assert result["success"] is False + assert result.success is False @pytest.mark.usefixtures("_dev_mode") async def test_logcat_clear(self, server): server.run_shell_args.return_value = ok() result = await server.logcat_clear() - assert result["success"] is True - assert result["action"] == "logcat_clear" + assert result.success is True + assert result.action == "logcat_clear" diff --git a/tests/test_files.py b/tests/test_files.py index c9d4524..28f5444 100644 --- a/tests/test_files.py +++ b/tests/test_files.py @@ -12,19 +12,19 @@ class TestFilePush: local_file.write_text("content") server.run_adb.return_value = ok(stdout="1 file pushed") result = await server.file_push(ctx, str(local_file), "/sdcard/test.txt") - assert result["success"] is True - assert result["device_path"] == "/sdcard/test.txt" + assert result.success is True + assert result.device_path == "/sdcard/test.txt" @pytest.mark.usefixtures("_dev_mode") async def test_local_not_found(self, server, ctx): result = await server.file_push(ctx, "/nonexistent/file.txt", "/sdcard/") - assert result["success"] is False - assert "not found" in result["error"] + assert result.success is False + assert "not found" in result.error @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server, ctx): result = await server.file_push(ctx, "/tmp/f", "/sdcard/f") - assert result["success"] is False + assert result.success is False class TestFilePull: @@ -34,18 +34,18 @@ class TestFilePull: result = await server.file_pull( ctx, "/sdcard/test.txt", str(tmp_path / "out.txt") ) - assert result["success"] is True + assert result.success is True @pytest.mark.usefixtures("_dev_mode") async def test_default_local_path(self, server, ctx): server.run_adb.return_value = ok() result = await server.file_pull(ctx, "/sdcard/data.db") - assert "data.db" in result["local_path"] + assert "data.db" in result.local_path @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server, ctx): result = await server.file_pull(ctx, "/sdcard/f") - assert result["success"] is False + assert result.success is False class TestFileList: @@ -59,23 +59,23 @@ class TestFileList: ) ) result = await server.file_list("/sdcard/") - assert result["success"] is True - assert result["count"] == 2 - assert result["files"][0]["name"] == "Documents" - assert result["files"][0]["is_directory"] is True - assert result["files"][1]["name"] == "test.txt" - assert result["files"][1]["is_directory"] is False + assert result.success is True + assert result.count == 2 + assert result.files[0]["name"] == "Documents" + assert result.files[0]["is_directory"] is True + assert result.files[1]["name"] == "test.txt" + assert result.files[1]["is_directory"] is False @pytest.mark.usefixtures("_dev_mode") async def test_failure(self, server): server.run_shell_args.return_value = fail("No such file") result = await server.file_list("/nonexistent/") - assert result["success"] is False + assert result.success is False @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.file_list() - assert result["success"] is False + assert result.success is False class TestFileDelete: @@ -84,20 +84,20 @@ class TestFileDelete: ctx.set_elicit("accept", "Yes, delete") server.run_shell_args.return_value = ok() result = await server.file_delete(ctx, "/sdcard/old.txt") - assert result["success"] is True - assert result["path"] == "/sdcard/old.txt" + assert result.success is True + assert result.path == "/sdcard/old.txt" @pytest.mark.usefixtures("_dev_mode") async def test_cancelled(self, server, ctx): ctx.set_elicit("accept", "Cancel") result = await server.file_delete(ctx, "/sdcard/keep.txt") - assert result["success"] is False - assert result.get("cancelled") is True + assert result.success is False + assert result.cancelled is True @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server, ctx): result = await server.file_delete(ctx, "/sdcard/f") - assert result["success"] is False + assert result.success is False class TestFileExists: @@ -105,16 +105,16 @@ class TestFileExists: async def test_exists(self, server): server.run_shell_args.return_value = ok() result = await server.file_exists("/sdcard/file.txt") - assert result["success"] is True - assert result["exists"] is True + assert result.success is True + assert result.exists is True @pytest.mark.usefixtures("_dev_mode") async def test_not_exists(self, server): server.run_shell_args.return_value = fail() result = await server.file_exists("/sdcard/missing.txt") - assert result["exists"] is False + assert result.exists is False @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.file_exists("/sdcard/f") - assert result["success"] is False + assert result.success is False diff --git a/tests/test_input.py b/tests/test_input.py index 79eafc6..4e187b7 100644 --- a/tests/test_input.py +++ b/tests/test_input.py @@ -9,8 +9,8 @@ class TestInputTap: async def test_tap(self, server): server.run_shell_args.return_value = ok() result = await server.input_tap(100, 200) - assert result["success"] is True - assert result["coordinates"] == {"x": 100, "y": 200} + assert result.success is True + assert result.coordinates == {"x": 100, "y": 200} server.run_shell_args.assert_called_once_with( ["input", "tap", "100", "200"], None ) @@ -25,23 +25,23 @@ class TestInputTap: async def test_tap_failure(self, server): server.run_shell_args.return_value = fail("no device") result = await server.input_tap(0, 0) - assert result["success"] is False - assert result["error"] == "no device" + assert result.success is False + assert result.error == "no device" class TestInputSwipe: async def test_swipe(self, server): server.run_shell_args.return_value = ok() result = await server.input_swipe(0, 100, 0, 500, duration_ms=500) - assert result["success"] is True - assert result["from"] == {"x": 0, "y": 100} - assert result["to"] == {"x": 0, "y": 500} - assert result["duration_ms"] == 500 + assert result.success is True + assert result.start == {"x": 0, "y": 100} + assert result.end == {"x": 0, "y": 500} + assert result.duration_ms == 500 async def test_swipe_default_duration(self, server): server.run_shell_args.return_value = ok() result = await server.input_swipe(0, 0, 100, 100) - assert result["duration_ms"] == 300 + assert result.duration_ms == 300 class TestInputScroll: @@ -52,8 +52,8 @@ class TestInputScroll: ok(), ] result = await server.input_scroll_down() - assert result["success"] is True - assert result["action"] == "scroll_down" + assert result.success is True + assert result.action == "scroll_down" # Verify swipe args: center x, 65% down to 25% down swipe_call = server.run_shell_args.call_args_list[1] @@ -70,8 +70,8 @@ class TestInputScroll: ok(), ] result = await server.input_scroll_up() - assert result["success"] is True - assert result["action"] == "scroll_up" + assert result.success is True + assert result.action == "scroll_up" async def test_scroll_fallback_dimensions(self, server): # wm size fails, should fall back to 1080x1920 @@ -80,14 +80,14 @@ class TestInputScroll: ok(), ] result = await server.input_scroll_down() - assert result["success"] is True + assert result.success is True class TestInputKeys: async def test_back(self, server): server.run_shell_args.return_value = ok() result = await server.input_back() - assert result["action"] == "back" + assert result.action == "back" server.run_shell_args.assert_called_once_with( ["input", "keyevent", "KEYCODE_BACK"], None ) @@ -95,46 +95,46 @@ class TestInputKeys: async def test_home(self, server): server.run_shell_args.return_value = ok() result = await server.input_home() - assert result["action"] == "home" + assert result.action == "home" async def test_recent_apps(self, server): server.run_shell_args.return_value = ok() result = await server.input_recent_apps() - assert result["action"] == "recent_apps" + assert result.action == "recent_apps" class TestInputKey: async def test_full_keycode(self, server): server.run_shell_args.return_value = ok() result = await server.input_key("KEYCODE_ENTER") - assert result["key_code"] == "KEYCODE_ENTER" + assert result.key_code == "KEYCODE_ENTER" async def test_auto_prefix(self, server): server.run_shell_args.return_value = ok() result = await server.input_key("ENTER") - assert result["key_code"] == "KEYCODE_ENTER" + assert result.key_code == "KEYCODE_ENTER" async def test_strips_dangerous_chars(self, server): server.run_shell_args.return_value = ok() result = await server.input_key("KEYCODE_ENTER; rm -rf /") # Shell metacharacters stripped - assert ";" not in result["key_code"] - assert " " not in result["key_code"] - assert "-" not in result["key_code"] - assert "/" not in result["key_code"] + assert ";" not in result.key_code + assert " " not in result.key_code + assert "-" not in result.key_code + assert "/" not in result.key_code async def test_lowercase_normalized(self, server): server.run_shell_args.return_value = ok() result = await server.input_key("enter") - assert result["key_code"] == "KEYCODE_ENTER" + assert result.key_code == "KEYCODE_ENTER" class TestInputText: async def test_simple_text(self, server): server.run_shell_args.return_value = ok() result = await server.input_text("hello") - assert result["success"] is True - assert result["text"] == "hello" + assert result.success is True + assert result.text == "hello" server.run_shell_args.assert_called_once_with(["input", "text", "hello"], None) async def test_spaces_escaped(self, server): @@ -147,20 +147,20 @@ class TestInputText: async def test_rejects_special_chars(self, server): for char in "'\"\\`$(){}[]|&;<>!~#%^*?": result = await server.input_text(f"text{char}here") - assert result["success"] is False - assert "clipboard_set" in result["error"] + assert result.success is False + assert "clipboard_set" in result.error async def test_rejects_semicolon_injection(self, server): result = await server.input_text("hello; rm -rf /") - assert result["success"] is False + assert result.success is False class TestClipboardSet: async def test_cmd_clipboard(self, server): server.run_shell_args.return_value = ok() result = await server.clipboard_set("test text") - assert result["success"] is True - assert result["action"] == "clipboard_set" + assert result.success is True + assert result.action == "clipboard_set" async def test_cmd_clipboard_not_implemented_falls_back(self, server): # First call: cmd clipboard returns "no shell command" @@ -170,7 +170,7 @@ class TestClipboardSet: ok(stdout="Broadcast completed: result=-1"), ] result = await server.clipboard_set("test") - assert result["success"] is True + assert result.success is True assert server.run_shell_args.call_count == 2 async def test_no_receiver_reports_failure(self, server): @@ -179,15 +179,15 @@ class TestClipboardSet: ok(stdout="Broadcast completed: result=0"), # No receiver ] result = await server.clipboard_set("test") - assert result["success"] is False - assert "no broadcast receiver" in result["error"].lower() + assert result.success is False + assert "no broadcast receiver" in result.error.lower() async def test_paste(self, server): # First call: cmd clipboard set, second call: paste keyevent server.run_shell_args.side_effect = [ok(), ok()] result = await server.clipboard_set("text", paste=True) - assert result["success"] is True - assert result["pasted"] is True + assert result.success is True + assert result.pasted is True # Verify KEYCODE_PASTE was sent paste_call = server.run_shell_args.call_args_list[1] assert "KEYCODE_PASTE" in paste_call[0][0] @@ -196,8 +196,8 @@ class TestClipboardSet: server.run_shell_args.return_value = ok() long_text = "x" * 200 result = await server.clipboard_set(long_text) - assert len(result["text"]) < 200 - assert result["text"].endswith("...") + assert len(result.text) < 200 + assert result.text.endswith("...") class TestInputLongPress: @@ -205,9 +205,9 @@ class TestInputLongPress: async def test_long_press(self, server): server.run_shell_args.return_value = ok() result = await server.input_long_press(100, 200, duration_ms=2000) - assert result["success"] is True - assert result["action"] == "long_press" - assert result["duration_ms"] == 2000 + assert result.success is True + assert result.action == "long_press" + assert result.duration_ms == 2000 # Long press = swipe from same point to same point args = server.run_shell_args.call_args[0][0] assert args[2] == args[4] # x1 == x2 @@ -216,8 +216,8 @@ class TestInputLongPress: @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.input_long_press(0, 0) - assert result["success"] is False - assert "developer mode" in result["error"].lower() + assert result.success is False + assert "developer mode" in result.error.lower() class TestShellCommand: @@ -225,12 +225,12 @@ class TestShellCommand: async def test_executes(self, server): server.run_shell.return_value = ok(stdout="output") result = await server.shell_command("ls /sdcard") - assert result["success"] is True - assert result["stdout"] == "output" - assert result["command"] == "ls /sdcard" + assert result.success is True + assert result.stdout == "output" + assert result.command == "ls /sdcard" @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.shell_command("ls") - assert result["success"] is False - assert "developer mode" in result["error"].lower() + assert result.success is False + assert "developer mode" in result.error.lower() diff --git a/tests/test_screenshot.py b/tests/test_screenshot.py index 880a9fa..21e11b9 100644 --- a/tests/test_screenshot.py +++ b/tests/test_screenshot.py @@ -34,49 +34,49 @@ class TestScreenSize: async def test_physical(self, server): server.run_shell_args.return_value = ok(stdout="Physical size: 1080x1920") result = await server.screen_size() - assert result["success"] is True - assert result["width"] == 1080 - assert result["height"] == 1920 + assert result.success is True + assert result.width == 1080 + assert result.height == 1920 async def test_override(self, server): server.run_shell_args.return_value = ok( stdout="Physical size: 1080x1920\nOverride size: 720x1280" ) result = await server.screen_size() - assert result["success"] is True + assert result.success is True # Should parse the first match - assert result["width"] == 1080 + assert result.width == 1080 async def test_failure(self, server): server.run_shell_args.return_value = fail("error") result = await server.screen_size() - assert result["success"] is False + assert result.success is False class TestScreenDensity: async def test_density(self, server): server.run_shell_args.return_value = ok(stdout="Physical density: 420") result = await server.screen_density() - assert result["success"] is True - assert result["dpi"] == 420 + assert result.success is True + assert result.dpi == 420 async def test_failure(self, server): server.run_shell_args.return_value = fail("error") result = await server.screen_density() - assert result["success"] is False + assert result.success is False class TestScreenOnOff: async def test_screen_on(self, server): server.run_shell_args.return_value = ok() result = await server.screen_on() - assert result["success"] is True - assert result["action"] == "screen_on" + assert result.success is True + assert result.action == "screen_on" async def test_screen_off(self, server): server.run_shell_args.return_value = ok() result = await server.screen_off() - assert result["action"] == "screen_off" + assert result.action == "screen_off" class TestScreenRecord: @@ -92,8 +92,8 @@ class TestScreenRecord: filename="test.mp4", duration_seconds=5, ) - assert result["success"] is True - assert result["duration_seconds"] == 5 + assert result.success is True + assert result.duration_seconds == 5 @pytest.mark.usefixtures("_dev_mode") async def test_duration_capped(self, server, ctx, tmp_path): @@ -103,12 +103,12 @@ class TestScreenRecord: server.run_shell_args.side_effect = [ok(), ok()] server.run_adb.return_value = ok() result = await server.screen_record(ctx, duration_seconds=999) - assert result["duration_seconds"] == 180 # Capped at 180 + assert result.duration_seconds == 180 # Capped at 180 @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server, ctx): result = await server.screen_record(ctx) - assert result["success"] is False + assert result.success is False class TestScreenSetSize: @@ -116,17 +116,17 @@ class TestScreenSetSize: async def test_set(self, server): server.run_shell_args.return_value = ok() result = await server.screen_set_size(720, 1280) - assert result["success"] is True - assert result["width"] == 720 + assert result.success is True + assert result.width == 720 @pytest.mark.usefixtures("_dev_mode") async def test_reset(self, server): server.run_shell_args.return_value = ok() result = await server.screen_reset_size() - assert result["success"] is True - assert result["action"] == "reset_size" + assert result.success is True + assert result.action == "reset_size" @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.screen_set_size(720, 1280) - assert result["success"] is False + assert result.success is False diff --git a/tests/test_server.py b/tests/test_server.py index dfd633b..619ef3c 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -4,33 +4,33 @@ class TestConfigStatus: async def test_status(self, server): result = await server.config_status() - assert "developer_mode" in result - assert "auto_select_single_device" in result - assert "current_device" in result + assert hasattr(result, "developer_mode") + assert hasattr(result, "auto_select_single_device") + assert hasattr(result, "current_device") async def test_reflects_current_device(self, server): server.set_current_device("ABC123") result = await server.config_status() - assert result["current_device"] == "ABC123" + assert result.current_device == "ABC123" class TestConfigSetDeveloperMode: async def test_enable(self, server): result = await server.config_set_developer_mode(True) - assert result["success"] is True - assert result["developer_mode"] is True + assert result.success is True + assert result.developer_mode is True async def test_disable(self, server): result = await server.config_set_developer_mode(False) - assert result["developer_mode"] is False + assert result.developer_mode is False class TestConfigSetScreenshotDir: async def test_set(self, server): result = await server.config_set_screenshot_dir("/tmp/shots") - assert result["success"] is True - assert result["screenshot_dir"] == "/tmp/shots" + assert result.success is True + assert result.screenshot_dir == "/tmp/shots" async def test_clear(self, server): result = await server.config_set_screenshot_dir(None) - assert result["screenshot_dir"] is None + assert result.screenshot_dir is None diff --git a/tests/test_settings.py b/tests/test_settings.py index 37b1f10..aaccbaf 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -10,37 +10,37 @@ class TestSettingsGet: async def test_valid(self, server): server.run_shell_args.return_value = ok(stdout="1") result = await server.settings_get("global", "wifi_on") - assert result["success"] is True - assert result["value"] == "1" - assert result["exists"] is True + assert result.success is True + assert result.value == "1" + assert result.exists is True async def test_null_value(self, server): server.run_shell_args.return_value = ok(stdout="null") result = await server.settings_get("global", "missing_key") - assert result["success"] is True - assert result["value"] is None - assert result["exists"] is False + assert result.success is True + assert result.value is None + assert result.exists is False async def test_invalid_namespace(self, server): result = await server.settings_get("invalid", "key") - assert result["success"] is False - assert "Invalid namespace" in result["error"] + assert result.success is False + assert "Invalid namespace" in result.error async def test_invalid_key(self, server): result = await server.settings_get("global", "bad key!") - assert result["success"] is False - assert "Invalid key" in result["error"] + assert result.success is False + assert "Invalid key" in result.error async def test_all_namespaces_valid(self, server): server.run_shell_args.return_value = ok(stdout="value") for ns in ("system", "global", "secure"): result = await server.settings_get(ns, "test_key") - assert result["success"] is True + assert result.success is True async def test_key_with_dots(self, server): server.run_shell_args.return_value = ok(stdout="value") result = await server.settings_get("global", "wifi.scan_always_enabled") - assert result["success"] is True + assert result.success is True class TestSettingsPut: @@ -48,26 +48,26 @@ class TestSettingsPut: async def test_write_and_verify(self, server, ctx): server.run_shell_args.side_effect = [ok(), ok(stdout="128")] result = await server.settings_put(ctx, "system", "screen_brightness", "128") - assert result["success"] is True - assert result["readback"] == "128" - assert result["verified"] is True + assert result.success is True + assert result.readback == "128" + assert result.verified is True @pytest.mark.usefixtures("_dev_mode") async def test_invalid_namespace(self, server, ctx): result = await server.settings_put(ctx, "bad", "key", "val") - assert result["success"] is False + assert result.success is False @pytest.mark.usefixtures("_dev_mode") async def test_invalid_key(self, server, ctx): result = await server.settings_put(ctx, "global", "k;ey", "val") - assert result["success"] is False + assert result.success is False @pytest.mark.usefixtures("_dev_mode") async def test_secure_namespace_elicits(self, server, ctx): ctx.set_elicit("accept", "Yes, write setting") server.run_shell_args.side_effect = [ok(), ok(stdout="val")] result = await server.settings_put(ctx, "secure", "key", "val") - assert result["success"] is True + assert result.success is True # Verify elicitation happened assert any("secure" in msg for _, msg in ctx.messages) @@ -75,13 +75,13 @@ class TestSettingsPut: async def test_secure_cancelled(self, server, ctx): ctx.set_elicit("accept", "Cancel") result = await server.settings_put(ctx, "secure", "key", "val") - assert result["success"] is False - assert result.get("cancelled") is True + assert result.success is False + assert result.cancelled is True @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server, ctx): result = await server.settings_put(ctx, "system", "k", "v") - assert result["success"] is False + assert result.success is False class TestWifiToggle: @@ -89,21 +89,21 @@ class TestWifiToggle: async def test_enable(self, server): server.run_shell_args.side_effect = [ok(), ok(stdout="1")] result = await server.wifi_toggle(True) - assert result["success"] is True - assert result["action"] == "enable" - assert result["verified"] is True + assert result.success is True + assert result.action == "enable" + assert result.verified is True @pytest.mark.usefixtures("_dev_mode") async def test_disable(self, server): server.run_shell_args.side_effect = [ok(), ok(stdout="0")] result = await server.wifi_toggle(False) - assert result["action"] == "disable" - assert result["verified"] is True + assert result.action == "disable" + assert result.verified is True @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.wifi_toggle(True) - assert result["success"] is False + assert result.success is False class TestBluetoothToggle: @@ -111,13 +111,13 @@ class TestBluetoothToggle: async def test_enable(self, server): server.run_shell_args.return_value = ok() result = await server.bluetooth_toggle(True) - assert result["success"] is True - assert result["action"] == "enable" + assert result.success is True + assert result.action == "enable" @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.bluetooth_toggle(False) - assert result["success"] is False + assert result.success is False class TestAirplaneModeToggle: @@ -126,8 +126,8 @@ class TestAirplaneModeToggle: ctx.set_elicit("accept", "Yes, enable airplane mode") server.run_shell_args.side_effect = [ok(), ok()] result = await server.airplane_mode_toggle(ctx, True) - assert result["success"] is True - assert result["airplane_mode"] is True + assert result.success is True + assert result.airplane_mode is True @pytest.mark.usefixtures("_dev_mode") async def test_enable_network_device_warns(self, server, ctx): @@ -135,7 +135,7 @@ class TestAirplaneModeToggle: ctx.set_elicit("accept", "Yes, enable airplane mode") server.run_shell_args.side_effect = [ok(), ok()] result = await server.airplane_mode_toggle(ctx, True) - assert result["success"] is True + assert result.success is True # Should have warned about network disconnection warns = [msg for level, msg in ctx.messages if "sever" in msg.lower()] assert len(warns) > 0 @@ -144,14 +144,14 @@ class TestAirplaneModeToggle: async def test_cancelled(self, server, ctx): ctx.set_elicit("accept", "Cancel") result = await server.airplane_mode_toggle(ctx, True) - assert result["success"] is False - assert result.get("cancelled") is True + assert result.success is False + assert result.cancelled is True @pytest.mark.usefixtures("_dev_mode") async def test_disable_no_elicitation(self, server, ctx): server.run_shell_args.side_effect = [ok(), ok()] result = await server.airplane_mode_toggle(ctx, False) - assert result["success"] is True + assert result.success is True # No elicitation for disable elicits = [m for level, m in ctx.messages if level == "elicit"] assert len(elicits) == 0 @@ -159,7 +159,7 @@ class TestAirplaneModeToggle: @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server, ctx): result = await server.airplane_mode_toggle(ctx, True) - assert result["success"] is False + assert result.success is False class TestScreenBrightness: @@ -167,25 +167,25 @@ class TestScreenBrightness: async def test_set(self, server): server.run_shell_args.side_effect = [ok(), ok()] result = await server.screen_brightness(128) - assert result["success"] is True - assert result["brightness"] == 128 - assert result["auto_brightness"] is False + assert result.success is True + assert result.brightness == 128 + assert result.auto_brightness is False @pytest.mark.usefixtures("_dev_mode") async def test_out_of_range(self, server): result = await server.screen_brightness(300) - assert result["success"] is False - assert "0-255" in result["error"] + assert result.success is False + assert "0-255" in result.error @pytest.mark.usefixtures("_dev_mode") async def test_negative(self, server): result = await server.screen_brightness(-1) - assert result["success"] is False + assert result.success is False @pytest.mark.usefixtures("_no_dev_mode") async def test_requires_dev_mode(self, server): result = await server.screen_brightness(128) - assert result["success"] is False + assert result.success is False class TestScreenTimeout: @@ -193,20 +193,20 @@ class TestScreenTimeout: async def test_set(self, server): server.run_shell_args.return_value = ok() result = await server.screen_timeout(30) - assert result["success"] is True - assert result["timeout_seconds"] == 30 - assert result["timeout_ms"] == 30000 + assert result.success is True + assert result.timeout_seconds == 30 + assert result.timeout_ms == 30000 @pytest.mark.usefixtures("_dev_mode") async def test_too_large(self, server): result = await server.screen_timeout(9999) - assert result["success"] is False - assert "1-1800" in result["error"] + assert result.success is False + assert "1-1800" in result.error @pytest.mark.usefixtures("_dev_mode") async def test_zero(self, server): result = await server.screen_timeout(0) - assert result["success"] is False + assert result.success is False class TestNotificationList: @@ -227,11 +227,11 @@ class TestNotificationList: """ server.run_shell_args.return_value = ok(stdout=dumpsys_output) result = await server.notification_list() - assert result["success"] is True - assert result["count"] == 2 - assert result["notifications"][0]["package"] == "com.example.app" - assert result["notifications"][0]["title"] == "Test Title" - assert result["notifications"][0]["text"] == "Test message body" + assert result.success is True + assert result.count == 2 + assert result.notifications[0]["package"] == "com.example.app" + assert result.notifications[0]["title"] == "Test Title" + assert result.notifications[0]["text"] == "Test message body" async def test_limit(self, server): # Build output with many notifications @@ -241,13 +241,13 @@ class TestNotificationList: lines.append(f" android.title=Title {i}") server.run_shell_args.return_value = ok(stdout="\n".join(lines)) result = await server.notification_list(limit=3) - assert result["count"] <= 3 + assert result.count <= 3 async def test_empty(self, server): server.run_shell_args.return_value = ok(stdout="") result = await server.notification_list() - assert result["success"] is True - assert result["count"] == 0 + assert result.success is True + assert result.count == 0 class TestClipboardGet: @@ -255,8 +255,8 @@ class TestClipboardGet: # Build parcel programmatically with correct encoding server.run_shell_args.return_value = ok(stdout=_build_parcel("hello world")) result = await server.clipboard_get() - assert result["success"] is True - assert result["text"] == "hello world" + assert result.success is True + assert result.text == "hello world" async def test_empty_clipboard(self, server): server.run_shell_args.return_value = ok( @@ -264,12 +264,12 @@ class TestClipboardGet: ) result = await server.clipboard_get() # No text/plain marker = not parseable - assert result["success"] is False + assert result.success is False async def test_failure(self, server): server.run_shell_args.return_value = fail("error") result = await server.clipboard_get() - assert result["success"] is False + assert result.success is False def _build_parcel(text: str) -> str: @@ -361,31 +361,31 @@ class TestMediaControl: async def test_play(self, server): server.run_shell_args.return_value = ok() result = await server.media_control("play") - assert result["success"] is True - assert result["action"] == "play" - assert result["keycode"] == "KEYCODE_MEDIA_PLAY" + assert result.success is True + assert result.action == "play" + assert result.keycode == "KEYCODE_MEDIA_PLAY" async def test_all_actions(self, server): server.run_shell_args.return_value = ok() for action, keycode in _MEDIA_KEYCODES.items(): result = await server.media_control(action) - assert result["success"] is True - assert result["keycode"] == keycode + assert result.success is True + assert result.keycode == keycode async def test_case_insensitive(self, server): server.run_shell_args.return_value = ok() result = await server.media_control("PLAY") - assert result["success"] is True - assert result["action"] == "play" + assert result.success is True + assert result.action == "play" async def test_unknown_action(self, server): result = await server.media_control("rewind") - assert result["success"] is False - assert "Unknown action" in result["error"] - assert "play" in result["error"] # Lists available actions + assert result.success is False + assert "Unknown action" in result.error + assert "play" in result.error # Lists available actions async def test_whitespace_stripped(self, server): server.run_shell_args.return_value = ok() result = await server.media_control(" pause ") - assert result["success"] is True - assert result["action"] == "pause" + assert result.success is True + assert result.action == "pause" diff --git a/tests/test_ui.py b/tests/test_ui.py index 17102e1..7033c28 100644 --- a/tests/test_ui.py +++ b/tests/test_ui.py @@ -28,14 +28,14 @@ class TestUiDump: ok(), # rm cleanup ] result = await server.ui_dump(ctx) - assert result["success"] is True - assert result["element_count"] >= 2 # Settings + Wi-Fi at minimum - assert "xml" in result + assert result.success is True + assert result.element_count >= 2 # Settings + Wi-Fi at minimum + assert result.xml is not None async def test_dump_failure(self, server, ctx): server.run_shell_args.return_value = fail("error") result = await server.ui_dump(ctx) - assert result["success"] is False + assert result.success is False class TestParseUiElements: @@ -64,30 +64,30 @@ class TestUiFindElement: async def test_find_by_text(self, server): server.run_shell_args.side_effect = [ok(), ok(stdout=SAMPLE_UI_XML), ok()] result = await server.ui_find_element(text="Settings") - assert result["success"] is True - assert result["count"] == 1 - assert result["matches"][0]["text"] == "Settings" + assert result.success is True + assert result.count == 1 + assert result.matches[0]["text"] == "Settings" async def test_find_by_resource_id(self, server): server.run_shell_args.side_effect = [ok(), ok(stdout=SAMPLE_UI_XML), ok()] result = await server.ui_find_element(resource_id="title") # Settings and Wi-Fi both have "title" in their resource-id - assert result["count"] >= 2 + assert result.count >= 2 async def test_not_found(self, server): server.run_shell_args.side_effect = [ok(), ok(stdout=SAMPLE_UI_XML), ok()] result = await server.ui_find_element(text="Missing") - assert result["success"] is True - assert result["count"] == 0 + assert result.success is True + assert result.count == 0 class TestWaitForText: async def test_found_immediately(self, server): server.run_shell_args.side_effect = [ok(), ok(stdout=SAMPLE_UI_XML), ok()] result = await server.wait_for_text("Settings", timeout_seconds=1) - assert result["success"] is True - assert result["found"] is True - assert result["attempts"] == 1 + assert result.success is True + assert result.found is True + assert result.attempts == 1 async def test_timeout(self, server): server.run_shell_args.side_effect = [ @@ -98,8 +98,8 @@ class TestWaitForText: result = await server.wait_for_text( "Missing", timeout_seconds=0.1, poll_interval=0.05 ) - assert result["success"] is False - assert result["found"] is False + assert result.success is False + assert result.found is False class TestWaitForTextGone: @@ -110,8 +110,8 @@ class TestWaitForTextGone: ok(), ] result = await server.wait_for_text_gone("Missing", timeout_seconds=1) - assert result["success"] is True - assert result["gone"] is True + assert result.success is True + assert result.gone is True class TestTapText: @@ -124,8 +124,8 @@ class TestTapText: ok(), # tap ] result = await server.tap_text("Settings") - assert result["success"] is True - assert result["coordinates"] == {"x": 100, "y": 125} + assert result.success is True + assert result.coordinates == {"x": 100, "y": 125} async def test_not_found(self, server): server.run_shell_args.side_effect = [ @@ -137,5 +137,5 @@ class TestTapText: ok(), # fallback search by content_desc ] result = await server.tap_text("NonExistent") - assert result["success"] is False - assert "No element found" in result["error"] + assert result.success is False + assert "No element found" in result.error