Artemis-Eapil/tests/unit/mcp/test_native_observation_selectors.py

"""Model-visible selectors must round-trip to exact native matching safely."""

from copy import deepcopy
import json
from unittest.mock import AsyncMock

import pytest

from artemis.mcp.native_actions import ControlSelector
from artemis.mcp.native_executor import execute_native
from artemis.utils.ui_filter import filter_ui_hierarchy
from artemis.utils.visualization import format_minimal_list_with_elements
from tests.unit.utils.test_native_control_state import encode_node, hierarchy_node


def native_node(raw):
    return {
        "resource_id": raw.get("resource-id", ""),
        "class_name": raw["class"],
        "package_name": raw["package"],
        "bounds": [30, 549, 110, 629],
        "text": "17:42:20",
        "content_desc": "current 17:42:20",
        "visible": True,
        "actions": ["set_selection"],
    }


def exact_native(nodes):
    async def inspect(command, args, *, timeout=None):
        assert command == "inspect_control", "A failed read must never dispatch input"
        matches = [
            node
            for node in nodes
            if all(node.get(key) == value for key, value in args["selector"].items())
        ]
        if len(matches) == 1:
            return {"success": True, "execution": "observed", "code": "OK", "node": matches[0]}
        return {
            "success": False,
            "execution": "not_started",
            "code": "NOT_FOUND" if not matches else "AMBIGUOUS",
            "error": "Exact native selector did not resolve uniquely",
        }

    return AsyncMock(side_effect=inspect)


@pytest.mark.parametrize("encoding", ["xml", "json"])
@pytest.mark.parametrize("screen", [(720, 1600), (1080, 2400), (1600, 720)])
@pytest.mark.asyncio
async def test_copy_visible_selector_reads_a_changing_value_without_lossy_ui_bounds(
    encoding, screen
):
    raw = hierarchy_node(
        text="17:42:17",
        **{
            "class": "android.widget.TextView",
            "content-desc": "current 17:42:17",
            "bounds": "[30,549][110,629]",
            "range": None,
            "actions": ["set_selection"],
        },
    )
    source = encode_node(raw, encoding)
    original = deepcopy(source)
    text, elements, _ = format_minimal_list_with_elements([source], *screen)
    selector = json.loads(text.split(" | native_selector: ", 1)[1])
    assert (
        selector
        == elements[0]["native_selector"]
        == {
            "resource_id": raw["resource-id"],
            "class_name": raw["class"],
            "package_name": raw["package"],
        }
    )
    assert source == original
    ControlSelector.model_validate(selector)
    native = exact_native([native_node(raw)])
    result = await execute_native("read_control", {"selector": selector}, *screen, native)
    assert result["success"] and result["node"]["text"] == "17:42:20"
    native.assert_awaited_once()


@pytest.mark.parametrize("resource_id", ["", "fixture:id/shared"])
@pytest.mark.asyncio
async def test_no_id_and_repeated_ids_use_original_pixels_to_resolve_each_node(resource_id):
    rows = [
        hierarchy_node(
            text=f"row {i}",
            **{
                "resource-id": resource_id,
                "bounds": f"[30,{549 + i * 100}][110,{629 + i * 100}]",
            },
        )
        for i in range(2)
    ]
    _, elements, _ = format_minimal_list_with_elements(rows, 720, 1600)
    nodes = [
        native_node(row) | {"bounds": [30, 549 + i * 100, 110, 629 + i * 100]}
        for i, row in enumerate(rows)
    ]
    native = exact_native(nodes)
    for element, node in zip(elements, nodes, strict=True):
        selector = element["native_selector"]
        assert selector["bounds"] == node["bounds"]
        result = await execute_native("read_control", {"selector": selector}, 720, 1600, native)
        assert result["success"] and result["node"]["bounds"] == node["bounds"]


def test_clipped_hit_area_does_not_replace_the_original_native_identity_bounds():
    raw = hierarchy_node(
        text="partly visible",
        **{
            "resource-id": "",
            "bounds": "[-20,-40][300,60]",
        },
    )
    filtered = filter_ui_hierarchy([raw], screen_width=720, screen_height=1600)
    _, elements, _ = format_minimal_list_with_elements(filtered, 720, 1600)
    assert elements[0]["bounds"] == [0, 0, 300, 60]
    assert elements[0]["native_selector"]["bounds"] == [-20, -40, 300, 60]
    ControlSelector.model_validate(elements[0]["native_selector"])


def test_ocr_only_geometry_is_not_presented_as_a_native_selector():
    text, elements, _ = format_minimal_list_with_elements(
        [{"text": "17:42:17", "class": "[OCR]", "bounds": "[30,549][110,629]"}],
        720,
        1600,
    )
    assert "native_selector" not in text
    assert "native_selector" not in elements[0]


@pytest.mark.parametrize("action", ["read_control", "control_action"])
@pytest.mark.parametrize("extra", [{"text": "17:42:17"}, {"bounds": [41, 343, 152, 393]}])
@pytest.mark.asyncio
async def test_explicit_stale_text_or_wrong_bounds_are_not_silently_dropped_or_retried(
    action, extra
):
    raw = hierarchy_node()
    native = exact_native([native_node(raw)])
    request = {"selector": {"resource_id": raw["resource-id"], **extra}}
    if action == "control_action":
        request["action"] = "click"
    result = await execute_native(action, request, 720, 1600, native)
    assert not result["success"] and result["code"] == "NOT_FOUND"
    assert result["execution"] == "not_started"
    assert native.await_args.args[1]["selector"] == request["selector"]
    native.assert_awaited_once()