MCP/mcp-adb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace dict returns with typed Pydantic response models across all 65 tools

Browse Source 
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.
This commit is contained in:
Ryan Malloy 2026-02-11 03:57:25 -07:00
parent fb297f7937
commit 3614ba8f8f
19 changed files with 1558 additions and 1054 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

252
src/mixins/apps.py
View File

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

118
src/mixins/connectivity.py
View File

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

216
src/mixins/devices.py
View File

@ -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,
)

156
src/mixins/files.py
View File

@ -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,
)

206
src/mixins/input.py
View File

@ -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

165
src/mixins/screenshot.py
View File

@ -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,
}

328
src/mixins/settings.py
View File

@ -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,
)

157
src/mixins/ui.py
View File

@ -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,
)

400
src/models.py
View File

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

37
src/server.py
View File

@ -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 ===

68
tests/test_apps.py
View File

@ -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

56
tests/test_connectivity.py
View File

@ -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

55
tests/test_devices.py
View File

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

50
tests/test_files.py
View File

@ -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

96
tests/test_input.py
View File

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

42
tests/test_screenshot.py
View File

@ -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

20
tests/test_server.py
View File

@ -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

148
tests/test_settings.py
View File

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

42
tests/test_ui.py
View File

@ -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