271 lines
9.7 KiB
Python
271 lines
9.7 KiB
Python
"""Mixed native sequences exercise production preflight and dispatch over mocks."""
|
|
|
|
from copy import deepcopy
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from artemis.mcp.action_manifest import DEVICE_ACTIONS
|
|
from artemis.mcp.actuators import MockActuator
|
|
|
|
|
|
SELECTOR = {"resource_id": "fixture:id/editor"}
|
|
TOKEN = "a" * 64
|
|
|
|
|
|
def wait_step():
|
|
return {"action": "wait_until", "selector": SELECTOR, "state": "visible", "timeout_ms": 100}
|
|
|
|
|
|
def control_step():
|
|
return {
|
|
"action": "control_action",
|
|
"request": {"selector": SELECTOR, "action": "set_text", "text": " exact π \n"},
|
|
}
|
|
|
|
|
|
def touch_step():
|
|
return {
|
|
"action": "touch_gesture",
|
|
"frames": [
|
|
{"at_ms": 0, "points": [[250, 500]]},
|
|
{"at_ms": 150, "points": [[250, 500]]},
|
|
{"at_ms": 400, "points": [[500, 750]]},
|
|
],
|
|
"coordinate_space": "normalized",
|
|
"target_description": "observed canvas",
|
|
}
|
|
|
|
|
|
def click_step():
|
|
return {"action": "click", "target": [50, 100], "target_description": "observed button"}
|
|
|
|
|
|
def observed():
|
|
return {
|
|
"success": True,
|
|
"execution": "observed",
|
|
"code": "OK",
|
|
"node": {
|
|
"token": TOKEN,
|
|
"content_signature": "b" * 64,
|
|
"visible": True,
|
|
"enabled": True,
|
|
"editable": True,
|
|
"password": False,
|
|
"text": " AπB ",
|
|
"selection_start": 1,
|
|
"selection_end": 3,
|
|
"actions": ["set_text", "set_selection"],
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def sequence_limit(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"artemis.config.agent.load_agent_config",
|
|
lambda: SimpleNamespace(
|
|
pro=SimpleNamespace(execution=SimpleNamespace(max_burst_actions=10))
|
|
),
|
|
)
|
|
|
|
|
|
def actuator_with_native(capabilities=DEVICE_ACTIONS):
|
|
actuator = MockActuator(width=720, height=1600, capabilities=capabilities)
|
|
|
|
async def native(command, args, *, timeout=None):
|
|
if command == "inspect_control":
|
|
return observed()
|
|
return {
|
|
"success": True,
|
|
"execution": "completed",
|
|
"code": "OK",
|
|
"native_marker": command,
|
|
}
|
|
|
|
actuator.controller.native_request = AsyncMock(side_effect=native)
|
|
actuator.controller.tap_at = AsyncMock(return_value=SimpleNamespace(error=None))
|
|
actuator.controller.swipe_coords = AsyncMock()
|
|
return actuator
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"bad_member",
|
|
[
|
|
{**wait_step(), "timeout_ms": 0},
|
|
{**wait_step(), "state": "text"},
|
|
{**wait_step(), "timeout_ms": 3001},
|
|
{"action": "control_action", "request": {"selector": SELECTOR, "action": "set_progress"}},
|
|
{
|
|
"action": "control_action",
|
|
"request": {"selector": SELECTOR, "action": "set_text", "text": "x" * 1025},
|
|
},
|
|
{
|
|
"action": "control_action",
|
|
"request": {"selector": SELECTOR, "action": "click", "text": "unexpected"},
|
|
},
|
|
{
|
|
**touch_step(),
|
|
"frames": [
|
|
{"at_ms": 0, "points": [[100, 100]]},
|
|
{"at_ms": 400, "points": [[1001, 100]]},
|
|
],
|
|
},
|
|
{
|
|
**touch_step(),
|
|
"frames": [
|
|
{"at_ms": 0, "points": [[100, 100], [300, 100]]},
|
|
{"at_ms": 400, "points": [[300, 100], [100, 100]]},
|
|
],
|
|
},
|
|
{
|
|
**touch_step(),
|
|
"frames": [
|
|
{"at_ms": 0, "points": [[100, 100]]},
|
|
{"at_ms": 400, "points": [[200, 200], [300, 300]]},
|
|
],
|
|
},
|
|
],
|
|
)
|
|
async def test_all_native_parameters_and_paths_are_preflighted_before_even_the_first_click(
|
|
bad_member,
|
|
):
|
|
actuator = actuator_with_native()
|
|
result = await actuator.action_sequence(
|
|
[click_step(), wait_step(), control_step(), touch_step(), bad_member]
|
|
)
|
|
assert not result.ok and result.code.value == "INVALID_ARGS"
|
|
actuator.controller.native_request.assert_not_awaited()
|
|
actuator.controller.tap_at.assert_not_awaited()
|
|
actuator.controller.swipe_coords.assert_not_awaited()
|
|
assert actuator.action_history == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("missing_capability", ["wait_until", "control_action", "touch_gesture"])
|
|
async def test_every_native_member_capability_is_checked_before_any_input(missing_capability):
|
|
actuator = actuator_with_native(DEVICE_ACTIONS - {missing_capability})
|
|
result = await actuator.action_sequence(
|
|
[click_step(), wait_step(), control_step(), touch_step()]
|
|
)
|
|
assert not result.ok and result.code.value == "UNSUPPORTED"
|
|
assert result.metadata == {"execution": "not_started", "side_effect": "none"}
|
|
actuator.controller.native_request.assert_not_awaited()
|
|
actuator.controller.tap_at.assert_not_awaited()
|
|
actuator.controller.swipe_coords.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mixed_wait_control_and_touch_keep_native_receipts_and_exact_text_in_order():
|
|
actuator = actuator_with_native()
|
|
steps = [wait_step(), control_step(), touch_step()]
|
|
original = deepcopy(steps)
|
|
result = await actuator.action_sequence(steps)
|
|
assert result.ok and result.metadata["side_effect"] == "possible"
|
|
assert steps == original
|
|
calls = actuator.controller.native_request.await_args_list
|
|
assert [call.args[0] for call in calls] == [
|
|
"inspect_control",
|
|
"inspect_control",
|
|
"control_action",
|
|
"touch_gesture",
|
|
]
|
|
assert calls[2].args[1]["text"] == " exact π \n"
|
|
assert calls[2].args[1]["expected_token"] == TOKEN
|
|
receipts = result.metadata["sequence_results"]
|
|
assert [item["action"] for item in receipts] == [
|
|
"wait_until",
|
|
"control_action",
|
|
"touch_gesture",
|
|
]
|
|
assert receipts[0]["read_only"] and receipts[0]["side_effect"] == "none"
|
|
assert receipts[0]["native_execution"] == "observed"
|
|
assert receipts[0]["node"]["selection_end"] == 3
|
|
assert receipts[1]["native_receipt"]["native_marker"] == "control_action"
|
|
assert receipts[2]["native_receipt"]["native_marker"] == "touch_gesture"
|
|
assert receipts[0]["ended_ms"] <= receipts[1]["started_ms"] <= receipts[2]["started_ms"]
|
|
actuator.controller.tap_at.assert_not_awaited()
|
|
actuator.controller.swipe_coords.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("failure", [False, True])
|
|
async def test_readonly_native_wait_never_counts_as_an_input_side_effect(failure):
|
|
actuator = actuator_with_native()
|
|
if failure:
|
|
actuator.controller.native_request.return_value = None
|
|
actuator.controller.native_request.side_effect = lambda *args, **kwargs: {
|
|
"success": False,
|
|
"execution": "not_started",
|
|
"code": "UNAVAILABLE",
|
|
"error": "No active native tree",
|
|
}
|
|
result = await actuator.action_sequence([wait_step()])
|
|
assert result.ok is not failure
|
|
assert result.metadata["side_effect"] == "none"
|
|
receipt = result.metadata["sequence_results"][0]
|
|
assert receipt["read_only"] and receipt["side_effect"] == "none"
|
|
assert receipt["native_execution"] == ("not_started" if failure else "observed")
|
|
assert [call.args[0] for call in actuator.controller.native_request.await_args_list] == [
|
|
"inspect_control"
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("unknown_member", ["control_action", "touch_gesture"])
|
|
async def test_unknown_native_input_stops_every_remaining_member_and_preserves_original_receipt(
|
|
unknown_member,
|
|
):
|
|
actuator = actuator_with_native()
|
|
uncertain = {
|
|
"success": False,
|
|
"execution": "unknown",
|
|
"code": "TRANSPORT_ERROR",
|
|
"error": "Reply lost; do not replay input",
|
|
"native_marker": "unacknowledged input",
|
|
}
|
|
|
|
async def native(command, args, *, timeout=None):
|
|
return observed() if command == "inspect_control" else uncertain
|
|
|
|
actuator.controller.native_request.side_effect = native
|
|
member = control_step() if unknown_member == "control_action" else touch_step()
|
|
result = await actuator.action_sequence([wait_step(), member, click_step(), touch_step()])
|
|
assert not result.ok and result.metadata["side_effect"] == "possible"
|
|
receipts = result.metadata["sequence_results"]
|
|
assert receipts[1]["execution"] == "unknown"
|
|
assert receipts[1]["native_execution"] == "unknown"
|
|
assert receipts[1]["native_receipt"] == uncertain
|
|
assert [item["execution"] for item in receipts[2:]] == ["not_started", "not_started"]
|
|
writes = [
|
|
call.args[0]
|
|
for call in actuator.controller.native_request.await_args_list
|
|
if call.args[0] != "inspect_control"
|
|
]
|
|
assert writes == [unknown_member]
|
|
actuator.controller.tap_at.assert_not_awaited()
|
|
actuator.controller.swipe_coords.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normalized_touch_in_sequence_is_mapped_to_physical_pixels_exactly_once():
|
|
actuator = actuator_with_native()
|
|
result = await actuator.action_sequence([touch_step()])
|
|
assert result.ok
|
|
command, sent = actuator.controller.native_request.await_args.args
|
|
assert command == "touch_gesture" and sent["screen_size"] == [720, 1600]
|
|
expected = [
|
|
{"at_ms": 0, "points": [[180, 800]]},
|
|
{"at_ms": 150, "points": [[180, 800]]},
|
|
{"at_ms": 400, "points": [[360, 1200]]},
|
|
]
|
|
assert sent["frames"] == expected
|
|
receipt = result.metadata["sequence_results"][0]
|
|
assert receipt["plan"]["frames"] == expected
|
|
assert receipt["plan"]["coordinate_space"] == "pixel"
|
|
assert receipt["args"]["coordinate_space"] == "normalized"
|
|
assert receipt["args"]["frames"] == touch_step()["frames"]
|