agentic-mobile-control/tests/test_android_driver.py

"""Mocked unit tests for ``AndroidDriver`` and its registry factory.

Coverage note: ``WDADriver`` has no mocked unit tests today (only the
hardware-gated ``tests/test_wda_integration.py``). These Android tests are a
deliberate quality-bar increase for the new driver, not parity work — there is
no equivalent WDA coverage to mirror here.
"""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest

from core.errors import DeviceOfflineError, DriverError
from driver.android_driver import AndroidDriver, AndroidDriverConfig
from driver.registry import build_android_driver_factory


def _connected_driver(config: AndroidDriverConfig | None = None) -> AndroidDriver:
    """Return an ``AndroidDriver`` whose ``_client`` is a MagicMock.

    Bypasses ``connect()`` so operation tests don't touch the Appium import
    surface or network layer.
    """
    driver = AndroidDriver(config)
    driver._client = MagicMock()
    return driver


# --------------------------------------------------------------------------- #
# connect()
# --------------------------------------------------------------------------- #


def test_connect_builds_client_with_expected_capabilities() -> None:
    config = AndroidDriverConfig(
        server_url="http://android-host:4723",
        device_name="pixel-7",
        udid="serial-abc",
        system_port=8201,
        extra_capabilities={"appPackage": "com.example"},
    )

    with patch("appium.webdriver.Remote") as mock_remote:
        mock_remote.return_value = MagicMock(name="appium-client")
        driver = AndroidDriver(config)
        driver.connect()

    assert mock_remote.called
    kwargs = mock_remote.call_args.kwargs
    caps = kwargs["options"].to_capabilities()
    assert caps["platformName"] == "Android"
    assert caps["appium:automationName"] == "UiAutomator2"
    assert caps["appium:noReset"] is True
    assert caps["appium:deviceName"] == "pixel-7"
    assert caps["appium:udid"] == "serial-abc"
    assert caps["appium:systemPort"] == 8201
    assert caps["appium:appPackage"] == "com.example"
    assert kwargs["client_config"].remote_server_addr == "http://android-host:4723"
    assert driver._client is mock_remote.return_value


def test_connect_omits_unset_optional_capabilities() -> None:
    with patch("appium.webdriver.Remote") as mock_remote:
        mock_remote.return_value = MagicMock()
        AndroidDriver(AndroidDriverConfig()).connect()

    caps = mock_remote.call_args.kwargs["options"].to_capabilities()
    # Only the always-present caps should appear; device/udid/systemPort are
    # all unset on the default config.
    assert "appium:deviceName" not in caps
    assert "appium:udid" not in caps
    assert "appium:systemPort" not in caps


def test_connect_failure_raises_device_offline_and_clears_client() -> None:
    with patch("appium.webdriver.Remote") as mock_remote:
        mock_remote.side_effect = ConnectionError("appium server unreachable")
        driver = AndroidDriver(AndroidDriverConfig())
        with pytest.raises(DeviceOfflineError):
            driver.connect()

    assert driver._client is None


# --------------------------------------------------------------------------- #
# Pre-connect guard
# --------------------------------------------------------------------------- #


@pytest.mark.parametrize(
    "method,kwargs",
    [
        ("screenshot", {}),
        ("tap", {"x": 10, "y": 20}),
        ("swipe", {"start_x": 0, "start_y": 0, "end_x": 5, "end_y": 5}),
        ("input", {"text": "hi"}),
        ("launch", {"app_id": "com.example"}),
        ("terminate", {"app_id": "com.example"}),
        ("tree", {}),
        ("home", {}),
        ("lock", {}),
        ("unlock", {}),
        ("disconnect", {}),
        ("long_press", {"x": 1, "y": 2}),
        ("swipe_path", {"waypoints": [(0, 0), (1, 1)], "duration_ms": 100}),
        ("double_tap", {"x": 1, "y": 2}),
    ],
)
def test_operation_before_connect_raises_device_offline(
    method: str, kwargs: dict
) -> None:
    driver = AndroidDriver()
    with pytest.raises(DeviceOfflineError):
        getattr(driver, method)(**kwargs)


# --------------------------------------------------------------------------- #
# Per-operation error wrapping
# --------------------------------------------------------------------------- #


def test_screenshot_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.get_screenshot_as_png.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.screenshot()


def test_tap_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.execute_script.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.tap(1, 2)


def test_swipe_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.execute_script.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.swipe(0, 0, 10, 10)


def test_input_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.switch_to.active_element.send_keys.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.input("text")


def test_launch_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.activate_app.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.launch("com.example")


def test_terminate_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.terminate_app.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.terminate("com.example")


def test_tree_wraps_exception_into_driver_error() -> None:
    # ``page_source`` is a property on the Selenium client; attribute access on
    # a MagicMock returns a child mock (never raises), so use a minimal stand-in
    # whose property raises to exercise the error-wrapping path.
    class _FailingPageSource:
        @property
        def page_source(self) -> str:
            raise RuntimeError("boom")

    driver = AndroidDriver()
    driver._client = _FailingPageSource()
    with pytest.raises(DriverError):
        driver.tree()


def test_home_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.execute_script.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.home()


def test_lock_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.lock.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.lock()


def test_unlock_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.unlock.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.unlock()


