showtan001
8c99dc015a
Tests / Test apps.device-host-agent.tests.test_mcp_token.test_load_or_create_concurrent_calls_do_not_corrupt failed
241 lines
8.3 KiB
Python
241 lines
8.3 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from core.errors import DeviceOfflineError, DriverError
|
|
from core.models import ActiveApp
|
|
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 health_check(self) -> None:
|
|
client = self._require_client()
|
|
try:
|
|
client.get_status()
|
|
except Exception as exc:
|
|
raise DriverError("health check 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 long_press(self, x: float, y: float, duration_ms: int = 1200) -> None:
|
|
# Android ``mobile: longClickGesture`` uses a system-fixed hold
|
|
# duration; the caller's ``duration_ms`` is intentionally ignored
|
|
# (see design R2). Verify on real Appium against the running device
|
|
# driver — see plan task 2.
|
|
client = self._require_client()
|
|
try:
|
|
client.execute_script("mobile: longClickGesture", {"x": x, "y": y})
|
|
except Exception as exc:
|
|
raise DriverError("long press failed") from exc
|
|
|
|
def double_tap(self, x: float, y: float, interval_ms: int = 80) -> None:
|
|
from driver._w3c_actions import build_double_tap_actions, perform_actions
|
|
|
|
client = self._require_client()
|
|
try:
|
|
perform_actions(client, build_double_tap_actions(x, y, interval_ms))
|
|
except Exception as exc:
|
|
raise DriverError("double 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 swipe_path(
|
|
self, waypoints: list[tuple[float, float]], duration_ms: int
|
|
) -> None:
|
|
from driver._w3c_actions import build_swipe_actions, perform_actions
|
|
|
|
client = self._require_client()
|
|
try:
|
|
perform_actions(client, build_swipe_actions(waypoints, duration_ms))
|
|
except Exception as exc:
|
|
raise DriverError("swipe_path 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 active_app(self) -> ActiveApp | None:
|
|
client = self._require_client()
|
|
try:
|
|
package = client.current_package
|
|
activity = client.current_activity
|
|
except Exception as exc:
|
|
raise DriverError("active app retrieval failed") from exc
|
|
package = package if isinstance(package, str) and package else None
|
|
activity = activity if isinstance(activity, str) and activity else None
|
|
if package is None and activity is None:
|
|
return None
|
|
return ActiveApp(platform="android", package=package, activity=activity)
|
|
|
|
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)
|