feat(driver): add Android UiAutomator2 driver
Tests / Test No test results found

Register `SUPPORTED_DRIVER_TYPES["uiautomator2"]` backed by the new
`AndroidDriver`, mirroring `WDADriver` method-for-method. tap/swipe/home use
the confirmed UiAutomator2 mobile commands (`clickGesture`, `dragGesture`,
`pressKey`); swipe converts `duration_ms` to a drag speed with a zero-distance
guard. Adds 34 mocked unit tests, an env-var-gated integration test, and
corrects the outdated "Android not registered" note in the iPhone setup doc.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-07-13 14:47:01 +08:00
co-authored by Claude Opus 4.6
parent 2ccbc63d95
commit 50f0b8ade9
6 changed files with 561 additions and 18 deletions
+188
View File
@@ -0,0 +1,188 @@
from __future__ import annotations
import math
from dataclasses import dataclass, field
from typing import Any
from core.errors import DeviceOfflineError, DriverError
from driver.base import Driver
# Android KeyEvent.KEYCODE_HOME. Kept as a literal rather than importing the
# full keycode table because this is the only keycode the Driver ABC exposes.
_KEYCODE_HOME = 3
# Default drag speed (pixels/second) used by ``swipe`` when the caller's start
# and end coordinates collapse to a zero/near-zero distance, where converting
# ``duration_ms`` via ``distance / seconds`` would divide by zero.
_DEFAULT_DRAG_SPEED_PX_PER_SEC = 2500
@dataclass(frozen=True)
class AndroidDriverConfig:
server_url: str = "http://127.0.0.1:4723"
platform_name: str = "Android"
automation_name: str = "UiAutomator2"
device_name: str | None = None
udid: str | None = None
system_port: int | None = None
no_reset: bool = True
extra_capabilities: dict[str, Any] = field(default_factory=dict)
class AndroidDriver(Driver):
def __init__(self, config: AndroidDriverConfig | None = None) -> None:
self.config = config or AndroidDriverConfig()
self._client: Any | None = None
def connect(self) -> None:
try:
from appium import webdriver
from appium.options.android import UiAutomator2Options
from appium.webdriver.client_config import AppiumClientConfig
except ImportError as exc:
raise DriverError("Appium Python client is not installed") from exc
capabilities: dict[str, Any] = {
"platformName": self.config.platform_name,
"automationName": self.config.automation_name,
"noReset": self.config.no_reset,
**self.config.extra_capabilities,
}
if self.config.device_name:
capabilities["deviceName"] = self.config.device_name
if self.config.udid:
capabilities["udid"] = self.config.udid
if self.config.system_port:
capabilities["systemPort"] = self.config.system_port
options = UiAutomator2Options().load_capabilities(capabilities)
client_config = AppiumClientConfig(remote_server_addr=self.config.server_url)
try:
self._client = webdriver.Remote(
options=options,
client_config=client_config,
)
except Exception as exc:
self._client = None
raise DeviceOfflineError("device offline") from exc
def disconnect(self) -> None:
client = self._require_client()
try:
client.quit()
finally:
self._client = None
def screenshot(self) -> bytes:
client = self._require_client()
try:
return client.get_screenshot_as_png()
except Exception as exc:
raise DriverError("screenshot failed") from exc
def tap(self, x: float, y: float) -> None:
client = self._require_client()
try:
client.execute_script("mobile: clickGesture", {"x": x, "y": y})
except Exception as exc:
raise DriverError("tap failed") from exc
def swipe(
self,
start_x: float,
start_y: float,
end_x: float,
end_y: float,
duration_ms: int = 500,
) -> None:
client = self._require_client()
speed = _drag_speed(start_x, start_y, end_x, end_y, duration_ms)
try:
client.execute_script(
"mobile: dragGesture",
{
"startX": start_x,
"startY": start_y,
"endX": end_x,
"endY": end_y,
"speed": speed,
},
)
except Exception as exc:
raise DriverError("swipe failed") from exc
def input(self, text: str) -> None:
client = self._require_client()
try:
client.switch_to.active_element.send_keys(text)
except Exception as exc:
raise DriverError("text input failed") from exc
def launch(self, app_id: str) -> None:
client = self._require_client()
try:
client.activate_app(app_id)
except Exception as exc:
raise DriverError("app launch failed") from exc
def terminate(self, app_id: str) -> None:
client = self._require_client()
try:
client.terminate_app(app_id)
except Exception as exc:
raise DriverError("app terminate failed") from exc
def tree(self) -> str:
client = self._require_client()
try:
return client.page_source
except Exception as exc:
raise DriverError("ui tree retrieval failed") from exc
def home(self) -> None:
client = self._require_client()
try:
client.execute_script("mobile: pressKey", {"keycode": _KEYCODE_HOME})
except Exception as exc:
raise DriverError("home failed") from exc
def lock(self) -> None:
client = self._require_client()
try:
client.lock()
except Exception as exc:
raise DriverError("lock failed") from exc
def unlock(self) -> None:
client = self._require_client()
try:
client.unlock()
except Exception as exc:
raise DriverError("unlock failed") from exc
def _require_client(self) -> Any:
if self._client is None:
raise DeviceOfflineError("device offline")
return self._client
def _drag_speed(
start_x: float,
start_y: float,
end_x: float,
end_y: float,
duration_ms: int,
) -> int:
"""Convert ``Driver.swipe``'s ``duration_ms`` into a drag speed (px/s).
``mobile: dragGesture`` takes a speed in pixels/second rather than a
duration. Convert via ``distance / seconds`` and guard against a
zero/near-zero distance that would otherwise divide by zero.
"""
distance = math.hypot(end_x - start_x, end_y - start_y)
seconds = duration_ms / 1000
if seconds <= 0:
return _DEFAULT_DRAG_SPEED_PX_PER_SEC
if distance < 1.0:
return _DEFAULT_DRAG_SPEED_PX_PER_SEC
return int(distance / seconds)
+21
View File
@@ -5,6 +5,7 @@ from dataclasses import fields
from typing import Any
from device.manager import DriverFactory
from driver.android_driver import AndroidDriver, AndroidDriverConfig
from driver.wda_driver import WDADriver, WDADriverConfig
DriverFactoryBuilder = Callable[[dict[str, Any]], DriverFactory]
@@ -29,8 +30,28 @@ def build_wda_driver_factory(connection_info: dict[str, Any]) -> DriverFactory:
return lambda: WDADriver(config)
def build_android_driver_factory(connection_info: dict[str, Any]) -> DriverFactory:
config_fields = {field.name for field in fields(AndroidDriverConfig)}
data = dict(connection_info)
raw_extra_capabilities = data.pop("extra_capabilities", {})
if not isinstance(raw_extra_capabilities, dict):
raise ValueError("extra_capabilities must be an object")
config_values: dict[str, Any] = {}
for key in list(data):
if key in config_fields and key != "extra_capabilities":
config_values[key] = data.pop(key)
config = AndroidDriverConfig(
**config_values,
extra_capabilities={**raw_extra_capabilities, **data},
)
return lambda: AndroidDriver(config)
SUPPORTED_DRIVER_TYPES: dict[str, DriverFactoryBuilder] = {
"wda": build_wda_driver_factory,
"uiautomator2": build_android_driver_factory,
}