def test_disconnect_wraps_exception_and_clears_client() -> None:
    driver = _connected_driver()
    driver._client.quit.side_effect = RuntimeError("boom")
    # disconnect() follows WDADriver's pattern: quit() error propagates only
    # after the finally block clears _client. WDADriver does not wrap quit()
    # failures into DriverError, so Android mirrors that exactly.
    with pytest.raises(RuntimeError):
        driver.disconnect()
    assert driver._client is None


# --------------------------------------------------------------------------- #
# tap / swipe / home command shape
# --------------------------------------------------------------------------- #


def test_tap_uses_click_gesture() -> None:
    driver = _connected_driver()
    driver.tap(123, 456)
    driver._client.execute_script.assert_called_once_with(
        "mobile: clickGesture", {"x": 123, "y": 456}
    )


def test_home_uses_presskey_with_home_keycode() -> None:
    driver = _connected_driver()
    driver.home()
    driver._client.execute_script.assert_called_once_with(
        "mobile: pressKey", {"keycode": 3}
    )


def test_swipe_path_sends_w3c_actions() -> None:
    from driver._w3c_actions import build_swipe_actions

    driver = _connected_driver()
    waypoints = [(0, 0), (50, 5), (100, 0)]
    driver.swipe_path(waypoints, 600)
    args = driver._client.execute.call_args.args
    assert args[1] == {"actions": build_swipe_actions(waypoints, 600)}


def test_swipe_path_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.execute.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.swipe_path([(0, 0), (1, 1)], 100)


def test_double_tap_sends_w3c_actions() -> None:
    from driver._w3c_actions import build_double_tap_actions

    driver = _connected_driver()
    driver.double_tap(15, 25, interval_ms=90)
    args = driver._client.execute.call_args.args
    assert args[1] == {"actions": build_double_tap_actions(15, 25, 90)}


def test_double_tap_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.execute.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.double_tap(1, 2)


def test_swipe_uses_drag_gesture_with_converted_speed() -> None:
    driver = _connected_driver()
    # 100px horizontal drag over 100ms => 100 / 0.1 = 1000 px/s
    driver.swipe(0, 0, 100, 0, duration_ms=100)
    driver._client.execute_script.assert_called_once_with(
        "mobile: dragGesture",
        {"startX": 0, "startY": 0, "endX": 100, "endY": 0, "speed": 1000},
    )


def test_swipe_zero_distance_uses_default_speed_without_div_by_zero() -> None:
    driver = _connected_driver()
    # start == end => zero distance; must not divide by zero.
    driver.swipe(50, 50, 50, 50, duration_ms=500)
    args = driver._client.execute_script.call_args.args[1]
    assert args["speed"] > 0
    assert args["speed"] == args["speed"]  # sanity: it's a positive int


def test_swipe_near_zero_distance_uses_default_speed() -> None:
    driver = _connected_driver()
    # 0.5px distance over 500ms would be 1 px/s but distance < 1.0 guard fires.
    driver.swipe(0, 0, 0, 0, duration_ms=500)
    speed = driver._client.execute_script.call_args.args[1]["speed"]
    assert speed > 0


def test_long_press_uses_long_click_gesture() -> None:
    driver = _connected_driver()
    driver.long_press(30, 40, duration_ms=1500)
    driver._client.execute_script.assert_called_once_with(
        "mobile: longClickGesture", {"x": 30, "y": 40}
    )


def test_long_press_wraps_exception_into_driver_error() -> None:
    driver = _connected_driver()
    driver._client.execute_script.side_effect = RuntimeError("boom")
    with pytest.raises(DriverError):
        driver.long_press(1, 2)


# --------------------------------------------------------------------------- #
# build_android_driver_factory
# --------------------------------------------------------------------------- #


def test_build_android_driver_factory_extracts_declared_fields() -> None:
    connection_info = {
        "server_url": "http://host:4723",
        "udid": "device-serial",
        "system_port": 8202,
        "extra_capabilities": {"appPackage": "com.example.app"},
    }
    factory = build_android_driver_factory(connection_info)
    driver = factory()
    assert isinstance(driver, AndroidDriver)
    assert driver.config.server_url == "http://host:4723"
    assert driver.config.udid == "device-serial"
    assert driver.config.system_port == 8202
    assert driver.config.extra_capabilities == {"appPackage": "com.example.app"}


def test_build_android_driver_factory_routes_unknown_keys_to_extra() -> None:
    connection_info = {
        "server_url": "http://host:4723",
        "appWaitActivity": "MainActivity",  # not a declared config field
        "autoGrantPermissions": True,  # not a declared config field
    }
    factory = build_android_driver_factory(connection_info)
    driver = factory()
    assert driver.config.extra_capabilities == {
        "appWaitActivity": "MainActivity",
        "autoGrantPermissions": True,
    }


def test_build_android_driver_factory_merges_inline_and_explicit_extras() -> None:
    connection_info = {
        "udid": "serial",
        "extra_capabilities": {"a": 1},
        "b": 2,  # inline extra
    }
    driver = build_android_driver_factory(connection_info)()
    assert driver.config.udid == "serial"
    assert driver.config.extra_capabilities == {"a": 1, "b": 2}


def test_build_android_driver_factory_rejects_non_dict_extras() -> None:
    with pytest.raises(ValueError, match="extra_capabilities"):
        build_android_driver_factory({"extra_capabilities": "not-a-dict"})