854 lines
33 KiB
Python
854 lines
33 KiB
Python
"""Behavioral regressions for short-lived controls and measured scale movement."""
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from artemis.mcp.action_server import build_action_server
|
|
from artemis.mcp.action_session import ActionSession
|
|
from artemis.mcp.actuators import MockActuator
|
|
from artemis.mcp.gestures import DragRequest, plan_drag
|
|
|
|
|
|
def prepared_sequence():
|
|
return [
|
|
{
|
|
"action": "ensure_visible",
|
|
"target": "test.app:id/panel",
|
|
"opener": "test.app:id/opener",
|
|
"target_description": "show panel",
|
|
"timeout_ms": 100,
|
|
},
|
|
{
|
|
"action": "click",
|
|
"target": "test.app:id/panel",
|
|
"target_description": "operate panel",
|
|
},
|
|
]
|
|
|
|
|
|
PANEL = {"resource_id": "test.app:id/panel", "bounds": [200, 300, 300, 400]}
|
|
OPENER = {"resource_id": "test.app:id/opener", "bounds": [100, 1400, 200, 1500]}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_preparation_keeps_open_panel_open_and_operates_its_live_target():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.get_ui_elements = AsyncMock(return_value=[OPENER, PANEL])
|
|
actuator.controller.tap_at = AsyncMock(return_value=SimpleNamespace(error=None))
|
|
result = await actuator.action_sequence(prepared_sequence())
|
|
assert result.ok
|
|
actuator.controller.tap_at.assert_awaited_once_with(250, 350)
|
|
assert result.metadata["sequence_results"][0]["execution"] == "skipped"
|
|
assert result.metadata["sequence_results"][0]["completed_taps"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_preparation_waits_for_delayed_panel_then_uses_current_bounds():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
moved = {**PANEL, "bounds": [300, 400, 400, 500]}
|
|
actuator.controller.get_ui_elements = AsyncMock(
|
|
side_effect=[[OPENER], [OPENER], [OPENER, PANEL], [OPENER, moved]]
|
|
)
|
|
actuator.controller.tap_at = AsyncMock(return_value=SimpleNamespace(error=None))
|
|
sequence = prepared_sequence()
|
|
sequence[0]["timeout_ms"] = 500
|
|
result = await actuator.action_sequence(sequence)
|
|
assert result.ok
|
|
assert [c.args for c in actuator.controller.tap_at.await_args_list] == [(150, 1450), (350, 450)]
|
|
receipt = result.metadata["sequence_results"][0]
|
|
assert receipt["completed_taps"] == 1
|
|
assert receipt["visibility_attempts"] == 3
|
|
assert receipt["visibility_guard"]["visible"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_preparation_timeout_never_retries_opener_or_executes_body():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.get_ui_elements = AsyncMock(return_value=[OPENER])
|
|
actuator.controller.tap_at = AsyncMock(return_value=SimpleNamespace(error=None))
|
|
result = await actuator.action_sequence(prepared_sequence())
|
|
assert not result.ok and result.code.value == "TIMEOUT"
|
|
actuator.controller.tap_at.assert_awaited_once_with(150, 1450)
|
|
assert result.metadata["side_effect"] == "possible"
|
|
assert [r["execution"] for r in result.metadata["sequence_results"]] == [
|
|
"unknown",
|
|
"not_started",
|
|
]
|
|
assert result.metadata["sequence_results"][0]["completed_taps"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("elements", [[PANEL, PANEL], [{"resource_id": "test.app:id/root"}], []])
|
|
async def test_preparation_refuses_ambiguous_target_missing_opener_or_unavailable_ui(elements):
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.get_ui_elements = AsyncMock(return_value=elements)
|
|
actuator.controller.tap_at = AsyncMock()
|
|
result = await actuator.action_sequence(prepared_sequence())
|
|
assert not result.ok and result.metadata["execution"] == "not_started"
|
|
assert result.metadata["side_effect"] == "none"
|
|
actuator.controller.tap_at.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_panel_disappearing_after_preparation_stops_without_reopening():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.get_ui_elements = AsyncMock(
|
|
side_effect=[[OPENER], [OPENER, PANEL], [OPENER]]
|
|
)
|
|
actuator.controller.tap_at = AsyncMock(return_value=SimpleNamespace(error=None))
|
|
result = await actuator.action_sequence(prepared_sequence())
|
|
assert not result.ok and result.code.value == "TARGET_NOT_FOUND"
|
|
actuator.controller.tap_at.assert_awaited_once_with(150, 1450)
|
|
assert result.metadata["sequence_results"][1]["execution"] == "not_started"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("invalid", ["same_id", "bad_late_member", "bad_timeout"])
|
|
async def test_preparation_validates_whole_sequence_before_opening(invalid):
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.tap_at = AsyncMock()
|
|
sequence = prepared_sequence()
|
|
if invalid == "same_id":
|
|
sequence[0]["opener"] = sequence[0]["target"]
|
|
elif invalid == "bad_timeout":
|
|
sequence[0]["timeout_ms"] = 3001
|
|
else:
|
|
sequence[1]["times"] = 0
|
|
result = await actuator.action_sequence(sequence)
|
|
assert not result.ok and result.code.value == "INVALID_ARGS"
|
|
actuator.controller.tap_at.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_preparation_round_trips_through_shared_mcp_action_server():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.get_ui_elements = AsyncMock(return_value=[OPENER, PANEL])
|
|
actuator.controller.tap_at = AsyncMock(return_value=SimpleNamespace(error=None))
|
|
session = await ActionSession(build_action_server(actuator)).start()
|
|
try:
|
|
result = await session.call("action_sequence", {"steps": prepared_sequence()})
|
|
assert result.ok
|
|
actuator.controller.tap_at.assert_awaited_once_with(250, 350)
|
|
finally:
|
|
await session.aclose()
|
|
|
|
|
|
def test_pan_uses_decreasing_observed_time_scale_without_overshoot():
|
|
request = DragRequest(
|
|
start=[190, 1059],
|
|
coordinate_space="pixel",
|
|
target_description="timeline",
|
|
calibration={
|
|
"axis": "y",
|
|
"mode": "pan",
|
|
"anchors": [
|
|
{"position": 1075, "value": 38400},
|
|
{"position": 1187, "value": 37800},
|
|
],
|
|
"target_value": 37800,
|
|
"indicator_position": 1215,
|
|
},
|
|
)
|
|
planned = plan_drag(request, 720, 1600)
|
|
assert planned["coordinates"] == [190, 1059, 190, 1087]
|
|
assert planned["calibration"]["value_per_pixel"] == pytest.approx(-600 / 112)
|
|
assert planned["calibration"]["verification"] == "pending_observation"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"axis,start,expected", [("x", [80, 200], [125, 200]), ("y", [200, 80], [200, 125])]
|
|
)
|
|
def test_thumb_maps_value_to_the_requested_axis(axis, start, expected):
|
|
planned = plan_drag(
|
|
DragRequest(
|
|
start=start,
|
|
coordinate_space="pixel",
|
|
target_description="slider",
|
|
calibration={
|
|
"axis": axis,
|
|
"mode": "thumb",
|
|
"anchors": [
|
|
{"position": 100, "value": 0},
|
|
{"position": 200, "value": 100},
|
|
],
|
|
"target_value": 25,
|
|
},
|
|
),
|
|
720,
|
|
1600,
|
|
)
|
|
assert planned["coordinates"] == [*start, *expected]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pixel_drag_is_not_quantized_through_normalized_coordinates():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
session = await ActionSession(build_action_server(actuator)).start()
|
|
try:
|
|
result = await session.call(
|
|
"drag",
|
|
{
|
|
"start": [191, 1001],
|
|
"end": [191, 1032],
|
|
"coordinate_space": "pixel",
|
|
"target_description": "timeline",
|
|
},
|
|
)
|
|
assert result.ok
|
|
assert result.metadata["coordinates"] == [191, 1001, 191, 1032]
|
|
assert actuator.action_history == [
|
|
{
|
|
"action": "swipe",
|
|
"start": (191, 1001),
|
|
"end": (191, 1032),
|
|
"duration_ms": 300,
|
|
}
|
|
]
|
|
finally:
|
|
await session.aclose()
|
|
|
|
|
|
def steps():
|
|
return [
|
|
{"action": "click", "target": [200, 900], "target_description": "open scale"},
|
|
{"action": "wait", "time_in_ms": 1},
|
|
{
|
|
"action": "drag",
|
|
"start": [190, 1001],
|
|
"end": [190, 1032],
|
|
"coordinate_space": "pixel",
|
|
"target_description": "scale handle",
|
|
},
|
|
]
|
|
|
|
|
|
def test_both_observation_lists_expose_real_resource_ids_without_changing_indices():
|
|
from artemis.utils.visualization import (
|
|
format_minimal_list_with_elements,
|
|
format_minimal_list_with_points,
|
|
)
|
|
|
|
nodes = [
|
|
{"text": "plus", "resource-id": "test.app:id/plus", "bounds": "[10,20][30,40]"},
|
|
{"text": "minus", "resource_id": "test.app:id/minus", "bounds": "[10,60][30,80]"},
|
|
]
|
|
for formatter in (format_minimal_list_with_points, format_minimal_list_with_elements):
|
|
text, _, labels = formatter(nodes, 100, 100)
|
|
assert (
|
|
"[1] Text: 'plus' | Bounds: [100,200][300,400] | resource_id: 'test.app:id/plus'"
|
|
in text
|
|
)
|
|
assert "[2] Text: 'minus'" in text and "resource_id: 'test.app:id/minus'" in text
|
|
assert labels == ["1", "2"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_live_resource_click_uses_current_bounds_and_reuses_guard_observation():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
panel, opener = "test.app:id/panel", "test.app:id/opener"
|
|
actuator.controller.get_ui_elements = AsyncMock(
|
|
return_value=[
|
|
{"resource_id": opener, "bounds": [100, 1400, 200, 1500]},
|
|
]
|
|
)
|
|
actuator.controller.tap_at = AsyncMock(return_value=SimpleNamespace(error=None))
|
|
result = await actuator.action_sequence(
|
|
[
|
|
{
|
|
"action": "click",
|
|
"target": opener,
|
|
"target_description": "open",
|
|
"unless_visible": panel,
|
|
}
|
|
]
|
|
)
|
|
assert result.ok
|
|
actuator.controller.tap_at.assert_awaited_once_with(150, 1450)
|
|
actuator.controller.get_ui_elements.assert_awaited_once()
|
|
receipt = result.metadata["sequence_results"][0]
|
|
assert receipt["coordinates"] == [150, 1450]
|
|
assert receipt["target_resource_id"] == opener
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("matches", [0, 2])
|
|
async def test_missing_or_ambiguous_resource_click_never_guesses_coordinates(matches):
|
|
actuator = MockActuator(width=720, height=1600)
|
|
target = "test.app:id/control"
|
|
actuator.controller.get_ui_elements = AsyncMock(
|
|
return_value=[
|
|
{
|
|
"resource_id": target if i < matches else "test.app:id/other",
|
|
"bounds": [10, 20, 30, 40],
|
|
}
|
|
for i in range(2)
|
|
]
|
|
)
|
|
actuator.controller.tap_at = AsyncMock()
|
|
result = await actuator.action_sequence(
|
|
[{"action": "click", "target": target, "target_description": "control"}]
|
|
)
|
|
assert not result.ok and result.code.value == "TARGET_NOT_FOUND"
|
|
assert result.metadata["execution"] == "not_started"
|
|
actuator.controller.tap_at.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"resource_id,bounds,success",
|
|
[
|
|
("test.app:id/control", [100, 1400, 200, 1500], True),
|
|
("test.app:id/other", [100, 1400, 200, 1500], False),
|
|
("test.app:id/control", [300, 1400, 400, 1500], False),
|
|
],
|
|
)
|
|
async def test_indexed_click_checks_live_identity_at_original_position(
|
|
resource_id, bounds, success
|
|
):
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.get_ui_elements = AsyncMock(
|
|
return_value=[{"resource_id": resource_id, "bounds": bounds}]
|
|
)
|
|
actuator.controller.tap_at = AsyncMock(return_value=SimpleNamespace(error=None))
|
|
result = await actuator.click(217, 906, expected_resource_id="test.app:id/control")
|
|
assert result.ok is success
|
|
assert actuator.controller.tap_at.await_count == int(success)
|
|
if success:
|
|
from artemis.agents.flash.runner import FlashRunner
|
|
|
|
assert result.metadata["pixel_coordinates"] == [156, 1449]
|
|
assert "coordinates" not in result.metadata
|
|
runner = object.__new__(FlashRunner)
|
|
record = runner._build_action_record(
|
|
"click", {"target": [217, 906]}, SimpleNamespace(metadata=result.metadata)
|
|
)
|
|
assert record["coordinates"] == [217, 906]
|
|
assert record["coordinate_space"] == "normalized"
|
|
assert record["completed_taps"] == 1
|
|
assert record["pixel_coordinates"] == [156, 1449]
|
|
else:
|
|
assert result.metadata["execution"] == "not_started"
|
|
assert result.metadata["side_effect"] == "none"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_coordinate_only_click_keeps_existing_direct_dispatch():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.get_ui_elements = AsyncMock()
|
|
result = await actuator.click(217, 906)
|
|
assert result.ok
|
|
actuator.controller.get_ui_elements.assert_not_awaited()
|
|
|
|
|
|
def test_flash_and_pro_forward_observed_resource_identity_without_inventing_it():
|
|
from artemis.mcp.action_executor import McpActionExecutor
|
|
from artemis.mcp.action_names import to_canonical_call
|
|
|
|
executor = object.__new__(McpActionExecutor)
|
|
executor._resolve_target = MagicMock(
|
|
return_value=([217, 906], {"target_resource_id": "test.app:id/control"})
|
|
)
|
|
name, args, _, _ = executor._translate("click", {"target": 1}, MagicMock())
|
|
assert name == "click" and args["expected_resource_id"] == "test.app:id/control"
|
|
_, args = to_canonical_call(
|
|
{
|
|
"action": "tap",
|
|
"normalized_coordinates": [217, 906],
|
|
"resource_id": "test.app:id/control",
|
|
}
|
|
)
|
|
assert args["expected_resource_id"] == "test.app:id/control"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_repeated_resource_taps_are_one_bounded_sequence_member():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.get_ui_elements = AsyncMock(
|
|
return_value=[{"resource_id": "test.app:id/plus", "bounds": [10, 20, 30, 40]}]
|
|
)
|
|
actuator.controller.tap_at = AsyncMock(return_value=SimpleNamespace(error=None))
|
|
result = await actuator.action_sequence(
|
|
[
|
|
{
|
|
"action": "click",
|
|
"target": "test.app:id/plus",
|
|
"target_description": "plus",
|
|
"times": 10,
|
|
"delay_ms": 50,
|
|
}
|
|
]
|
|
)
|
|
assert result.ok and result.metadata["completed_steps"] == 1
|
|
assert actuator.controller.tap_at.await_count == 10
|
|
actuator.controller.tap_at.assert_awaited_with(20, 30)
|
|
assert actuator.controller.get_ui_elements.await_count == 10
|
|
receipt = result.metadata["sequence_results"][0]
|
|
assert receipt["times"] == receipt["completed_taps"] == 10
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("sequence", [True, False])
|
|
async def test_repeated_native_taps_stop_when_target_disappears(sequence):
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.get_ui_elements = AsyncMock(
|
|
side_effect=[
|
|
[{"resource_id": "test.app:id/plus", "bounds": [10, 20, 30, 40]}],
|
|
[{"resource_id": "test.app:id/background", "bounds": [0, 0, 720, 1600]}],
|
|
]
|
|
)
|
|
actuator.controller.tap_at = AsyncMock(return_value=SimpleNamespace(error=None))
|
|
if sequence:
|
|
result = await actuator.action_sequence(
|
|
[
|
|
{
|
|
"action": "click",
|
|
"target": "test.app:id/plus",
|
|
"target_description": "plus",
|
|
"times": 3,
|
|
"delay_ms": 0,
|
|
},
|
|
{"action": "click", "target": [100, 100], "target_description": "next"},
|
|
]
|
|
)
|
|
receipt = result.metadata["sequence_results"][0]
|
|
assert result.metadata["sequence_results"][1]["execution"] == "not_started"
|
|
else:
|
|
result = await actuator.click(
|
|
28, 19, times=3, delay_ms=0, expected_resource_id="test.app:id/plus"
|
|
)
|
|
receipt = result.metadata
|
|
assert not result.ok and result.code.value == "TARGET_NOT_FOUND"
|
|
assert receipt["completed_taps"] == 1
|
|
assert receipt["execution"] == "unknown"
|
|
actuator.controller.tap_at.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("times", [0, 11, True])
|
|
async def test_invalid_repeat_count_rejects_entire_sequence_before_input(times):
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.tap_at = AsyncMock()
|
|
invalid = steps()
|
|
invalid[-1] = {
|
|
"action": "click",
|
|
"target": [200, 900],
|
|
"target_description": "plus",
|
|
"times": times,
|
|
}
|
|
result = await actuator.action_sequence(invalid)
|
|
assert not result.ok and result.code.value == "INVALID_ARGS"
|
|
actuator.controller.tap_at.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.parametrize("limit", [2, 4, 8])
|
|
def test_runtime_sequence_contract_matches_both_modes_and_counts_waits(limit):
|
|
from langchain_core.utils.function_calling import convert_to_openai_tool
|
|
from artemis.mcp.action_specs import operator_shell_tool, tool_declaration
|
|
from artemis.mcp.gestures import validate_sequence
|
|
|
|
pro = operator_shell_tool("action_sequence", sequence_limit=limit)
|
|
flash = tool_declaration("action_sequence", sequence_limit=limit)
|
|
assert flash.parameters == convert_to_openai_tool(pro)["function"]["parameters"]
|
|
assert flash.parameters["properties"]["steps"]["maxItems"] == limit
|
|
assert f"at most {limit} members" in pro.description
|
|
with pytest.raises(ValueError, match=f"at most {limit}, including waits"):
|
|
validate_sequence({"steps": [{"action": "wait", "time_in_ms": 1}] * 13}, limit)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("already_open", [False, True])
|
|
async def test_guarded_popup_opener_handles_open_and_closed_without_toggling(already_open):
|
|
actuator = MockActuator(width=720, height=1600)
|
|
panel = "test.app:id/panel_control"
|
|
guarded = steps()
|
|
guarded[0]["unless_visible"] = panel
|
|
guarded[-1]["require_visible"] = panel
|
|
actuator.controller.get_ui_elements = AsyncMock(
|
|
side_effect=[
|
|
[{"resource_id": panel if already_open else "test.app:id/opener"}],
|
|
[{"resource_id": panel}],
|
|
]
|
|
)
|
|
actuator.controller.tap_at = AsyncMock(return_value=SimpleNamespace(error=None))
|
|
actuator.controller.swipe_coords = AsyncMock(return_value=None)
|
|
result = await actuator.action_sequence(guarded)
|
|
assert result.ok
|
|
assert actuator.controller.tap_at.await_count == (0 if already_open else 1)
|
|
actuator.controller.swipe_coords.assert_awaited_once()
|
|
assert actuator.controller.get_ui_elements.await_count == 2
|
|
assert result.metadata["skipped_steps"] == int(already_open)
|
|
assert result.metadata["sequence_results"][0]["execution"] == (
|
|
"skipped" if already_open else "completed"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_popup_expiring_before_drag_never_touches_underlying_content():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
panel = "test.app:id/panel_control"
|
|
guarded = steps()
|
|
guarded[0]["unless_visible"] = panel
|
|
guarded[-1]["require_visible"] = panel
|
|
actuator.controller.get_ui_elements = AsyncMock(
|
|
side_effect=[
|
|
[{"resource_id": panel}],
|
|
[{"resource_id": "test.app:id/card"}],
|
|
]
|
|
)
|
|
actuator.controller.tap_at = AsyncMock()
|
|
actuator.controller.swipe_coords = AsyncMock()
|
|
result = await actuator.action_sequence(guarded)
|
|
assert not result.ok and result.code.value == "TARGET_NOT_FOUND"
|
|
assert result.metadata["execution"] == "not_started"
|
|
assert result.metadata["side_effect"] == "none"
|
|
assert [r["execution"] for r in result.metadata["sequence_results"]] == [
|
|
"skipped",
|
|
"completed",
|
|
"not_started",
|
|
]
|
|
actuator.controller.tap_at.assert_not_awaited()
|
|
actuator.controller.swipe_coords.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("hierarchy", [[], RuntimeError("hierarchy unavailable")])
|
|
async def test_unavailable_hierarchy_does_not_guess_popup_state(hierarchy):
|
|
actuator = MockActuator(width=720, height=1600)
|
|
guarded = steps()
|
|
guarded[0]["unless_visible"] = "test.app:id/panel_control"
|
|
actuator.controller.get_ui_elements = AsyncMock(
|
|
side_effect=hierarchy if isinstance(hierarchy, Exception) else lambda: hierarchy
|
|
)
|
|
actuator.controller.tap_at = AsyncMock()
|
|
actuator.controller.swipe_coords = AsyncMock()
|
|
result = await actuator.action_sequence(guarded)
|
|
assert not result.ok
|
|
assert result.metadata["execution"] == "not_started"
|
|
assert result.metadata["side_effect"] == "none"
|
|
actuator.controller.tap_at.assert_not_awaited()
|
|
actuator.controller.swipe_coords.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_guarded_drag_round_trips_and_refuses_stale_target():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.get_ui_elements = AsyncMock(
|
|
return_value=[{"resource_id": "test.app:id/card"}]
|
|
)
|
|
actuator.controller.swipe_coords = AsyncMock()
|
|
session = ActionSession(server=build_action_server(actuator=actuator))
|
|
try:
|
|
await session.start()
|
|
request = {k: v for k, v in steps()[-1].items() if k != "action"}
|
|
result = await session.call(
|
|
"drag", {**request, "require_visible": "test.app:id/panel_control"}
|
|
)
|
|
assert not result.ok and result.code.value == "TARGET_NOT_FOUND"
|
|
assert result.metadata["execution"] == "not_started"
|
|
assert not result.metadata["visibility_guard"]["visible"]
|
|
actuator.controller.swipe_coords.assert_not_awaited()
|
|
finally:
|
|
await session.aclose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sequence_has_no_observation_between_inputs_and_returns_member_receipts():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.tap_at = AsyncMock(return_value=MagicMock(error=None))
|
|
actuator.controller.swipe_coords = AsyncMock(return_value=None)
|
|
actuator.take_screenshot = AsyncMock()
|
|
result = await actuator.action_sequence(steps())
|
|
assert result.ok
|
|
actuator.controller.tap_at.assert_awaited_once_with(144, 1440)
|
|
actuator.controller.swipe_coords.assert_awaited_once_with(190, 1001, 190, 1032, 300)
|
|
actuator.take_screenshot.assert_not_awaited()
|
|
receipts = result.metadata["sequence_results"]
|
|
assert [r["execution"] for r in receipts] == ["completed"] * 3
|
|
assert receipts[0]["ended_ms"] <= receipts[1]["started_ms"] <= receipts[2]["started_ms"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_late_invalid_member_prevents_even_the_first_input():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.tap_at = AsyncMock()
|
|
invalid = steps()
|
|
invalid[-1]["end"] = [720, 1032]
|
|
result = await actuator.action_sequence(invalid)
|
|
assert not result.ok and result.code.value == "INVALID_ARGS"
|
|
actuator.controller.tap_at.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failure_keeps_completed_members_and_does_not_dispatch_the_remainder():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.controller.tap_at = AsyncMock(
|
|
side_effect=[MagicMock(error=None), RuntimeError("transport lost")]
|
|
)
|
|
member = {"action": "click", "target": [200, 900], "target_description": "control"}
|
|
result = await actuator.action_sequence([member, member, member])
|
|
assert not result.ok
|
|
assert actuator.controller.tap_at.await_count == 2
|
|
assert [r["execution"] for r in result.metadata["sequence_results"]] == [
|
|
"completed",
|
|
"unknown",
|
|
"not_started",
|
|
]
|
|
assert result.metadata["execution"] == "partial"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancellation_is_not_swallowed_or_retried():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
actuator.ctx.data_engine = MagicMock(current_step_id="step")
|
|
actuator.controller.swipe_coords = AsyncMock(side_effect=asyncio.CancelledError)
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await actuator.action_sequence(steps())
|
|
assert len(actuator.action_history) == 1
|
|
assert actuator.controller.swipe_coords.await_count == 1
|
|
receipts = [
|
|
c.kwargs["payload"]["receipt"]
|
|
for c in actuator.ctx.data_engine.record_trace.call_args_list
|
|
if "receipt" in c.kwargs["payload"]
|
|
]
|
|
assert [r["execution"] for r in receipts] == ["completed", "completed", "unknown"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pause_at_member_boundary_does_not_dispatch_later_inputs(tmp_path):
|
|
from artemis.runtime.task_control import TaskControl
|
|
|
|
actuator = MockActuator(width=720, height=1600)
|
|
control = object.__new__(TaskControl)
|
|
control.directory = tmp_path
|
|
actuator.ctx.task_control = control
|
|
|
|
async def tap(*_):
|
|
(tmp_path / "pause.request").write_text("{}")
|
|
return SimpleNamespace(error=None)
|
|
|
|
actuator.controller.tap_at = AsyncMock(side_effect=tap)
|
|
actuator.controller.swipe_coords = AsyncMock()
|
|
result = await actuator.action_sequence(steps())
|
|
assert not result.ok
|
|
assert result.metadata["completed_steps"] == 1
|
|
assert result.metadata["sequence_results"][1]["execution"] == "not_started"
|
|
actuator.controller.swipe_coords.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_over_limit_is_rejected_before_dispatch(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"artemis.config.agent.load_agent_config",
|
|
lambda: SimpleNamespace(
|
|
pro=SimpleNamespace(execution=SimpleNamespace(max_burst_actions=2)),
|
|
),
|
|
)
|
|
actuator = MockActuator(width=720, height=1600)
|
|
result = await actuator.action_sequence(steps())
|
|
assert not result.ok and "at most 2, including waits" in result.message
|
|
assert actuator.action_history == []
|
|
|
|
|
|
@pytest.mark.parametrize("values", [(0, 0), (1e308, -1e308), (float("nan"), 1), (True, 5)])
|
|
def test_degenerate_and_nonfinite_calibration_is_rejected(values):
|
|
with pytest.raises(ValueError):
|
|
request = DragRequest(
|
|
start=[100, 100],
|
|
coordinate_space="pixel",
|
|
target_description="timeline",
|
|
calibration={
|
|
"axis": "y",
|
|
"mode": "pan",
|
|
"indicator_position": 100,
|
|
"anchors": [
|
|
{"position": 100, "value": values[0]},
|
|
{"position": 200, "value": values[1]},
|
|
],
|
|
"target_value": 3,
|
|
},
|
|
)
|
|
plan_drag(request, 720, 1600)
|
|
|
|
|
|
def test_flash_and_pro_schemas_preserve_null_for_calculated_endpoints():
|
|
from langchain_core.utils.function_calling import convert_to_openai_tool
|
|
from artemis.mcp.action_specs import tool_declaration, operator_shell_tool
|
|
|
|
flash = tool_declaration("drag").parameters
|
|
pro = convert_to_openai_tool(operator_shell_tool("drag"))["function"]["parameters"]
|
|
assert flash == pro
|
|
assert {"type": "null"} in flash["properties"]["end"]["anyOf"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_conflicting_endpoint_has_actionable_error_and_no_input():
|
|
actuator = MockActuator(width=720, height=1600)
|
|
result = await actuator.drag(
|
|
start=[190, 1200],
|
|
end=[190, 1100],
|
|
coordinate_space="pixel",
|
|
target_description="timeline",
|
|
calibration={
|
|
"axis": "y",
|
|
"mode": "pan",
|
|
"anchors": [{"position": 1000, "value": 60}, {"position": 1100, "value": 0}],
|
|
"target_value": 30,
|
|
"indicator_position": 1050,
|
|
},
|
|
)
|
|
assert not result.ok and "set end=null" in result.message
|
|
assert actuator.action_history == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mode", ["flash", "pro"])
|
|
async def test_each_mode_exposes_the_receipt_to_its_next_model_turn(mode):
|
|
from artemis.agents.validator.action_execution import exec_action
|
|
from artemis.mcp.action_executor import McpActionExecutor
|
|
from artemis.mcp.action_types import ActionResult, ObserveResult
|
|
|
|
ctx = SimpleNamespace(
|
|
data_engine=None, device=SimpleNamespace(device_width=720, device_height=1600)
|
|
)
|
|
receipt = {
|
|
"sequence_results": [{"action": "click", "execution": "completed"}],
|
|
"execution": "partial",
|
|
}
|
|
session = MagicMock()
|
|
session.call = AsyncMock(
|
|
return_value=ActionResult.failure("action_sequence", "Interrupted", metadata=receipt)
|
|
)
|
|
args = {"steps": steps()}
|
|
if mode == "pro":
|
|
action = {"action": "action_sequence", "args": args}
|
|
ok, _ = await exec_action(ctx, session, action)
|
|
assert not ok
|
|
assert action["execution"] == "partial"
|
|
from artemis.utils.task_tree import format_result_clean
|
|
|
|
text = format_result_clean({"status": "failed", "execution": [action]})
|
|
else:
|
|
executor = McpActionExecutor(ctx, actuator=MagicMock())
|
|
executor._session_or_start = AsyncMock(return_value=session)
|
|
executor._observe_after_action = AsyncMock(return_value=(ObserveResult(ok=True), True))
|
|
result = await executor._execute_device_action(
|
|
"action_sequence",
|
|
"action_sequence",
|
|
args,
|
|
"gesture",
|
|
SimpleNamespace(latest_screenshot=None, indexed_elements=[]),
|
|
)
|
|
assert result.status == "error"
|
|
executor._observe_after_action.assert_awaited_once()
|
|
text = result.text_summary
|
|
assert '"execution": "partial"' in text
|
|
assert '"execution": "completed"' in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pro_translation_and_wire_dispatch_preserve_exact_pixel_drag():
|
|
from artemis.agents.operator.operator import OperatorNode
|
|
from artemis.agents.validator.action_execution import exec_action
|
|
|
|
actuator = MockActuator(width=720, height=1600)
|
|
node = object.__new__(OperatorNode)
|
|
node.ctx = actuator.ctx
|
|
request = {
|
|
"start": [191, 1001],
|
|
"end": [191, 1032],
|
|
"coordinate_space": "pixel",
|
|
"target_description": "timeline",
|
|
}
|
|
state = SimpleNamespace(operator_raw_data={"width": 720, "height": 1600})
|
|
actions, error = node._translate_and_validate_tool_inner(
|
|
{"name": "drag", "args": request}, state
|
|
)
|
|
assert error is None
|
|
assert actions[0]["coordinates"] == [191, 1001, 191, 1032]
|
|
assert actions[0]["normalized_coordinates"] == [265, 626, 265, 645]
|
|
session = await ActionSession(build_action_server(actuator)).start()
|
|
try:
|
|
ok, _ = await exec_action(actuator.ctx, session, actions[0])
|
|
assert ok
|
|
assert actuator.action_history[-1]["end"] == (191, 1032)
|
|
assert actions[0]["execution"] == "completed"
|
|
finally:
|
|
await session.aclose()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"change",
|
|
[
|
|
{"start": [True, 10]},
|
|
{"duration_ms": 0},
|
|
{"target_description": " "},
|
|
{"end": None},
|
|
{"coordinate_space": "screen"},
|
|
],
|
|
)
|
|
def test_bad_drag_arguments_are_rejected(change):
|
|
args = {"start": [10, 20], "end": [20, 30], "target_description": "slider", **change}
|
|
with pytest.raises(ValueError):
|
|
DragRequest.model_validate(args)
|
|
|
|
|
|
def test_pixel_history_under_1000_still_uses_its_actual_screen_geometry():
|
|
from artemis.utils.coordinates import normalize_step_actions
|
|
|
|
action = {
|
|
"action": "drag",
|
|
"coordinates": [191, 900, 191, 920],
|
|
"coordinate_space": "pixel",
|
|
"screen_size": [720, 1600],
|
|
}
|
|
step = normalize_step_actions({"action_taken": action})
|
|
assert step["action_taken"]["coordinates"] == [191, 900, 191, 920]
|
|
assert step["action_taken"]["normalized_coordinates"] == [265, 562, 265, 575]
|
|
|
|
|
|
def test_overlay_uses_executed_pixels_instead_of_original_request_units():
|
|
import io
|
|
from PIL import Image
|
|
from artemis.utils.visualization import overlay_action_on_screenshot
|
|
|
|
buffer = io.BytesIO()
|
|
Image.new("RGB", (720, 1600), "white").save(buffer, format="PNG")
|
|
action = {
|
|
"action": "drag",
|
|
"coordinates": [190, 900, 190, 930],
|
|
"coordinate_space": "pixel",
|
|
"args": {"start": [264, 563], "end": [264, 581], "coordinate_space": "normalized"},
|
|
}
|
|
drawn = overlay_action_on_screenshot(buffer.getvalue(), action)
|
|
assert drawn is not None
|
|
result = Image.open(io.BytesIO(drawn))
|
|
red, green, blue = result.getpixel((190, 915))
|
|
assert red > 150 and green < 120 and blue < 120
|
|
assert min(result.getpixel((137, 1450))) > 240
|
|
|
|
|
|
def test_sequence_overlay_does_not_draw_unexecuted_targets():
|
|
from artemis.utils.visualization import draw_action_overlay_on_image
|
|
|
|
raw = b"unused original image"
|
|
assert (
|
|
draw_action_overlay_on_image(
|
|
raw,
|
|
"action_sequence",
|
|
{
|
|
"sequence_results": [
|
|
{
|
|
"action": "click",
|
|
"coordinates": [190, 900],
|
|
"coordinate_space": "pixel",
|
|
"execution": "not_started",
|
|
},
|
|
]
|
|
},
|
|
)
|
|
== raw
|
|
)
|