showtan001
8c99dc015a
Tests / Test apps.device-host-agent.tests.test_mcp_token.test_load_or_create_concurrent_calls_do_not_corrupt failed
100 lines
2.8 KiB
Python
100 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any
|
|
|
|
from core.models import ActiveApp
|
|
|
|
|
|
class Driver(ABC):
|
|
"""Driver-independent device capability interface.
|
|
|
|
Implementations may hold a live connection handle, but no task or business
|
|
state belongs here.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def connect(self) -> None:
|
|
"""Open the device connection."""
|
|
|
|
@abstractmethod
|
|
def disconnect(self) -> None:
|
|
"""Close the device connection."""
|
|
|
|
@abstractmethod
|
|
def screenshot(self) -> bytes:
|
|
"""Return the current screen as image bytes."""
|
|
|
|
def health_check(self) -> None:
|
|
"""Verify the live session without reading the device screen.
|
|
|
|
Drivers with a transport-level status endpoint should override this
|
|
method. The default is a no-op for legacy drivers that do not expose
|
|
a separate health check.
|
|
"""
|
|
return None
|
|
|
|
@abstractmethod
|
|
def tap(self, x: float, y: float) -> None:
|
|
"""Tap the screen at the given coordinates."""
|
|
|
|
@abstractmethod
|
|
def long_press(self, x: float, y: float, duration_ms: int = 1200) -> None:
|
|
"""Press and hold at the given coordinates for ``duration_ms``.
|
|
|
|
Some platforms ignore ``duration_ms`` (fixed system hold duration).
|
|
"""
|
|
|
|
@abstractmethod
|
|
def swipe(
|
|
self,
|
|
start_x: float,
|
|
start_y: float,
|
|
end_x: float,
|
|
end_y: float,
|
|
duration_ms: int = 500,
|
|
) -> None:
|
|
"""Swipe between two screen coordinates."""
|
|
|
|
@abstractmethod
|
|
def swipe_path(
|
|
self, waypoints: list[tuple[float, float]], duration_ms: int
|
|
) -> None:
|
|
"""Swipe through a sequence of waypoints in one continuous touch."""
|
|
|
|
@abstractmethod
|
|
def double_tap(self, x: float, y: float, interval_ms: int = 80) -> None:
|
|
"""Tap twice at the given coordinates with ``interval_ms`` between."""
|
|
|
|
@abstractmethod
|
|
def input(self, text: str) -> None:
|
|
"""Input text into the current focused field."""
|
|
|
|
@abstractmethod
|
|
def launch(self, app_id: str) -> None:
|
|
"""Launch an app by bundle id or driver-supported app identifier."""
|
|
|
|
@abstractmethod
|
|
def terminate(self, app_id: str) -> None:
|
|
"""Terminate an app by bundle id or driver-supported app identifier."""
|
|
|
|
@abstractmethod
|
|
def tree(self) -> Any:
|
|
"""Return the raw UI tree from the device driver."""
|
|
|
|
def active_app(self) -> ActiveApp | None:
|
|
"""Return foreground application metadata when the driver supports it."""
|
|
return None
|
|
|
|
@abstractmethod
|
|
def home(self) -> None:
|
|
"""Press the device home button."""
|
|
|
|
@abstractmethod
|
|
def lock(self) -> None:
|
|
"""Lock the device."""
|
|
|
|
@abstractmethod
|
|
def unlock(self) -> None:
|
|
"""Unlock the device."""
|