Compare commits

...

3 Commits

Author SHA1 Message Date
321b6073da Bump version to 0.5.0
Typed Pydantic response models replace dict returns across all 65 tools.
216-test pytest suite added with full coverage.
2026-02-11 04:24:24 -07:00
3614ba8f8f Replace dict returns with typed Pydantic response models across all 65 tools
Every tool now returns a structured BaseModel instead of dict[str, Any],
giving callers attribute access, IDE autocomplete, and schema validation.
Adds ~30 model classes to models.py and updates all test assertions.
2026-02-11 03:57:25 -07:00
fb297f7937 Add pytest suite (216 tests) and fix UI/notification parser bugs
Test infrastructure with conftest fixtures mocking run_shell_args/run_adb
for device-free testing across all 8 mixins.

Fixed: UI parser regex couldn't match hyphenated XML attributes
(content-desc, resource-id). Notification parser captured trailing
parenthesis in package names.
2026-02-11 03:38:37 -07:00
25 changed files with 3360 additions and 769 deletions

View File

@ -1,6 +1,6 @@
[project]
name = "mcadb"
version = "0.4.0"
version = "0.5.0"
description = "Android ADB MCP Server for device automation via Model Context Protocol"
authors = [
{name = "Ryan Malloy", email = "ryan@supported.systems"}
@ -52,6 +52,9 @@ line-length = 88
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
[tool.mypy]
python_version = "3.11"
strict = true

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

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

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

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

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

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

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] = {}
@ -513,7 +528,7 @@ class SettingsMixin(ADBBaseMixin):
break
current = {}
# Extract package from NotificationRecord line
pkg_match = re.search(r"pkg=(\S+)", stripped)
pkg_match = re.search(r"pkg=([\w.]+)", stripped)
if pkg_match:
current["package"] = pkg_match.group(1)
@ -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,
)

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."""
@ -102,7 +103,7 @@ class UIMixin(ADBBaseMixin):
# Regex to find node elements with their attributes
node_pattern = re.compile(r"<node\s+([^>]+?)(?:/>|>)", re.DOTALL)
attr_pattern = re.compile(r'(\w+)="([^"]*)"')
attr_pattern = re.compile(r'([\w-]+)="([^"]*)"')
for match in node_pattern.finditer(xml_content):
attrs_str = match.group(1)
@ -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,
)

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

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

0
tests/__init__.py Normal file
View File

124
tests/conftest.py Normal file
View File

@ -0,0 +1,124 @@
"""Shared test fixtures for mcadb tests."""
from dataclasses import dataclass, field
from typing import Any
from unittest.mock import AsyncMock
import pytest
from src.models import CommandResult
from src.server import ADBServer
# --- Helpers ---
def ok(stdout: str = "", stderr: str = "") -> CommandResult:
"""Create a successful CommandResult."""
return CommandResult(success=True, stdout=stdout, stderr=stderr, returncode=0)
def fail(stderr: str = "error", stdout: str = "") -> CommandResult:
"""Create a failed CommandResult."""
return CommandResult(success=False, stdout=stdout, stderr=stderr, returncode=1)
# --- Mock Context ---
@dataclass
class ElicitResult:
"""Minimal stand-in for FastMCP's ElicitationResult."""
action: str = "accept"
content: str = ""
@dataclass
class MockContext:
"""Mock MCP Context that records calls for assertion."""
messages: list[tuple[str, str]] = field(default_factory=list)
_elicit_response: ElicitResult = field(default_factory=ElicitResult)
async def info(self, msg: str) -> None:
self.messages.append(("info", msg))
async def warning(self, msg: str) -> None:
self.messages.append(("warning", msg))
async def error(self, msg: str) -> None:
self.messages.append(("error", msg))
async def elicit(self, msg: str, options: list[str] | None = None) -> ElicitResult:
self.messages.append(("elicit", msg))
return self._elicit_response
def set_elicit(self, action: str = "accept", content: str = "") -> None:
"""Configure the next elicit response."""
self._elicit_response = ElicitResult(action=action, content=content)
# --- Fixtures ---
def _reset_config(monkeypatch: pytest.MonkeyPatch, config_dir: Any) -> None:
"""Reset the Config singleton and point it at a temp directory.
CONFIG_DIR and CONFIG_FILE are module-level variables computed at
import time, so setting the env var isn't enough — we must patch
the variables directly.
"""
from pathlib import Path
config_path = Path(config_dir)
monkeypatch.setattr("src.config.Config._instance", None)
monkeypatch.setattr("src.config.CONFIG_DIR", config_path)
monkeypatch.setattr("src.config.CONFIG_FILE", config_path / "config.json")
@pytest.fixture
def server() -> ADBServer:
"""Create an ADBServer with mocked ADB execution.
Both run_adb and run_shell_args are replaced with AsyncMock,
so no real subprocess calls are made. Configure return values
per-test with server.run_adb.return_value = ok("...").
"""
s = ADBServer()
s.run_adb = AsyncMock(return_value=ok()) # type: ignore[method-assign]
s.run_shell_args = AsyncMock(return_value=ok()) # type: ignore[method-assign]
s.run_shell = AsyncMock(return_value=ok()) # type: ignore[method-assign]
s.get_device_property = AsyncMock(return_value=None) # type: ignore[method-assign]
return s
@pytest.fixture
def ctx() -> MockContext:
"""Create a mock MCP Context."""
return MockContext()
@pytest.fixture
def _dev_mode(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None:
"""Enable developer mode for the test."""
_reset_config(monkeypatch, tmp_path / "dev-config")
from src.config import get_config
config = get_config()
config._settings["developer_mode"] = True
@pytest.fixture
def _no_dev_mode(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None:
"""Disable developer mode for the test."""
_reset_config(monkeypatch, tmp_path / "nodev-config")
from src.config import get_config
config = get_config()
config._settings["developer_mode"] = False
@pytest.fixture(autouse=True)
def _isolate_config(monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None:
"""Isolate config to a temp directory so tests don't touch real config."""
_reset_config(monkeypatch, tmp_path / "config")

233
tests/test_apps.py Normal file
View File

@ -0,0 +1,233 @@
"""Tests for apps mixin (launch, close, current, install, intents)."""
import pytest
from tests.conftest import fail, ok
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"
args = server.run_shell_args.call_args[0][0]
assert "monkey" in args
assert "com.android.chrome" in args
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
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"
args = server.run_shell_args.call_args[0][0]
assert "am" in args
assert "android.intent.action.VIEW" in args
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"
args = server.run_shell_args.call_args[0][0]
assert "am" in args
assert "force-stop" in args
class TestAppCurrent:
async def test_parse_focused(self, server):
focused = (
" mCurrentFocus=Window{abc com.android.chrome"
"/org.chromium.chrome.browser.ChromeTabbedActivity}"
)
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"
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"
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
class TestAppListPackages:
@pytest.mark.usefixtures("_dev_mode")
async def test_list(self, server):
server.run_shell_args.return_value = ok(
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
@pytest.mark.usefixtures("_dev_mode")
async def test_filter(self, server):
server.run_shell_args.return_value = ok(
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
@pytest.mark.usefixtures("_dev_mode")
async def test_third_party(self, server):
server.run_shell_args.return_value = ok(stdout="package:com.user.app\n")
await server.app_list_packages(third_party_only=True)
args = server.run_shell_args.call_args[0][0]
assert "-3" in args
@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
class TestAppInstall:
@pytest.mark.usefixtures("_dev_mode")
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
args = server.run_adb.call_args[0][0]
assert "install" in args
assert "-r" in args
@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
class TestAppUninstall:
@pytest.mark.usefixtures("_dev_mode")
async def test_uninstall(self, server, ctx):
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"
@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
args = server.run_adb.call_args[0][0]
assert "-k" in args
@pytest.mark.usefixtures("_dev_mode")
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.cancelled is True
class TestAppClearData:
@pytest.mark.usefixtures("_dev_mode")
async def test_clear(self, server, ctx):
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
@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.cancelled is True
class TestActivityStart:
@pytest.mark.usefixtures("_dev_mode")
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"
args = server.run_shell_args.call_args[0][0]
assert "am" in args
assert "start" in args
assert "-n" in args
@pytest.mark.usefixtures("_dev_mode")
async def test_with_action_and_data(self, server):
server.run_shell_args.return_value = ok()
await server.activity_start(
"com.example/.DeepLink",
action="android.intent.action.VIEW",
data_uri="myapp://product/123",
)
args = server.run_shell_args.call_args[0][0]
assert "-a" in args
assert "-d" in args
@pytest.mark.usefixtures("_dev_mode")
async def test_with_extras(self, server):
server.run_shell_args.return_value = ok()
await server.activity_start(
"com.example/.Act",
extras={"key": "value", "flag": "true", "count": "42"},
)
args = server.run_shell_args.call_args[0][0]
assert "--es" in args # string extra
assert "--ez" in args # boolean extra
assert "--ei" in args # integer extra
@pytest.mark.usefixtures("_dev_mode")
async def test_with_flags(self, server):
server.run_shell_args.return_value = ok()
await server.activity_start(
"com.example/.Act",
flags=["FLAG_ACTIVITY_NEW_TASK"],
)
args = server.run_shell_args.call_args[0][0]
assert "-f" in args
@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
class TestBroadcastSend:
@pytest.mark.usefixtures("_dev_mode")
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"
@pytest.mark.usefixtures("_dev_mode")
async def test_with_package(self, server):
server.run_shell_args.return_value = ok()
await server.broadcast_send("ACTION", package="com.target")
args = server.run_shell_args.call_args[0][0]
assert "-p" in args
assert "com.target" in args
@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

234
tests/test_base.py Normal file
View File

@ -0,0 +1,234 @@
"""Tests for base ADB execution mixin."""
import shlex
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from src.mixins.base import ADBBaseMixin
from src.models import CommandResult
class TestRunAdb:
@pytest.fixture
def base(self):
return ADBBaseMixin()
async def test_basic_command(self, base):
mock_proc = AsyncMock()
mock_proc.communicate.return_value = (b"output\n", b"")
mock_proc.returncode = 0
patcher = patch(
"asyncio.create_subprocess_exec",
return_value=mock_proc,
)
with patcher as mock_exec:
result = await base.run_adb(["devices"])
mock_exec.assert_called_once_with(
"adb",
"devices",
stdout=-1,
stderr=-1,
)
assert result.success is True
assert result.stdout == "output"
async def test_device_targeting(self, base):
mock_proc = AsyncMock()
mock_proc.communicate.return_value = (b"ok", b"")
mock_proc.returncode = 0
patcher = patch(
"asyncio.create_subprocess_exec",
return_value=mock_proc,
)
with patcher as mock_exec:
await base.run_adb(["shell", "ls"], device_id="ABC123")
# Should insert -s ABC123 before the command
mock_exec.assert_called_once_with(
"adb",
"-s",
"ABC123",
"shell",
"ls",
stdout=-1,
stderr=-1,
)
async def test_current_device_fallback(self, base):
base.set_current_device("DEF456")
mock_proc = AsyncMock()
mock_proc.communicate.return_value = (b"ok", b"")
mock_proc.returncode = 0
patcher = patch(
"asyncio.create_subprocess_exec",
return_value=mock_proc,
)
with patcher as mock_exec:
await base.run_adb(["devices"])
args = mock_exec.call_args[0]
assert "-s" in args
assert "DEF456" in args
async def test_device_id_overrides_current(self, base):
base.set_current_device("OLD")
mock_proc = AsyncMock()
mock_proc.communicate.return_value = (b"ok", b"")
mock_proc.returncode = 0
patcher = patch(
"asyncio.create_subprocess_exec",
return_value=mock_proc,
)
with patcher as mock_exec:
await base.run_adb(["shell", "ls"], device_id="NEW")
args = mock_exec.call_args[0]
assert "NEW" in args
assert "OLD" not in args
async def test_failure(self, base):
mock_proc = AsyncMock()
mock_proc.communicate.return_value = (b"", b"not found")
mock_proc.returncode = 1
with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
result = await base.run_adb(["shell", "missing"])
assert result.success is False
assert result.stderr == "not found"
assert result.returncode == 1
async def test_timeout(self, base):
mock_proc = AsyncMock()
mock_proc.communicate.side_effect = TimeoutError()
mock_proc.kill = MagicMock()
with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
result = await base.run_adb(["shell", "hang"], timeout=1)
assert result.success is False
assert "timed out" in result.stderr
assert result.returncode == -1
async def test_exception(self, base):
with patch(
"asyncio.create_subprocess_exec",
side_effect=FileNotFoundError("adb not found"),
):
result = await base.run_adb(["devices"])
assert result.success is False
assert "adb not found" in result.stderr
class TestRunShellArgs:
@pytest.fixture
def base(self):
return ADBBaseMixin()
async def test_quotes_arguments(self, base):
mock_proc = AsyncMock()
mock_proc.communicate.return_value = (b"ok", b"")
mock_proc.returncode = 0
patcher = patch(
"asyncio.create_subprocess_exec",
return_value=mock_proc,
)
with patcher as mock_exec:
await base.run_shell_args(["input", "text", "hello world"])
args = mock_exec.call_args[0]
# "shell" should be in the args
assert "shell" in args
# Arguments should be shlex-quoted
quoted_hello = shlex.quote("hello world")
assert quoted_hello in args
async def test_injection_safety(self, base):
"""Verify dangerous characters get quoted."""
mock_proc = AsyncMock()
mock_proc.communicate.return_value = (b"ok", b"")
mock_proc.returncode = 0
patcher = patch(
"asyncio.create_subprocess_exec",
return_value=mock_proc,
)
with patcher as mock_exec:
await base.run_shell_args(["echo", "; rm -rf /"])
args = mock_exec.call_args[0]
# The dangerous string should be quoted, not bare
assert "; rm -rf /" not in args
quoted = shlex.quote("; rm -rf /")
assert quoted in args
class TestRunShell:
@pytest.fixture
def base(self):
return ADBBaseMixin()
async def test_splits_command(self, base):
mock_proc = AsyncMock()
mock_proc.communicate.return_value = (b"ok", b"")
mock_proc.returncode = 0
patcher = patch(
"asyncio.create_subprocess_exec",
return_value=mock_proc,
)
with patcher as mock_exec:
await base.run_shell("ls -la /sdcard")
args = mock_exec.call_args[0]
assert "shell" in args
assert "ls" in args
assert "-la" in args
assert "/sdcard" in args
class TestGetDeviceProperty:
@pytest.fixture
def base(self):
b = ADBBaseMixin()
b.run_shell_args = AsyncMock() # type: ignore[method-assign]
return b
async def test_returns_value(self, base):
base.run_shell_args.return_value = CommandResult(
success=True, stdout="Pixel 6", stderr="", returncode=0
)
result = await base.get_device_property("ro.product.model")
assert result == "Pixel 6"
async def test_returns_none_on_empty(self, base):
base.run_shell_args.return_value = CommandResult(
success=True, stdout="", stderr="", returncode=0
)
result = await base.get_device_property("ro.missing")
assert result is None
async def test_returns_none_on_failure(self, base):
base.run_shell_args.return_value = CommandResult(
success=False, stdout="", stderr="err", returncode=1
)
result = await base.get_device_property("ro.missing")
assert result is None
class TestDeviceState:
def test_set_get_device(self):
base = ADBBaseMixin()
assert base.get_current_device() is None
base.set_current_device("ABC")
assert base.get_current_device() == "ABC"
base.set_current_device(None)
assert base.get_current_device() is None

78
tests/test_config.py Normal file
View File

@ -0,0 +1,78 @@
"""Tests for configuration management."""
import json
from pathlib import Path
from src.config import get_config, is_developer_mode
def _fresh_config(monkeypatch, config_dir):
"""Reset singleton and point Config at a specific directory."""
config_path = Path(config_dir)
monkeypatch.setattr("src.config.Config._instance", None)
monkeypatch.setattr("src.config.CONFIG_DIR", config_path)
monkeypatch.setattr("src.config.CONFIG_FILE", config_path / "config.json")
class TestConfig:
def test_defaults(self, tmp_path, monkeypatch):
_fresh_config(monkeypatch, tmp_path)
config = get_config()
assert config.developer_mode is False
assert config.auto_select_single_device is True
assert config.default_screenshot_dir is None
def test_developer_mode_toggle(self, tmp_path, monkeypatch):
_fresh_config(monkeypatch, tmp_path)
config = get_config()
assert config.developer_mode is False
config.developer_mode = True
assert config.developer_mode is True
def test_persistence(self, tmp_path, monkeypatch):
_fresh_config(monkeypatch, tmp_path)
config = get_config()
config.developer_mode = True
config_file = tmp_path / "config.json"
assert config_file.exists()
data = json.loads(config_file.read_text())
assert data["developer_mode"] is True
def test_screenshot_dir(self, tmp_path, monkeypatch):
_fresh_config(monkeypatch, tmp_path)
config = get_config()
config.default_screenshot_dir = "/tmp/shots"
assert config.default_screenshot_dir == "/tmp/shots"
def test_get_set(self, tmp_path, monkeypatch):
_fresh_config(monkeypatch, tmp_path)
config = get_config()
config.set("custom_key", "custom_value")
assert config.get("custom_key") == "custom_value"
def test_to_dict(self, tmp_path, monkeypatch):
_fresh_config(monkeypatch, tmp_path)
config = get_config()
d = config.to_dict()
assert "developer_mode" in d
assert "auto_select_single_device" in d
def test_load_corrupt_file(self, tmp_path, monkeypatch):
_fresh_config(monkeypatch, tmp_path)
(tmp_path / "config.json").write_text("{invalid json")
# Need a fresh singleton to trigger _load with corrupt file
monkeypatch.setattr("src.config.Config._instance", None)
config = get_config()
assert config.developer_mode is False
class TestIsDeveloperMode:
def test_off_by_default(self, tmp_path, monkeypatch):
_fresh_config(monkeypatch, tmp_path)
assert is_developer_mode() is False
def test_on_when_enabled(self, tmp_path, monkeypatch):
_fresh_config(monkeypatch, tmp_path)
get_config().developer_mode = True
assert is_developer_mode() is True

122
tests/test_connectivity.py Normal file
View File

@ -0,0 +1,122 @@
"""Tests for connectivity mixin (connect, disconnect, tcpip, pair, properties)."""
import pytest
from tests.conftest import fail, ok
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"
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"
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
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
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"
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
class TestAdbTcpip:
@pytest.mark.usefixtures("_dev_mode")
async def test_success(self, server, ctx):
server.run_shell_args.return_value = ok(
stdout="10: wlan0 inet 192.168.1.100/24"
)
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"
@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
@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
@pytest.mark.usefixtures("_dev_mode")
async def test_custom_port(self, server, ctx):
server.run_shell_args.return_value = ok(
stdout="10: wlan0 inet 192.168.1.50/24"
)
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"
@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()
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
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
class TestDeviceProperties:
async def test_returns_properties(self, server):
props = {
"ro.product.model": "Pixel 6",
"ro.product.manufacturer": "Google",
"ro.build.version.release": "14",
"ro.build.version.sdk": "34",
"ro.board.platform": "gs101",
}
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"
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

182
tests/test_devices.py Normal file
View File

@ -0,0 +1,182 @@
"""Tests for devices mixin (list, use, current, info, reboot, logcat)."""
import pytest
from tests.conftest import fail, ok
class TestDevicesList:
async def test_parse_devices(self, server):
server.run_adb.return_value = ok(
stdout=(
"List of devices attached\n"
"ABC123\tdevice\tmodel:Pixel_6 product:oriole\n"
"10.20.0.25:5555\tdevice\tmodel:K2401 product:K2401\n"
)
)
devices = await server.devices_list()
assert len(devices) == 2
assert devices[0].device_id == "ABC123"
assert devices[0].model == "Pixel_6"
assert devices[1].device_id == "10.20.0.25:5555"
async def test_empty(self, server):
server.run_adb.return_value = ok(stdout="List of devices attached\n")
devices = await server.devices_list()
assert len(devices) == 0
async def test_failure(self, server):
server.run_adb.return_value = fail("adb not found")
devices = await server.devices_list()
assert len(devices) == 0
class TestDevicesUse:
async def test_select_device(self, server):
server.run_adb.return_value = ok(
stdout="List of devices attached\nABC123\tdevice\n"
)
result = await server.devices_use("ABC123")
assert result.success is True
assert server.get_current_device() == "ABC123"
async def test_not_found(self, server):
server.run_adb.return_value = ok(
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
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
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
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.available is not None
async def test_device_set(self, server):
# Pre-populate cache and set device
server.run_adb.return_value = ok(
stdout="List of devices attached\nABC123\tdevice\tmodel:Pixel\n"
)
await server.devices_list()
server.set_current_device("ABC123")
result = await server.devices_current()
# device is a dict from model_dump()
assert result.device["device_id"] == "ABC123"
class TestDeviceInfo:
async def test_full_info(self, server):
battery = (
"Current Battery Service state:\n level: 85\n status: 2\n plugged: 2"
)
df_out = (
"Filesystem 1K-blocks Used Available\n"
"/data 64000000 32000000 32000000"
)
server.run_shell_args.side_effect = [
ok(stdout=battery),
ok(stdout="10: wlan0 inet 192.168.1.100/24"),
ok(stdout="mWifiInfo SSID: MyNetwork, BSSID: ..."),
ok(stdout=df_out),
]
server.get_device_property.side_effect = lambda p, d=None: {
"ro.build.version.release": "14",
"ro.build.version.sdk": "34",
"ro.product.model": "Pixel 6",
"ro.product.manufacturer": "Google",
"ro.product.device": "oriole",
}.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"
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
class TestDeviceReboot:
@pytest.mark.usefixtures("_dev_mode")
async def test_reboot(self, server, ctx):
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"
@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"
@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.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
class TestLogcat:
@pytest.mark.usefixtures("_dev_mode")
async def test_capture(self, server):
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")
@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"
@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 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
@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"

120
tests/test_files.py Normal file
View File

@ -0,0 +1,120 @@
"""Tests for files mixin (push, pull, list, delete, exists)."""
import pytest
from tests.conftest import fail, ok
class TestFilePush:
@pytest.mark.usefixtures("_dev_mode")
async def test_push(self, server, ctx, tmp_path):
local_file = tmp_path / "test.txt"
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"
@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
@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
class TestFilePull:
@pytest.mark.usefixtures("_dev_mode")
async def test_pull(self, server, ctx, tmp_path):
server.run_adb.return_value = ok(stdout="1 file pulled")
result = await server.file_pull(
ctx, "/sdcard/test.txt", str(tmp_path / "out.txt")
)
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
@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
class TestFileList:
@pytest.mark.usefixtures("_dev_mode")
async def test_parse_ls(self, server):
server.run_shell_args.return_value = ok(
stdout=(
"total 16\n"
"drwxr-xr-x 2 root root 4096 2024-01-15 10:30 Documents\n"
"-rw-r--r-- 1 root root 1234 2024-01-15 10:31 test.txt\n"
)
)
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
@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
@pytest.mark.usefixtures("_no_dev_mode")
async def test_requires_dev_mode(self, server):
result = await server.file_list()
assert result.success is False
class TestFileDelete:
@pytest.mark.usefixtures("_dev_mode")
async def test_delete(self, server, ctx):
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"
@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.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
class TestFileExists:
@pytest.mark.usefixtures("_dev_mode")
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
@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
@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

236
tests/test_input.py Normal file
View File

@ -0,0 +1,236 @@
"""Tests for input mixin (tap, swipe, keys, text, clipboard)."""
import pytest
from tests.conftest import fail, ok
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}
server.run_shell_args.assert_called_once_with(
["input", "tap", "100", "200"], None
)
async def test_tap_with_device(self, server):
server.run_shell_args.return_value = ok()
await server.input_tap(10, 20, device_id="ABC")
server.run_shell_args.assert_called_once_with(
["input", "tap", "10", "20"], "ABC"
)
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"
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.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
class TestInputScroll:
async def test_scroll_down(self, server):
# First call: wm size, second call: the swipe
server.run_shell_args.side_effect = [
ok("Physical size: 1080x1920"),
ok(),
]
result = await server.input_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]
args = swipe_call[0][0]
assert args[0] == "input"
assert args[1] == "swipe"
assert args[2] == "540" # 1080 // 2
assert args[3] == "1248" # int(1920 * 0.65)
assert args[5] == "480" # int(1920 * 0.25)
async def test_scroll_up(self, server):
server.run_shell_args.side_effect = [
ok("Physical size: 1080x1920"),
ok(),
]
result = await server.input_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
server.run_shell_args.side_effect = [
fail("error"),
ok(),
]
result = await server.input_scroll_down()
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"
server.run_shell_args.assert_called_once_with(
["input", "keyevent", "KEYCODE_BACK"], None
)
async def test_home(self, server):
server.run_shell_args.return_value = ok()
result = await server.input_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"
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"
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"
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
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"
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"
server.run_shell_args.assert_called_once_with(["input", "text", "hello"], None)
async def test_spaces_escaped(self, server):
server.run_shell_args.return_value = ok()
await server.input_text("hello world")
server.run_shell_args.assert_called_once_with(
["input", "text", "hello%sworld"], None
)
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
async def test_rejects_semicolon_injection(self, server):
result = await server.input_text("hello; rm -rf /")
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"
async def test_cmd_clipboard_not_implemented_falls_back(self, server):
# First call: cmd clipboard returns "no shell command"
# Second call: am broadcast succeeds with result=-1
server.run_shell_args.side_effect = [
ok(stderr="No shell command implementation."),
ok(stdout="Broadcast completed: result=-1"),
]
result = await server.clipboard_set("test")
assert result.success is True
assert server.run_shell_args.call_count == 2
async def test_no_receiver_reports_failure(self, server):
server.run_shell_args.side_effect = [
ok(stderr="No shell command implementation."),
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()
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
# Verify KEYCODE_PASTE was sent
paste_call = server.run_shell_args.call_args_list[1]
assert "KEYCODE_PASTE" in paste_call[0][0]
async def test_text_preview_truncated(self, server):
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("...")
class TestInputLongPress:
@pytest.mark.usefixtures("_dev_mode")
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
# 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
assert args[3] == args[5] # y1 == y2
@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()
class TestShellCommand:
@pytest.mark.usefixtures("_dev_mode")
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"
@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()

56
tests/test_models.py Normal file
View File

@ -0,0 +1,56 @@
"""Tests for Pydantic models."""
from src.models import CommandResult, DeviceInfo, ScreenshotResult
class TestCommandResult:
def test_success(self):
r = CommandResult(success=True, stdout="ok", stderr="", returncode=0)
assert r.success is True
assert r.returncode == 0
def test_failure(self):
r = CommandResult(success=False, stdout="", stderr="err", returncode=1)
assert r.success is False
assert r.stderr == "err"
def test_defaults(self):
r = CommandResult(success=True, returncode=0)
assert r.stdout == ""
assert r.stderr == ""
def test_model_copy(self):
r = CommandResult(success=True, stdout="ok", stderr="", returncode=0)
r2 = r.model_copy(update={"success": False, "stderr": "changed"})
assert r2.success is False
assert r2.stderr == "changed"
assert r.success is True # original unchanged
class TestDeviceInfo:
def test_basic(self):
d = DeviceInfo(device_id="ABC123", status="device")
assert d.device_id == "ABC123"
assert d.model is None
def test_full(self):
d = DeviceInfo(
device_id="ABC123",
status="device",
model="Pixel_6",
product="oriole",
)
assert d.model == "Pixel_6"
dump = d.model_dump()
assert dump["product"] == "oriole"
class TestScreenshotResult:
def test_success(self):
r = ScreenshotResult(success=True, local_path="/tmp/shot.png")
assert r.local_path == "/tmp/shot.png"
def test_failure(self):
r = ScreenshotResult(success=False, error="No device")
assert r.error == "No device"
assert r.local_path is None

132
tests/test_screenshot.py Normal file
View File

@ -0,0 +1,132 @@
"""Tests for screenshot mixin (capture, screen size, density, record)."""
import pytest
from tests.conftest import fail, ok
class TestScreenshot:
async def test_capture(self, server, ctx, tmp_path, monkeypatch):
from src.config import get_config
get_config().default_screenshot_dir = str(tmp_path)
server.run_shell_args.return_value = ok()
server.run_adb.return_value = ok()
result = await server.screenshot(ctx, filename="test.png")
assert result.success is True
assert result.local_path is not None
assert "test.png" in result.local_path
async def test_capture_failure(self, server, ctx):
server.run_shell_args.return_value = fail("no screen")
result = await server.screenshot(ctx)
assert result.success is False
assert result.error is not None
async def test_pull_failure(self, server, ctx):
server.run_shell_args.return_value = ok()
server.run_adb.return_value = fail("pull failed")
result = await server.screenshot(ctx)
assert result.success is False
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
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
# Should parse the first match
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
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
async def test_failure(self, server):
server.run_shell_args.return_value = fail("error")
result = await server.screen_density()
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"
async def test_screen_off(self, server):
server.run_shell_args.return_value = ok()
result = await server.screen_off()
assert result.action == "screen_off"
class TestScreenRecord:
@pytest.mark.usefixtures("_dev_mode")
async def test_record(self, server, ctx, tmp_path):
from src.config import get_config
get_config().default_screenshot_dir = str(tmp_path)
server.run_shell_args.side_effect = [ok(), ok()] # record + rm
server.run_adb.return_value = ok() # pull
result = await server.screen_record(
ctx,
filename="test.mp4",
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):
from src.config import get_config
get_config().default_screenshot_dir = str(tmp_path)
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
@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
class TestScreenSetSize:
@pytest.mark.usefixtures("_dev_mode")
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
@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"
@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

36
tests/test_server.py Normal file
View File

@ -0,0 +1,36 @@
"""Tests for server-level tools (config, help resource)."""
class TestConfigStatus:
async def test_status(self, server):
result = await server.config_status()
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"
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
async def test_disable(self, server):
result = await server.config_set_developer_mode(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"
async def test_clear(self, server):
result = await server.config_set_screenshot_dir(None)
assert result.screenshot_dir is None

391
tests/test_settings.py Normal file
View File

@ -0,0 +1,391 @@
"""Tests for settings mixin (settings, toggles, notifications, clipboard, media)."""
import pytest
from src.mixins.settings import _MEDIA_KEYCODES, SettingsMixin
from tests.conftest import fail, ok
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
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
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
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
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
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
class TestSettingsPut:
@pytest.mark.usefixtures("_dev_mode")
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
@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
@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
@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
# Verify elicitation happened
assert any("secure" in msg for _, msg in ctx.messages)
@pytest.mark.usefixtures("_dev_mode")
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.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
class TestWifiToggle:
@pytest.mark.usefixtures("_dev_mode")
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
@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
@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
class TestBluetoothToggle:
@pytest.mark.usefixtures("_dev_mode")
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"
@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
class TestAirplaneModeToggle:
@pytest.mark.usefixtures("_dev_mode")
async def test_enable_usb_device(self, server, ctx):
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
@pytest.mark.usefixtures("_dev_mode")
async def test_enable_network_device_warns(self, server, ctx):
server.set_current_device("10.20.0.25:5555")
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
# Should have warned about network disconnection
warns = [msg for level, msg in ctx.messages if "sever" in msg.lower()]
assert len(warns) > 0
@pytest.mark.usefixtures("_dev_mode")
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.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
# No elicitation for disable
elicits = [m for level, m in ctx.messages if level == "elicit"]
assert len(elicits) == 0
@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
class TestScreenBrightness:
@pytest.mark.usefixtures("_dev_mode")
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
@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
@pytest.mark.usefixtures("_dev_mode")
async def test_negative(self, server):
result = await server.screen_brightness(-1)
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
class TestScreenTimeout:
@pytest.mark.usefixtures("_dev_mode")
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
@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
@pytest.mark.usefixtures("_dev_mode")
async def test_zero(self, server):
result = await server.screen_timeout(0)
assert result.success is False
class TestNotificationList:
async def test_parse_notifications(self, server):
dumpsys_output = """
NotificationRecord(0x1234 pkg=com.example.app)
extras {
android.title=Test Title
android.text=Test message body
}
postTime=1700000000000
NotificationRecord(0x5678 pkg=com.other.app)
extras {
android.title=Second
android.text=Another notification
}
postTime=1700000001000
"""
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"
async def test_limit(self, server):
# Build output with many notifications
lines = []
for i in range(10):
lines.append(f" NotificationRecord(0x{i:04x} pkg=com.app{i})")
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
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
class TestClipboardGet:
async def test_parses_parcel(self, server):
# 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"
async def test_empty_clipboard(self, server):
server.run_shell_args.return_value = ok(
stdout="Result: Parcel(00000000 00000000 '........')"
)
result = await server.clipboard_get()
# No text/plain marker = not parseable
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
def _build_parcel(text: str) -> str:
"""Build a fake service call parcel output containing clipboard text.
Mimics the format of `service call clipboard 4` output with a
ClipData Parcel containing text/plain MIME type and UTF-8 text.
"""
import struct
parts = []
# Status word (0 = success)
parts.append(struct.pack("<I", 0))
# Non-null marker
parts.append(struct.pack("<I", 1))
# ClipDescription: label length = 0 (no label)
parts.append(struct.pack("<I", 0))
# MIME type count = 1
parts.append(struct.pack("<I", 1))
# MIME type "text/plain" in UTF-16LE with length prefix
mime = "text/plain"
mime_utf16 = mime.encode("utf-16-le")
parts.append(struct.pack("<I", len(mime)))
parts.append(mime_utf16)
# Pad to 4-byte boundary
pad = (4 - len(mime_utf16) % 4) % 4
parts.append(b"\x00" * pad)
# Extras (none), timestamps, flags
parts.append(struct.pack("<I", 0xFFFFFFFF)) # no extras
parts.append(struct.pack("<I", 0xFFFFFFFF)) # no extras
parts.append(struct.pack("<I", 0)) # flags
# Item count = 1
parts.append(struct.pack("<I", 1))
# CharSequence type marker
parts.append(struct.pack("<I", 0))
# Text as length-prefixed UTF-8
text_bytes = text.encode("utf-8")
parts.append(struct.pack("<I", len(text_bytes)))
parts.append(text_bytes)
pad = (4 - len(text_bytes) % 4) % 4
parts.append(b"\x00" * pad)
data = b"".join(parts)
# Format as hex words like real parcel output
hex_lines = []
for i in range(0, len(data), 16):
chunk = data[i : i + 16]
words = []
for j in range(0, len(chunk), 4):
word = chunk[j : j + 4].ljust(4, b"\x00")
words.append(int.from_bytes(word, "little"))
hex_str = " ".join(f"{w:08x}" for w in words)
hex_lines.append(f" 0x{i:08x}: {hex_str} '...'")
return "Result: Parcel(\n" + "\n".join(hex_lines) + "\n)"
class TestParseClipboardParcel:
"""Direct tests for the static parcel parser."""
def test_valid_parcel(self):
raw = _build_parcel("test data here")
result = SettingsMixin._parse_clipboard_parcel(raw)
assert result == "test data here"
def test_nonzero_status(self):
raw = "Result: Parcel(00000001 '....')"
result = SettingsMixin._parse_clipboard_parcel(raw)
assert result is None
def test_no_hex_words(self):
result = SettingsMixin._parse_clipboard_parcel("no hex here")
assert result is None
def test_no_mime_marker(self):
raw = "Result: Parcel(00000000 00000001 '........')"
result = SettingsMixin._parse_clipboard_parcel(raw)
assert result is None
def test_long_text(self):
long_text = "The quick brown fox jumps over the lazy dog. " * 10
raw = _build_parcel(long_text)
result = SettingsMixin._parse_clipboard_parcel(raw)
assert result == long_text
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"
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
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"
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
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"

141
tests/test_ui.py Normal file
View File

@ -0,0 +1,141 @@
"""Tests for UI inspection mixin (dump, find, wait, tap_text)."""
from tests.conftest import fail, ok
SAMPLE_UI_XML = """<?xml version="1.0" encoding="UTF-8"?>
<hierarchy>
<node text="Settings" class="android.widget.TextView"
resource-id="com.android.settings:id/title"
bounds="[0,100][200,150]" clickable="true" focusable="true"
content-desc="" />
<node text="" class="android.widget.ImageView"
resource-id="com.android.settings:id/icon"
bounds="[0,50][48,98]" clickable="false" focusable="false"
content-desc="Settings icon" />
<node text="Wi-Fi" class="android.widget.TextView"
resource-id="com.android.settings:id/title"
bounds="[200,100][400,150]" clickable="true" focusable="false"
content-desc="" />
</hierarchy>
"""
class TestUiDump:
async def test_dump(self, server, ctx):
server.run_shell_args.side_effect = [
ok(), # uiautomator dump
ok(stdout=SAMPLE_UI_XML), # cat
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 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
class TestParseUiElements:
def test_parse_clickable(self, server):
elements = server._parse_ui_elements(SAMPLE_UI_XML)
texts = [e["text"] for e in elements]
assert "Settings" in texts
assert "Wi-Fi" in texts
def test_center_coordinates(self, server):
elements = server._parse_ui_elements(SAMPLE_UI_XML)
settings = [e for e in elements if e["text"] == "Settings"][0]
assert settings["center"] == {"x": 100, "y": 125}
def test_content_desc_included(self, server):
elements = server._parse_ui_elements(SAMPLE_UI_XML)
icon = [e for e in elements if e["content_desc"] == "Settings icon"]
assert len(icon) == 1
def test_empty_xml(self, server):
elements = server._parse_ui_elements("")
assert elements == []
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"
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
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
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
async def test_timeout(self, server):
server.run_shell_args.side_effect = [
ok(),
ok(stdout="<hierarchy></hierarchy>"),
ok(),
] * 10
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
class TestWaitForTextGone:
async def test_already_gone(self, server):
server.run_shell_args.side_effect = [
ok(),
ok(stdout="<hierarchy></hierarchy>"),
ok(),
]
result = await server.wait_for_text_gone("Missing", timeout_seconds=1)
assert result.success is True
assert result.gone is True
class TestTapText:
async def test_tap_found(self, server):
# find_element calls ui_dump which has 3 calls, then tap has 1
server.run_shell_args.side_effect = [
ok(),
ok(stdout=SAMPLE_UI_XML),
ok(), # ui_dump for find
ok(), # tap
]
result = await server.tap_text("Settings")
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 = [
ok(),
ok(stdout=SAMPLE_UI_XML),
ok(), # first search by text
ok(),
ok(stdout=SAMPLE_UI_XML),
ok(), # fallback search by content_desc
]
result = await server.tap_text("NonExistent")
assert result.success is False
assert "No element found" in result.error