merge(checker): integrate verification persistence
This commit is contained in:
somew1nd committed
commit
17f734a473
37 files changed
+748
-1985
No files matched your search
@@ -0,0 +1,223 @@
|
||||
{
|
||||
"tap": {
|
||||
"description": "Taps on the screen at coordinates.\n\n'coordinates' is a list [x, y].\n'times' is the number of consecutive clicks (default 1).\n'delay_ms' is the delay in milliseconds between consecutive clicks (default\n100).\n",
|
||||
"inputSchema": {
|
||||
"properties": {
|
||||
"coordinates": {
|
||||
"items": {
|
||||
"type": "integer"
|
||||
},
|
||||
"title": "Coordinates",
|
||||
"type": "array"
|
||||
},
|
||||
"times": {
|
||||
"default": 1,
|
||||
"title": "Times",
|
||||
"type": "integer"
|
||||
},
|
||||
"delay_ms": {
|
||||
"default": 100,
|
||||
"title": "Delay Ms",
|
||||
"type": "integer"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"coordinates"
|
||||
],
|
||||
"title": "tapArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"long_press_on": {
|
||||
"description": "Long presses on the screen at coordinates.\n\n'coordinates' is a list [x, y].\n'duration' is in milliseconds (default 1000).\n",
|
||||
"inputSchema": {
|
||||
"properties": {
|
||||
"coordinates": {
|
||||
"items": {
|
||||
"type": "integer"
|
||||
},
|
||||
"title": "Coordinates",
|
||||
"type": "array"
|
||||
},
|
||||
"duration": {
|
||||
"default": 1000,
|
||||
"title": "Duration",
|
||||
"type": "integer"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"coordinates"
|
||||
],
|
||||
"title": "long_press_onArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"swipe": {
|
||||
"description": "Swipes from start coordinates to end coordinates.\n\n'coordinates' is a list [start_x, start_y, end_x, end_y].\n'duration' is in milliseconds (default 400).\n\nSet duration >= 1000 to drag-and-drop. Drag slightly past the target\nposition to trigger reordering.\n",
|
||||
"inputSchema": {
|
||||
"properties": {
|
||||
"coordinates": {
|
||||
"items": {
|
||||
"type": "integer"
|
||||
},
|
||||
"title": "Coordinates",
|
||||
"type": "array"
|
||||
},
|
||||
"duration": {
|
||||
"default": 400,
|
||||
"title": "Duration",
|
||||
"type": "integer"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"coordinates"
|
||||
],
|
||||
"title": "swipeArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"back": {
|
||||
"description": "Simulates pressing the system back button.",
|
||||
"inputSchema": {
|
||||
"properties": {},
|
||||
"title": "backArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"launch_app": {
|
||||
"description": "Launches an application by its Android package name with retries and smart polling.",
|
||||
"inputSchema": {
|
||||
"properties": {
|
||||
"package_name": {
|
||||
"title": "Package Name",
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"package_name"
|
||||
],
|
||||
"title": "launch_appArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"stop_app": {
|
||||
"description": "Force stops an application by its Android package name.",
|
||||
"inputSchema": {
|
||||
"properties": {
|
||||
"package_name": {
|
||||
"title": "Package Name",
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"package_name"
|
||||
],
|
||||
"title": "stop_appArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"open_link": {
|
||||
"description": "Opens a URL or deep link on the device.",
|
||||
"inputSchema": {
|
||||
"properties": {
|
||||
"url": {
|
||||
"title": "Url",
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"url"
|
||||
],
|
||||
"title": "open_linkArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"focus_and_input_text": {
|
||||
"description": "Focuses on a UI element at coordinates and inputs text.\n\n'coordinates' is a list [x, y] to tap first to gain focus.\n'clear_before_input' if True, clears all existing text before typing. If False, appends text at the end of existing content.\n'text' supports multi-line content with '\\n'.\n",
|
||||
"inputSchema": {
|
||||
"properties": {
|
||||
"coordinates": {
|
||||
"items": {
|
||||
"type": "integer"
|
||||
},
|
||||
"title": "Coordinates",
|
||||
"type": "array"
|
||||
},
|
||||
"text": {
|
||||
"title": "Text",
|
||||
"type": "string"
|
||||
},
|
||||
"clear_before_input": {
|
||||
"default": false,
|
||||
"title": "Clear Before Input",
|
||||
"type": "boolean"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"coordinates",
|
||||
"text"
|
||||
],
|
||||
"title": "focus_and_input_textArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"focus_and_clear_text": {
|
||||
"description": "Focuses on a UI element at coordinates and clears its text content.\n\n'coordinates' is a list [x, y] to tap first to gain focus.\n",
|
||||
"inputSchema": {
|
||||
"properties": {
|
||||
"coordinates": {
|
||||
"items": {
|
||||
"type": "integer"
|
||||
},
|
||||
"title": "Coordinates",
|
||||
"type": "array"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"coordinates"
|
||||
],
|
||||
"title": "focus_and_clear_textArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"erase_one_char": {
|
||||
"description": "Erases a single character (simulates Backspace).",
|
||||
"inputSchema": {
|
||||
"properties": {},
|
||||
"title": "erase_one_charArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"press_key": {
|
||||
"description": "Presses a specific Android key event (e.g., KEYCODE_ENTER, KEYCODE_HOME).",
|
||||
"inputSchema": {
|
||||
"properties": {
|
||||
"keycode": {
|
||||
"title": "Keycode",
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"keycode"
|
||||
],
|
||||
"title": "press_keyArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"take_screenshot": {
|
||||
"description": "Takes a screenshot of the device screen.\n\nReturns the screenshot as a base64 encoded JPEG string.\n",
|
||||
"inputSchema": {
|
||||
"properties": {},
|
||||
"title": "take_screenshotArguments",
|
||||
"type": "object"
|
||||
}
|
||||
},
|
||||
"get_ui_hierarchy": {
|
||||
"description": "Retrieves the current UI elements hierarchy from the device.",
|
||||
"inputSchema": {
|
||||
"properties": {},
|
||||
"title": "get_ui_hierarchyArguments",
|
||||
"type": "object"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -210,15 +210,14 @@ def test_operator_verbs_lower_onto_manifest_actions():
|
||||
|
||||
|
||||
def test_registry_wait_for_delay_seconds_drift_stays_dead():
|
||||
"""Guards the removal of the duplicate device-action registration.
|
||||
"""Guards against re-growing a shadow device-action registration.
|
||||
|
||||
``artemis/tools/actions/device_actions.py`` used to shadow-register a second
|
||||
``wait_for_delay`` taking *seconds* while every prompt teaches milliseconds. The
|
||||
registry (a direct-execution SDK surface, artemis/tools/mobile/exec_tools.py)
|
||||
must never again carry a ``wait_for_delay`` whose unit disagrees with the
|
||||
manifest's ``time_in_ms``.
|
||||
``wait_for_delay`` taking *seconds* while every prompt teaches milliseconds
|
||||
(the last such surface, ``artemis/tools/mobile/exec_tools.py``, is deleted).
|
||||
The ToolRegistry must never again carry a ``wait_for_delay`` whose unit
|
||||
disagrees with the manifest's ``time_in_ms``.
|
||||
"""
|
||||
import artemis.tools.mobile.exec_tools # noqa: F401 - registration side effect
|
||||
from artemis.tools.base import ToolRegistry
|
||||
|
||||
tool = ToolRegistry.get("wait_for_delay")
|
||||
|
||||
@@ -0,0 +1,227 @@
|
||||
# Copyright 2026 Google LLC
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""Consistency contract between the stdio ADB MCP server and the action manifest.
|
||||
|
||||
``artemis/mcp/adb_server.py`` deliberately keeps its historical *pixel-space*
|
||||
contract for external MCP clients (absolute pixel coordinates, combined
|
||||
``coordinates`` list parameters, ``KEYCODE_*`` vocabulary), so its FastMCP tools
|
||||
are hand-written rather than generated from ``artemis.mcp.action_specs``. This
|
||||
suite is the compensating contract:
|
||||
|
||||
* **Fixture pin** -- the full generated schema of every adb_server tool is
|
||||
frozen; any signature or docstring change must regenerate the fixture in the
|
||||
same commit (external clients see this schema directly).
|
||||
* **Projection check** -- every device-action tool is mapped onto its canonical
|
||||
``ActionSpec`` wire dialect. Parameters that mirror a wire parameter must
|
||||
agree on JSON type, requiredness, and default. Deliberate legacy divergences
|
||||
are declared in ``DECLARED_DIVERGENCES`` and are *ratcheted*: if the two
|
||||
surfaces converge, the stale declaration fails too, so the table always
|
||||
matches reality.
|
||||
|
||||
Together the two guards make silent drift between the second declaration
|
||||
surface and the canonical manifest impossible in either direction.
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from artemis.mcp.action_specs import ACTION_SPECS, ParamSpec
|
||||
|
||||
FIXTURES = Path(__file__).resolve().parents[2] / "fixtures" / "action_surfaces"
|
||||
|
||||
FIXTURE_HINT = (
|
||||
"The generated schema differs from the pinned fixture. If this change is"
|
||||
" intentional, regenerate the fixture in the same commit and explain the schema"
|
||||
" change in the commit message; external MCP clients see this schema directly."
|
||||
)
|
||||
|
||||
|
||||
# --- The declared projection ----------------------------------------------------------
|
||||
|
||||
#: adb_server tool name -> (canonical action name, {adb param -> wire param}).
|
||||
#: Params absent from the mapping are legacy packaging with no one-to-one wire
|
||||
#: counterpart (the combined ``coordinates`` lists); their shape is asserted
|
||||
#: separately below and pinned by the fixture.
|
||||
ADB_TOOL_PROJECTION: dict[str, tuple[str, dict[str, str]]] = {
|
||||
"tap": ("click", {"times": "times", "delay_ms": "delay_ms"}),
|
||||
"long_press_on": ("long_press", {"duration": "duration_ms"}),
|
||||
"swipe": ("swipe", {"duration": "duration_ms"}),
|
||||
"focus_and_input_text": ("input_text", {"text": "text", "clear_before_input": "clear_exist"}),
|
||||
"press_key": ("press_key", {"keycode": "key"}),
|
||||
"open_link": ("open_link", {"url": "url"}),
|
||||
"erase_one_char": ("erase_one_char", {}),
|
||||
"focus_and_clear_text": ("focus_and_clear_text", {}),
|
||||
}
|
||||
|
||||
#: Combined-coordinates legacy packaging: adb tool -> (marker in docstring, arity).
|
||||
#: ``coordinates`` packs the wire dialect's coordinate pair(s) into one list.
|
||||
COORDINATE_PACKING: dict[str, tuple[str, int]] = {
|
||||
"tap": ("[x, y]", 2),
|
||||
"long_press_on": ("[x, y]", 2),
|
||||
"swipe": ("[start_x, start_y, end_x, end_y]", 4),
|
||||
"focus_and_input_text": ("[x, y]", 2),
|
||||
"focus_and_clear_text": ("[x, y]", 2),
|
||||
}
|
||||
|
||||
#: Deliberate legacy divergences, (adb tool, adb param) -> reason. Each entry is
|
||||
#: ratcheted: the test fails if the surfaces actually agree, so this table can
|
||||
#: never go stale.
|
||||
DECLARED_DIVERGENCES: dict[tuple[str, str], str] = {
|
||||
("swipe", "duration"): (
|
||||
"legacy stdio default is 400ms; the canonical wire duration_ms defaults"
|
||||
" to 800ms (adaptive-duration smart swipes only exist in agent dialects)"
|
||||
),
|
||||
("focus_and_input_text", "clear_before_input"): (
|
||||
"legacy stdio appends by default (False); the canonical wire clear_exist"
|
||||
" defaults to True (replace)"
|
||||
),
|
||||
}
|
||||
|
||||
_JSON_TYPE: dict[type, str] = {int: "integer", str: "string", bool: "boolean"}
|
||||
|
||||
|
||||
def _wire_param(action: str, name: str) -> ParamSpec:
|
||||
spec = ACTION_SPECS[action]
|
||||
assert spec.wire is not None, f"'{action}' has no wire dialect."
|
||||
for p in spec.wire.params:
|
||||
if p.name == name:
|
||||
return p
|
||||
raise AssertionError(f"wire dialect of '{action}' has no param '{name}'")
|
||||
|
||||
|
||||
async def _adb_server_tools() -> dict:
|
||||
from artemis.mcp import adb_server
|
||||
|
||||
tools = await adb_server.mcp.list_tools()
|
||||
return {t.name: t for t in tools}
|
||||
|
||||
|
||||
def _prop(tool, name: str) -> dict:
|
||||
props = tool.inputSchema.get("properties", {})
|
||||
assert name in props, f"tool '{tool.name}' lost parameter '{name}'"
|
||||
return props[name]
|
||||
|
||||
|
||||
# --- Fixture pin -----------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_adb_server_manifest_matches_fixture():
|
||||
expected = json.loads(
|
||||
(FIXTURES / "adb_server_manifest.json").read_text(encoding="utf-8")
|
||||
)
|
||||
tools = await _adb_server_tools()
|
||||
generated = {
|
||||
name: {"description": t.description, "inputSchema": t.inputSchema}
|
||||
for name, t in tools.items()
|
||||
}
|
||||
assert set(generated) == set(expected), FIXTURE_HINT
|
||||
for name in expected:
|
||||
assert generated[name] == expected[name], f"adb_server tool '{name}': {FIXTURE_HINT}"
|
||||
|
||||
|
||||
# --- Projection consistency -------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_projection_covers_all_wire_backed_adb_tools():
|
||||
"""Every adb_server tool that mirrors a canonical action is in the projection."""
|
||||
tools = await _adb_server_tools()
|
||||
missing = set(ADB_TOOL_PROJECTION) - set(tools)
|
||||
assert not missing, f"projection references adb_server tools that no longer exist: {missing}"
|
||||
for adb_name, (canonical, _) in ADB_TOOL_PROJECTION.items():
|
||||
assert canonical in ACTION_SPECS, (
|
||||
f"adb_server tool '{adb_name}' maps to '{canonical}', which the canonical"
|
||||
" manifest does not know."
|
||||
)
|
||||
assert ACTION_SPECS[canonical].wire is not None, (
|
||||
f"adb_server tool '{adb_name}' maps to '{canonical}', which has no wire dialect."
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mirrored_params_agree_with_wire_dialect():
|
||||
"""Type/requiredness/default of every mirrored parameter matches the wire spec,
|
||||
except where a divergence is declared -- and declared divergences must be real."""
|
||||
tools = await _adb_server_tools()
|
||||
for adb_name, (canonical, param_map) in ADB_TOOL_PROJECTION.items():
|
||||
tool = tools[adb_name]
|
||||
required = set(tool.inputSchema.get("required", []))
|
||||
for adb_param, wire_name in param_map.items():
|
||||
wire = _wire_param(canonical, wire_name)
|
||||
prop = _prop(tool, adb_param)
|
||||
expected_type = _JSON_TYPE.get(wire.annotation)
|
||||
if expected_type is not None:
|
||||
assert prop.get("type") == expected_type, (
|
||||
f"'{adb_name}.{adb_param}': schema type {prop.get('type')!r} !="
|
||||
f" canonical wire '{canonical}.{wire_name}' type {expected_type!r}"
|
||||
)
|
||||
adb_required = adb_param in required
|
||||
adb_default = prop.get("default")
|
||||
diverged_key = (adb_name, adb_param)
|
||||
agrees = adb_required == wire.required and (
|
||||
wire.required or adb_default == wire.default
|
||||
)
|
||||
if diverged_key in DECLARED_DIVERGENCES:
|
||||
assert not agrees, (
|
||||
f"'{adb_name}.{adb_param}' is declared divergent from"
|
||||
f" '{canonical}.{wire_name}' but the surfaces now agree -- remove"
|
||||
f" the stale entry from DECLARED_DIVERGENCES:"
|
||||
f" {DECLARED_DIVERGENCES[diverged_key]}"
|
||||
)
|
||||
else:
|
||||
assert agrees, (
|
||||
f"'{adb_name}.{adb_param}' drifted from canonical wire"
|
||||
f" '{canonical}.{wire_name}': required {adb_required} vs"
|
||||
f" {wire.required}, default {adb_default!r} vs {wire.default!r}."
|
||||
" Either re-align, or declare the divergence in"
|
||||
" DECLARED_DIVERGENCES with a reason."
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_coordinate_packing_shape_is_documented():
|
||||
"""The combined `coordinates` legacy parameter keeps its documented arity."""
|
||||
tools = await _adb_server_tools()
|
||||
for adb_name, (marker, _arity) in COORDINATE_PACKING.items():
|
||||
tool = tools[adb_name]
|
||||
prop = _prop(tool, "coordinates")
|
||||
assert prop.get("type") == "array", f"'{adb_name}.coordinates' must be an array"
|
||||
assert prop.get("items", {}).get("type") == "integer", (
|
||||
f"'{adb_name}.coordinates' must be a list of integers"
|
||||
)
|
||||
assert "coordinates" in set(tool.inputSchema.get("required", [])), (
|
||||
f"'{adb_name}.coordinates' must stay required"
|
||||
)
|
||||
assert marker in (tool.description or ""), (
|
||||
f"'{adb_name}' docstring no longer documents the coordinate shape"
|
||||
f" '{marker}'; external clients rely on this wording."
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_press_key_vocabulary_divergence_is_documented():
|
||||
"""adb_server's press_key takes raw KEYCODE_* strings; the canonical wire takes
|
||||
bare lowercase words. The canonical spec documents that divergence in prose."""
|
||||
tools = await _adb_server_tools()
|
||||
desc = tools["press_key"].description or ""
|
||||
assert "KEYCODE_" in desc, (
|
||||
"adb_server press_key docstring must keep teaching KEYCODE_* vocabulary"
|
||||
)
|
||||
assert "key" in ACTION_SPECS["press_key"].differences, (
|
||||
"the canonical manifest no longer documents the press_key vocabulary split"
|
||||
)
|
||||
@@ -12,80 +12,33 @@
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""Unit tests for Dynamic Model Router."""
|
||||
"""Unit tests for ModelFactory instantiation and grounding tool wiring."""
|
||||
|
||||
from artemis.llm.router import ModelEndpoint, ModelProvider, ModelRouter
|
||||
from artemis.llm.router import ModelEndpoint, ModelFactory, ModelProvider
|
||||
|
||||
|
||||
def test_model_router_default_routes():
|
||||
"""Verify default role routing returns expected models."""
|
||||
router = ModelRouter()
|
||||
|
||||
# Planner route
|
||||
planner_ep = router.get_endpoint("planner")
|
||||
assert planner_ep.provider == ModelProvider.GEMINI
|
||||
assert "pro" in planner_ep.model_name
|
||||
|
||||
# Operator route
|
||||
operator_ep = router.get_endpoint("operator")
|
||||
assert operator_ep.provider == ModelProvider.GEMINI
|
||||
assert "flash" in operator_ep.model_name
|
||||
|
||||
|
||||
def test_model_router_custom_route_override():
|
||||
"""Verify dynamic role route override."""
|
||||
router = ModelRouter()
|
||||
custom_ep = ModelEndpoint(
|
||||
provider=ModelProvider.OPENAI,
|
||||
model_name="gpt-4o",
|
||||
api_base="https://custom.api.endpoint/v1",
|
||||
)
|
||||
router.set_role_route("operator", custom_ep)
|
||||
|
||||
ep = router.get_endpoint("operator")
|
||||
assert ep.provider == ModelProvider.OPENAI
|
||||
assert ep.model_name == "gpt-4o"
|
||||
assert ep.api_base == "https://custom.api.endpoint/v1"
|
||||
|
||||
|
||||
def test_model_router_fallbacks():
|
||||
"""Verify fallback endpoint configuration."""
|
||||
router = ModelRouter()
|
||||
fallbacks = [
|
||||
ModelEndpoint(provider=ModelProvider.GEMINI, model_name="gemini-2.5-flash"),
|
||||
ModelEndpoint(provider=ModelProvider.OPENAI, model_name="gpt-4o-mini"),
|
||||
]
|
||||
router.set_role_route("custom_worker", "gemini-2.5-pro", fallbacks=fallbacks)
|
||||
fb = router.get_fallbacks("custom_worker")
|
||||
assert len(fb) == 2
|
||||
assert fb[0].model_name == "gemini-2.5-flash"
|
||||
assert fb[1].model_name == "gpt-4o-mini"
|
||||
|
||||
|
||||
def test_model_router_anthropic_instantiation():
|
||||
def test_model_factory_anthropic_instantiation():
|
||||
"""Verify Anthropic Claude model instantiation with thinking budget."""
|
||||
router = ModelRouter()
|
||||
anth_ep = ModelEndpoint(
|
||||
provider=ModelProvider.ANTHROPIC,
|
||||
model_name="claude-3-7-sonnet-20250219",
|
||||
api_key="sk-ant-test-key",
|
||||
reasoning_effort="high",
|
||||
)
|
||||
model = router.instantiate_model(anth_ep)
|
||||
model = ModelFactory.create_model(anth_ep)
|
||||
assert model.model == "claude-3-7-sonnet-20250219"
|
||||
assert getattr(model, "thinking", None) == {"type": "enabled", "budget_tokens": 32768}
|
||||
|
||||
|
||||
def test_model_router_openai_instantiation():
|
||||
def test_model_factory_openai_instantiation():
|
||||
"""Verify OpenAI model instantiation with reasoning effort."""
|
||||
router = ModelRouter()
|
||||
oai_ep = ModelEndpoint(
|
||||
provider=ModelProvider.OPENAI,
|
||||
model_name="o3-mini",
|
||||
api_key="sk-test-key",
|
||||
reasoning_effort="medium",
|
||||
)
|
||||
model = router.instantiate_model(oai_ep)
|
||||
model = ModelFactory.create_model(oai_ep)
|
||||
assert model.model_name == "o3-mini"
|
||||
assert getattr(model, "reasoning_effort", None) == "medium"
|
||||
|
||||
@@ -1,70 +0,0 @@
|
||||
# Copyright 2026 Google LLC
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""Unit tests for Universal Tool Protocol and Actions."""
|
||||
|
||||
import pytest
|
||||
from artemis.drivers.mock.mock_driver import MockDeviceDriver
|
||||
from artemis.tools.base import ToolRegistry
|
||||
import artemis.tools.mobile.exec_tools # noqa: F401 - Register action tools
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_registry_registration():
|
||||
"""Verify standard device actions are registered in ToolRegistry."""
|
||||
click_tool = ToolRegistry.get("click")
|
||||
assert click_tool is not None
|
||||
assert click_tool.name == "click"
|
||||
assert click_tool.category == "action"
|
||||
|
||||
swipe_tool = ToolRegistry.get("swipe")
|
||||
assert swipe_tool is not None
|
||||
assert swipe_tool.name == "swipe"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_tool_execution_with_driver():
|
||||
"""Verify tool execution correctly dispatches to MockDeviceDriver."""
|
||||
driver = MockDeviceDriver(width=1000, height=2000)
|
||||
|
||||
# Click normalized [500, 500] -> pixel (500, 1000)
|
||||
result = await ToolRegistry.execute("click", {"target": [500, 500]}, driver=driver, ctx=None)
|
||||
assert "Click" in result or "Clicked" in result
|
||||
assert driver.action_history[-1]["action"] == "tap"
|
||||
assert driver.action_history[-1]["x"] == 500
|
||||
assert driver.action_history[-1]["y"] == 1000
|
||||
|
||||
# Swipe cardinal "up"
|
||||
result = await ToolRegistry.execute("swipe", {"action": "up"}, driver=driver, ctx=None)
|
||||
assert "Swipe" in result or "Swiped" in result
|
||||
assert driver.action_history[-1]["action"] == "swipe_direction"
|
||||
|
||||
|
||||
def test_tool_to_genai_declaration():
|
||||
"""Verify tools export to Google GenAI FunctionDeclaration properly."""
|
||||
declarations = ToolRegistry.get_genai_declarations(["click", "swipe", "input_text"])
|
||||
assert len(declarations) == 3
|
||||
|
||||
decl_names = [d.name for d in declarations]
|
||||
assert "click" in decl_names
|
||||
assert "swipe" in decl_names
|
||||
assert "input_text" in decl_names
|
||||
|
||||
|
||||
def test_tool_to_langchain_tool():
|
||||
"""Verify tools export to LangChain BaseTool."""
|
||||
lc_tools = ToolRegistry.get_langchain_tools(ctx=None, names=["click", "press_key"])
|
||||
assert len(lc_tools) == 2
|
||||
assert lc_tools[0].name == "click"
|
||||
assert lc_tools[1].name == "press_key"
|
||||
@@ -1,148 +0,0 @@
|
||||
# Copyright 2026 Google LLC
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from artemis.context import ArtemisContext
|
||||
from artemis.drivers.base import BaseDeviceDriver
|
||||
from artemis.graph.state import State
|
||||
from artemis.tools.base import ArtemisTool, ToolRegistry
|
||||
from artemis.tools.mobile.exec_tools import (
|
||||
Click,
|
||||
ClickArgs,
|
||||
ClickTool,
|
||||
click,
|
||||
click_wrapper,
|
||||
get_click_tool,
|
||||
)
|
||||
from langchain_core.messages import ToolMessage
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_ctx():
|
||||
ctx = MagicMock(spec=ArtemisContext)
|
||||
ctx.device = MagicMock()
|
||||
ctx.device.device_width = 1080
|
||||
ctx.device.device_height = 2400
|
||||
return ctx
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_driver():
|
||||
driver = MagicMock(spec=BaseDeviceDriver)
|
||||
driver.screen_size = (1080, 2400)
|
||||
driver.tap = AsyncMock(return_value=True)
|
||||
return driver
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_state():
|
||||
return MagicMock(spec=State)
|
||||
|
||||
|
||||
def test_click_tool_subclass_and_registry():
|
||||
"""Verify ClickTool is a subclass of ArtemisTool and properly registered."""
|
||||
assert issubclass(ClickTool, ArtemisTool)
|
||||
assert issubclass(Click, ArtemisTool)
|
||||
assert isinstance(click, ArtemisTool)
|
||||
assert isinstance(click, ClickTool)
|
||||
|
||||
assert click.name == "click"
|
||||
assert click.category == "action"
|
||||
assert click.args_schema == ClickArgs
|
||||
|
||||
# Registry lookup
|
||||
reg_tool = ToolRegistry.get("click")
|
||||
assert reg_tool is not None
|
||||
assert isinstance(reg_tool, ClickTool)
|
||||
|
||||
# GenAI FunctionDeclaration export
|
||||
declaration = click.to_genai_declaration()
|
||||
assert declaration.name == "click"
|
||||
assert "target" in declaration.parameters.properties
|
||||
|
||||
# Wrapper check
|
||||
assert click_wrapper is not None
|
||||
assert click_wrapper.tool_fn_getter == get_click_tool
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_click_direct_execution_with_driver(mock_driver):
|
||||
"""Verify direct execution of ClickTool with BaseDeviceDriver."""
|
||||
result = await click.execute(driver=mock_driver, target=[500, 500])
|
||||
mock_driver.tap.assert_called_once_with(540, 1200)
|
||||
assert "Clicked at [500, 500] (normalized) successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_click_direct_execution_with_ctx(mock_ctx):
|
||||
"""Verify direct execution of ClickTool with ArtemisContext."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None))
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
result = await click.execute(ctx=mock_ctx, target=[250, 750])
|
||||
mock_controller_inst.tap_at.assert_called_once_with(270, 1800)
|
||||
assert "Clicked at [250, 750] (normalized) successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_click_with_state_command(mock_ctx, mock_state):
|
||||
"""Verify ClickTool returns ToolMessage when state is provided."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None))
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
cmd = await click.execute(
|
||||
ctx=mock_ctx,
|
||||
target=[100, 200],
|
||||
tool_call_id="call_click_1",
|
||||
state=mock_state,
|
||||
)
|
||||
assert isinstance(cmd, ToolMessage)
|
||||
assert cmd.tool_call_id == "call_click_1"
|
||||
assert cmd.status == "success"
|
||||
assert "Clicked at [100, 200]" in cmd.content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_click_callable_execution(mock_driver):
|
||||
"""Verify invoking click directly as a callable."""
|
||||
result = await click(driver=mock_driver, target=[100, 100])
|
||||
assert "Clicked at [100, 100] (normalized) successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_click_invalid_target(mock_ctx):
|
||||
"""Verify error handling on invalid target coordinates."""
|
||||
result = await click.execute(ctx=mock_ctx, target=[])
|
||||
assert "Error during click: Invalid target coordinates" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_click_tool_langchain_ainvoke(mock_ctx):
|
||||
"""Verify get_click_tool exports a LangChain tool that works with ainvoke."""
|
||||
click_tool = get_click_tool(mock_ctx)
|
||||
assert click_tool.name == "click"
|
||||
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None))
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
result = await click_tool.ainvoke({"target": [500, 500]})
|
||||
assert "Clicked at [500, 500] (normalized) successfully." in result
|
||||
@@ -1,172 +0,0 @@
|
||||
# Copyright 2026 Google LLC
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from artemis.context import ArtemisContext
|
||||
from artemis.drivers.base import BaseDeviceDriver
|
||||
from artemis.graph.state import State
|
||||
from artemis.tools.base import ArtemisTool, ToolRegistry
|
||||
from artemis.tools.mobile.exec_tools import (
|
||||
InputText,
|
||||
InputTextArgs,
|
||||
InputTextTool,
|
||||
get_input_text_tool,
|
||||
input_text,
|
||||
input_text_wrapper,
|
||||
)
|
||||
from langchain_core.messages import ToolMessage
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_ctx():
|
||||
ctx = MagicMock(spec=ArtemisContext)
|
||||
ctx.device = MagicMock()
|
||||
ctx.device.device_width = 1080
|
||||
ctx.device.device_height = 2400
|
||||
return ctx
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_driver():
|
||||
driver = MagicMock(spec=BaseDeviceDriver)
|
||||
driver.screen_size = (1080, 2400)
|
||||
driver.tap = AsyncMock(return_value=True)
|
||||
driver.input_text = AsyncMock(return_value=True)
|
||||
return driver
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_state():
|
||||
return MagicMock(spec=State)
|
||||
|
||||
|
||||
def test_input_text_tool_subclass_and_registry():
|
||||
"""Verify InputTextTool is a subclass of ArtemisTool and properly registered."""
|
||||
assert issubclass(InputTextTool, ArtemisTool)
|
||||
assert issubclass(InputText, ArtemisTool)
|
||||
assert isinstance(input_text, ArtemisTool)
|
||||
assert isinstance(input_text, InputTextTool)
|
||||
|
||||
assert input_text.name == "input_text"
|
||||
assert input_text.category == "action"
|
||||
assert input_text.args_schema == InputTextArgs
|
||||
|
||||
# Registry lookup
|
||||
reg_tool = ToolRegistry.get("input_text")
|
||||
assert reg_tool is not None
|
||||
assert isinstance(reg_tool, InputTextTool)
|
||||
|
||||
# GenAI FunctionDeclaration export
|
||||
declaration = input_text.to_genai_declaration()
|
||||
assert declaration.name == "input_text"
|
||||
assert "text" in declaration.parameters.properties
|
||||
assert "target" in declaration.parameters.properties
|
||||
assert "clear_exist" in declaration.parameters.properties
|
||||
|
||||
# Wrapper check
|
||||
assert input_text_wrapper is not None
|
||||
assert input_text_wrapper.tool_fn_getter == get_input_text_tool
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_input_text_direct_execution_with_driver(mock_driver):
|
||||
"""Verify direct execution of InputTextTool with BaseDeviceDriver."""
|
||||
result = await input_text.execute(
|
||||
driver=mock_driver, text="hello world", target=[500, 500], clear_exist=True
|
||||
)
|
||||
mock_driver.tap.assert_called_once_with(540, 1200)
|
||||
mock_driver.input_text.assert_called_once_with("hello world", clear_existing=True)
|
||||
assert "Typed 'hello world' successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_input_text_direct_execution_with_ctx(mock_ctx):
|
||||
"""Verify direct execution of InputTextTool with ArtemisContext."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None))
|
||||
mock_controller_inst.erase_text = AsyncMock(return_value=True)
|
||||
mock_controller_inst.type_text = AsyncMock(return_value=True)
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
result = await input_text.execute(
|
||||
ctx=mock_ctx, text="test input", target=[250, 750], clear_exist=True
|
||||
)
|
||||
mock_controller_inst.tap_at.assert_called_once_with(270, 1800)
|
||||
mock_controller_inst.erase_text.assert_called_once()
|
||||
mock_controller_inst.type_text.assert_called_once_with("test input", clear_existing=False)
|
||||
assert "Typed 'test input' successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_input_text_with_state_command(mock_ctx, mock_state):
|
||||
"""Verify InputTextTool returns ToolMessage when state is provided."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None))
|
||||
mock_controller_inst.erase_text = AsyncMock(return_value=True)
|
||||
mock_controller_inst.type_text = AsyncMock(return_value=True)
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
cmd = await input_text.execute(
|
||||
ctx=mock_ctx,
|
||||
text="query",
|
||||
target=[100, 200],
|
||||
tool_call_id="call_it_1",
|
||||
state=mock_state,
|
||||
)
|
||||
assert isinstance(cmd, ToolMessage)
|
||||
assert cmd.tool_call_id == "call_it_1"
|
||||
assert cmd.status == "success"
|
||||
assert "Typed 'query'" in cmd.content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_input_text_callable_execution(mock_driver):
|
||||
"""Verify invoking input_text directly as a callable."""
|
||||
result = await input_text(driver=mock_driver, text="call text", target=[100, 100])
|
||||
assert "Typed 'call text' successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_input_text_invalid_target(mock_ctx):
|
||||
"""Verify error handling on invalid target coordinates."""
|
||||
result = await input_text.execute(ctx=mock_ctx, text="bad", target=[])
|
||||
assert "Error during input text: Invalid target coordinates" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_input_text_tool_langchain_ainvoke(mock_ctx):
|
||||
"""Verify get_input_text_tool exports a LangChain tool that works with ainvoke."""
|
||||
it_tool = get_input_text_tool(mock_ctx)
|
||||
assert it_tool.name == "input_text"
|
||||
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None))
|
||||
mock_controller_inst.press_key = AsyncMock(return_value=True)
|
||||
mock_controller_inst.type_text = AsyncMock(return_value=True)
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
result = await it_tool.ainvoke(
|
||||
{
|
||||
"text": "abc",
|
||||
"target": [500, 500],
|
||||
"clear_exist": False,
|
||||
}
|
||||
)
|
||||
mock_controller_inst.press_key.assert_called_once_with("123")
|
||||
assert "Typed 'abc' successfully." in result
|
||||
@@ -1,152 +0,0 @@
|
||||
# Copyright 2026 Google LLC
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from artemis.context import ArtemisContext
|
||||
from artemis.drivers.base import BaseDeviceDriver
|
||||
from artemis.graph.state import State
|
||||
from artemis.tools.base import ArtemisTool, ToolRegistry
|
||||
from artemis.tools.mobile.exec_tools import (
|
||||
LongPress,
|
||||
LongPressArgs,
|
||||
LongPressTool,
|
||||
get_long_press_tool,
|
||||
long_press,
|
||||
long_press_wrapper,
|
||||
)
|
||||
from langchain_core.messages import ToolMessage
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_ctx():
|
||||
ctx = MagicMock(spec=ArtemisContext)
|
||||
ctx.device = MagicMock()
|
||||
ctx.device.device_width = 1080
|
||||
ctx.device.device_height = 2400
|
||||
return ctx
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_driver():
|
||||
driver = MagicMock(spec=BaseDeviceDriver)
|
||||
driver.screen_size = (1080, 2400)
|
||||
driver.long_press = AsyncMock(return_value=True)
|
||||
return driver
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_state():
|
||||
return MagicMock(spec=State)
|
||||
|
||||
|
||||
def test_long_press_tool_subclass_and_registry():
|
||||
"""Verify LongPressTool is a subclass of ArtemisTool and properly registered."""
|
||||
assert issubclass(LongPressTool, ArtemisTool)
|
||||
assert issubclass(LongPress, ArtemisTool)
|
||||
assert isinstance(long_press, ArtemisTool)
|
||||
assert isinstance(long_press, LongPressTool)
|
||||
|
||||
assert long_press.name == "long_press"
|
||||
assert long_press.category == "action"
|
||||
assert long_press.args_schema == LongPressArgs
|
||||
|
||||
# Registry lookup
|
||||
reg_tool = ToolRegistry.get("long_press")
|
||||
assert reg_tool is not None
|
||||
assert isinstance(reg_tool, LongPressTool)
|
||||
|
||||
# GenAI FunctionDeclaration export
|
||||
declaration = long_press.to_genai_declaration()
|
||||
assert declaration.name == "long_press"
|
||||
assert "target" in declaration.parameters.properties
|
||||
assert "duration" in declaration.parameters.properties
|
||||
|
||||
# Wrapper check
|
||||
assert long_press_wrapper is not None
|
||||
assert long_press_wrapper.tool_fn_getter == get_long_press_tool
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_long_press_direct_execution_with_driver(mock_driver):
|
||||
"""Verify direct execution of LongPressTool with BaseDeviceDriver."""
|
||||
result = await long_press.execute(driver=mock_driver, target=[500, 500], duration=1500)
|
||||
mock_driver.long_press.assert_called_once_with(540, 1200, duration_ms=1500)
|
||||
assert "Long pressed at [500, 500] (normalized) successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_long_press_direct_execution_with_ctx(mock_ctx):
|
||||
"""Verify direct execution of LongPressTool with ArtemisContext."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None))
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
result = await long_press.execute(ctx=mock_ctx, target=[250, 750], duration=2000)
|
||||
mock_controller_inst.tap_at.assert_called_once_with(
|
||||
270, 1800, long_press=True, long_press_duration=2000
|
||||
)
|
||||
assert "Long pressed at [250, 750] (normalized) successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_long_press_with_state_command(mock_ctx, mock_state):
|
||||
"""Verify LongPressTool returns ToolMessage when state is provided."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None))
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
cmd = await long_press.execute(
|
||||
ctx=mock_ctx,
|
||||
target=[100, 200],
|
||||
duration=1200,
|
||||
tool_call_id="call_lp_1",
|
||||
state=mock_state,
|
||||
)
|
||||
assert isinstance(cmd, ToolMessage)
|
||||
assert cmd.tool_call_id == "call_lp_1"
|
||||
assert cmd.status == "success"
|
||||
assert "Long pressed at [100, 200]" in cmd.content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_long_press_callable_execution(mock_driver):
|
||||
"""Verify invoking long_press directly as a callable."""
|
||||
result = await long_press(driver=mock_driver, target=[100, 100])
|
||||
assert "Long pressed at [100, 100] (normalized) successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_long_press_invalid_target(mock_ctx):
|
||||
"""Verify error handling on invalid target coordinates."""
|
||||
result = await long_press.execute(ctx=mock_ctx, target=[])
|
||||
assert "Error during long press: Invalid target coordinates" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_long_press_tool_langchain_ainvoke(mock_ctx):
|
||||
"""Verify get_long_press_tool exports a LangChain tool that works with ainvoke."""
|
||||
lp_tool = get_long_press_tool(mock_ctx)
|
||||
assert lp_tool.name == "long_press"
|
||||
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.tap_at = AsyncMock(return_value=MagicMock(error=None))
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
result = await lp_tool.ainvoke({"target": [500, 500], "duration": 1500})
|
||||
assert "Long pressed at [500, 500] (normalized) successfully." in result
|
||||
@@ -1,164 +0,0 @@
|
||||
# Copyright 2026 Google LLC
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from artemis.context import ArtemisContext
|
||||
from artemis.drivers.base import BaseDeviceDriver
|
||||
from artemis.graph.state import State
|
||||
from artemis.tools.base import ArtemisTool, ToolRegistry
|
||||
from artemis.tools.mobile.exec_tools import (
|
||||
PressKey,
|
||||
PressKeyArgs,
|
||||
PressKeyTool,
|
||||
get_press_key_tool,
|
||||
press_key,
|
||||
press_key_wrapper,
|
||||
)
|
||||
from langchain_core.messages import ToolMessage
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_ctx():
|
||||
ctx = MagicMock(spec=ArtemisContext)
|
||||
ctx.device = MagicMock()
|
||||
return ctx
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_driver():
|
||||
driver = MagicMock(spec=BaseDeviceDriver)
|
||||
driver.press_key = AsyncMock(return_value=True)
|
||||
return driver
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_state():
|
||||
return MagicMock(spec=State)
|
||||
|
||||
|
||||
def test_press_key_tool_subclass_and_registry():
|
||||
"""Verify PressKeyTool is a subclass of ArtemisTool and properly registered."""
|
||||
assert issubclass(PressKeyTool, ArtemisTool)
|
||||
assert issubclass(PressKey, ArtemisTool)
|
||||
assert isinstance(press_key, ArtemisTool)
|
||||
assert isinstance(press_key, PressKeyTool)
|
||||
|
||||
assert press_key.name == "press_key"
|
||||
assert press_key.category == "action"
|
||||
assert press_key.args_schema == PressKeyArgs
|
||||
|
||||
# Registry lookup
|
||||
reg_tool = ToolRegistry.get("press_key")
|
||||
assert reg_tool is not None
|
||||
assert isinstance(reg_tool, PressKeyTool)
|
||||
|
||||
# GenAI FunctionDeclaration export
|
||||
declaration = press_key.to_genai_declaration()
|
||||
assert declaration.name == "press_key"
|
||||
assert "key" in declaration.parameters.properties
|
||||
|
||||
# Wrapper check
|
||||
assert press_key_wrapper is not None
|
||||
assert press_key_wrapper.tool_fn_getter == get_press_key_tool
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_press_key_direct_execution_with_driver(mock_driver):
|
||||
"""Verify direct execution with BaseDeviceDriver."""
|
||||
result = await press_key.execute(driver=mock_driver, key="BACK")
|
||||
mock_driver.press_key.assert_called_once_with("BACK")
|
||||
assert "Pressed key 'BACK' successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_press_key_direct_execution_with_ctx(mock_ctx):
|
||||
"""Verify direct execution with ArtemisContext for various keys."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.press_enter = AsyncMock(return_value=True)
|
||||
mock_controller_inst.go_back = AsyncMock(return_value=True)
|
||||
mock_controller_inst.go_home = AsyncMock(return_value=True)
|
||||
mock_controller_inst.press_key = AsyncMock(return_value=True)
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
# Test ENTER
|
||||
res_enter = await press_key.execute(ctx=mock_ctx, key="ENTER")
|
||||
mock_controller_inst.press_enter.assert_called_once()
|
||||
assert "Pressed key 'ENTER' successfully." in res_enter
|
||||
|
||||
# Test BACK
|
||||
res_back = await press_key.execute(ctx=mock_ctx, key="BACK")
|
||||
mock_controller_inst.go_back.assert_called_once()
|
||||
assert "Pressed key 'BACK' successfully." in res_back
|
||||
|
||||
# Test HOME
|
||||
res_home = await press_key.execute(ctx=mock_ctx, key="HOME")
|
||||
mock_controller_inst.go_home.assert_called_once()
|
||||
assert "Pressed key 'HOME' successfully." in res_home
|
||||
|
||||
# Test APP_SWITCH
|
||||
res_app_switch = await press_key.execute(ctx=mock_ctx, key="APP_SWITCH")
|
||||
mock_controller_inst.press_key.assert_called_with("KEYCODE_APP_SWITCH")
|
||||
assert "Pressed key 'APP_SWITCH' successfully." in res_app_switch
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_press_key_with_state_command(mock_ctx, mock_state):
|
||||
"""Verify PressKeyTool returns ToolMessage when state is provided."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.go_home = AsyncMock(return_value=True)
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
cmd = await press_key.execute(
|
||||
ctx=mock_ctx,
|
||||
key="HOME",
|
||||
tool_call_id="call_pk_1",
|
||||
state=mock_state,
|
||||
)
|
||||
assert isinstance(cmd, ToolMessage)
|
||||
assert cmd.tool_call_id == "call_pk_1"
|
||||
assert cmd.status == "success"
|
||||
assert "Pressed key 'HOME' successfully." in cmd.content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_press_key_callable_execution(mock_driver):
|
||||
"""Verify invoking press_key directly as a callable."""
|
||||
result = await press_key(driver=mock_driver, key="ENTER")
|
||||
assert "Pressed key 'ENTER' successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_press_key_missing_key(mock_ctx):
|
||||
"""Verify error handling when key is empty or missing."""
|
||||
result = await press_key.execute(ctx=mock_ctx, key="")
|
||||
assert "Error during press key: Key parameter is required." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_press_key_tool_langchain_ainvoke(mock_ctx):
|
||||
"""Verify get_press_key_tool exports a LangChain tool that works with ainvoke."""
|
||||
pk_tool = get_press_key_tool(mock_ctx)
|
||||
assert pk_tool.name == "press_key"
|
||||
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.go_back = AsyncMock(return_value=True)
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
result = await pk_tool.ainvoke({"key": "BACK"})
|
||||
assert "Pressed key 'BACK' successfully." in result
|
||||
@@ -1,316 +0,0 @@
|
||||
# Copyright 2026 Google LLC
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from artemis.context import ArtemisContext
|
||||
from artemis.drivers.base import BaseDeviceDriver
|
||||
from artemis.graph.state import State
|
||||
from artemis.tools.base import ArtemisTool, ToolRegistry
|
||||
from artemis.tools.mobile.exec_tools import (
|
||||
Swipe,
|
||||
SwipeArgs,
|
||||
SwipeTool,
|
||||
get_swipe_tool,
|
||||
swipe,
|
||||
swipe_wrapper,
|
||||
)
|
||||
from langchain_core.messages import ToolMessage
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_ctx():
|
||||
ctx = MagicMock(spec=ArtemisContext)
|
||||
ctx.device = MagicMock()
|
||||
ctx.device.device_width = 1080
|
||||
ctx.device.device_height = 2400
|
||||
return ctx
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_driver():
|
||||
driver = MagicMock(spec=BaseDeviceDriver)
|
||||
driver.screen_size = (1080, 2400)
|
||||
driver.swipe_direction = AsyncMock(return_value=True)
|
||||
driver.swipe = AsyncMock(return_value=True)
|
||||
return driver
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_state():
|
||||
return MagicMock(spec=State)
|
||||
|
||||
|
||||
def test_swipe_tool_subclass_and_registry():
|
||||
"""Verify SwipeTool is a subclass of ArtemisTool and properly registered."""
|
||||
assert issubclass(SwipeTool, ArtemisTool)
|
||||
assert issubclass(Swipe, ArtemisTool)
|
||||
assert isinstance(swipe, ArtemisTool)
|
||||
assert isinstance(swipe, SwipeTool)
|
||||
|
||||
assert swipe.name == "swipe"
|
||||
assert swipe.category == "action"
|
||||
assert swipe.args_schema == SwipeArgs
|
||||
|
||||
# Registry lookup
|
||||
reg_tool = ToolRegistry.get("swipe")
|
||||
assert reg_tool is not None
|
||||
assert isinstance(reg_tool, ArtemisTool)
|
||||
|
||||
# GenAI FunctionDeclaration export
|
||||
declaration = swipe.to_genai_declaration()
|
||||
assert declaration.name == "swipe"
|
||||
assert "action" in declaration.parameters.properties
|
||||
assert "duration" in declaration.parameters.properties
|
||||
|
||||
# Wrapper check
|
||||
assert swipe_wrapper is not None
|
||||
assert swipe_wrapper.tool_fn_getter == get_swipe_tool
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_swipe_direct_execution_with_driver_direction(mock_driver):
|
||||
"""Verify direct execution with direction on BaseDeviceDriver."""
|
||||
result = await swipe.execute(driver=mock_driver, action="up", duration=500)
|
||||
mock_driver.swipe_direction.assert_called_once_with("up", duration_ms=500)
|
||||
assert "Swipe completed successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_swipe_direct_execution_with_driver_coords(mock_driver):
|
||||
"""Verify direct execution with coordinates on BaseDeviceDriver."""
|
||||
result = await swipe.execute(driver=mock_driver, action=[100, 200, 300, 400], duration=600)
|
||||
mock_driver.swipe.assert_called_once_with(108, 480, 324, 960, duration_ms=600)
|
||||
assert "Swipe completed successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_swipe_direct_execution_with_ctx_direction(mock_ctx):
|
||||
"""Verify direct execution with direction on ArtemisContext."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.swipe_coords = AsyncMock(return_value=None)
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
result = await swipe.execute(ctx=mock_ctx, action="down", duration=400)
|
||||
mock_controller_inst.swipe_coords.assert_called_once_with(648, 720, 648, 1680, 400)
|
||||
assert "Swipe completed successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_swipe_direct_execution_with_ctx_coords(mock_ctx):
|
||||
"""Verify direct execution with coordinates on ArtemisContext."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.swipe_coords = AsyncMock(return_value=None)
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
result = await swipe.execute(ctx=mock_ctx, action=[200, 300, 800, 900], duration=800)
|
||||
mock_controller_inst.swipe_coords.assert_called_once_with(216, 720, 864, 2160, 800)
|
||||
assert "Swipe completed successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_swipe_with_state_command(mock_ctx, mock_state):
|
||||
"""Verify SwipeTool returns ToolMessage when state is provided."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.swipe_coords = AsyncMock(return_value=None)
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
cmd = await swipe.execute(
|
||||
ctx=mock_ctx,
|
||||
action="left",
|
||||
tool_call_id="call_sw_1",
|
||||
state=mock_state,
|
||||
)
|
||||
assert isinstance(cmd, ToolMessage)
|
||||
assert cmd.tool_call_id == "call_sw_1"
|
||||
assert cmd.status == "success"
|
||||
assert "Swipe completed successfully." in cmd.content
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_swipe_callable_execution(mock_driver):
|
||||
"""Verify invoking swipe directly as a callable."""
|
||||
result = await swipe(driver=mock_driver, action="right")
|
||||
assert "Swipe completed successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_swipe_invalid_action(mock_ctx):
|
||||
"""Verify error handling on invalid action."""
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController"):
|
||||
result = await swipe.execute(ctx=mock_ctx, action="diagonal")
|
||||
assert "Error during swipe: Invalid" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_swipe_tool_langchain_ainvoke(mock_ctx):
|
||||
"""Verify get_swipe_tool exports a LangChain tool that works with ainvoke."""
|
||||
sw_tool = get_swipe_tool(mock_ctx)
|
||||
assert sw_tool.name == "swipe"
|
||||
|
||||
with patch("artemis.tools.mobile.exec_tools.UnifiedMobileController") as mock_controller_cls:
|
||||
mock_controller_inst = MagicMock()
|
||||
mock_controller_inst.swipe_coords = AsyncMock(return_value=None)
|
||||
mock_controller_cls.return_value = mock_controller_inst
|
||||
|
||||
result = await sw_tool.ainvoke({"action": "up"})
|
||||
assert "Swipe completed successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_swipe_with_direction_and_start_end(mock_driver):
|
||||
"""Verify swipe works with explicit direction and start/end parameters."""
|
||||
# Direction parameter
|
||||
result_dir = await swipe.execute(driver=mock_driver, direction="left", duration=500)
|
||||
mock_driver.swipe_direction.assert_called_with("left", duration_ms=500)
|
||||
assert "Swipe completed successfully." in result_dir
|
||||
|
||||
# Start and End parameters
|
||||
result_drag = await swipe.execute(
|
||||
driver=mock_driver, start=[100, 200], end=[300, 400], duration=1500
|
||||
)
|
||||
mock_driver.swipe.assert_called_with(108, 480, 324, 960, duration_ms=1500)
|
||||
assert "Swipe completed successfully." in result_drag
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_swipe_with_coordinates_alias(mock_driver):
|
||||
"""Verify swipe works with coordinates alias parameter."""
|
||||
result = await swipe.execute(driver=mock_driver, coordinates=[200, 300, 400, 500], duration=800)
|
||||
mock_driver.swipe.assert_called_with(216, 720, 432, 1200, duration_ms=800)
|
||||
assert "Swipe completed successfully." in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_swipe_with_string_coordinates(mock_driver):
|
||||
"""Verify swipe works with string-formatted coordinates."""
|
||||
# List format in string
|
||||
result1 = await swipe.execute(driver=mock_driver, action="[920, 290, 920, 180]", duration=1000)
|
||||
mock_driver.swipe.assert_called_with(993, 696, 993, 432, duration_ms=1000)
|
||||
assert "Swipe completed successfully." in result1
|
||||
|
||||
# Space-separated format in string
|
||||
result2 = await swipe.execute(driver=mock_driver, action="920 290 920 180", duration=1000)
|
||||
assert "Swipe completed successfully." in result2
|
||||
|
||||
|
||||
def test_parse_swipe_parameters_direct():
|
||||
"""Verify parse_swipe_parameters with various parameter formats."""
|
||||
from artemis.utils.coordinates import parse_swipe_parameters
|
||||
|
||||
# List
|
||||
k, t, d = parse_swipe_parameters([100, 200, 300, 400])
|
||||
assert k == "coords"
|
||||
assert t == [100, 200, 300, 400]
|
||||
|
||||
# String list
|
||||
k, t, d = parse_swipe_parameters("[920, 290, 920, 180]")
|
||||
assert k == "coords"
|
||||
assert t == [920, 290, 920, 180]
|
||||
|
||||
# String space separated
|
||||
k, t, d = parse_swipe_parameters("920 290 920 180")
|
||||
assert k == "coords"
|
||||
assert t == [920, 290, 920, 180]
|
||||
|
||||
# Dict with action string
|
||||
k, t, d = parse_swipe_parameters({"action": "[920, 290, 920, 180]", "duration": 1000})
|
||||
assert k == "coords"
|
||||
assert t == [920, 290, 920, 180]
|
||||
assert d == 1000
|
||||
|
||||
# Dict with start and end
|
||||
k, t, d = parse_swipe_parameters({"start": [100, 200], "end": [300, 400]})
|
||||
assert k == "coords"
|
||||
assert t == [100, 200, 300, 400]
|
||||
|
||||
# Direction string
|
||||
k, t, d = parse_swipe_parameters({"action": "up"})
|
||||
assert k == "direction"
|
||||
assert t == "up"
|
||||
|
||||
|
||||
def test_compute_smart_swipe_coordinates_fallback():
|
||||
"""Verify fallback smart swipe coordinates on full screen."""
|
||||
from artemis.utils.coordinates import compute_smart_swipe_coordinates
|
||||
|
||||
sx, sy, ex, ey, dur = compute_smart_swipe_coordinates("up", width=1080, height=2400)
|
||||
assert sx == 648
|
||||
assert ex == 648
|
||||
assert sy == 1680
|
||||
assert ey == 720
|
||||
assert 700 <= dur <= 900
|
||||
|
||||
|
||||
def test_compute_smart_swipe_coordinates_with_target_element():
|
||||
"""Verify container-aware coordinates when target element index is specified."""
|
||||
from artemis.utils.coordinates import compute_smart_swipe_coordinates
|
||||
|
||||
indexed_elements = [
|
||||
{"bounds": "[0, 200][1080, 1200]"}, # Element 1
|
||||
{"bounds": "[100, 500][900, 1500]"}, # Element 2 (width=800, height=1000)
|
||||
]
|
||||
# Swipe up within element 2
|
||||
sx, sy, ex, ey, dur = compute_smart_swipe_coordinates(
|
||||
direction="up",
|
||||
target=2,
|
||||
indexed_elements=indexed_elements,
|
||||
width=1080,
|
||||
height=2400,
|
||||
)
|
||||
# c_left=100, c_top=500, c_width=800, c_height=1000
|
||||
# sx = ex = 100 + 800 * 0.6 = 580
|
||||
# sy = 500 + 1000 * 0.7 = 1200
|
||||
# ey = 500 + 1000 * 0.3 = 800
|
||||
assert sx == 580
|
||||
assert ex == 580
|
||||
assert sy == 1200
|
||||
assert ey == 800
|
||||
|
||||
|
||||
def test_compute_smart_swipe_coordinates_with_scrollable_hierarchy():
|
||||
"""Verify auto-detection of scrollable container in UI hierarchy."""
|
||||
from artemis.utils.coordinates import compute_smart_swipe_coordinates
|
||||
|
||||
ui_hierarchy = [
|
||||
{
|
||||
"class": "android.widget.FrameLayout",
|
||||
"bounds": "[0, 0][1080, 2400]",
|
||||
"children": [
|
||||
{
|
||||
"class": "androidx.recyclerview.widget.RecyclerView",
|
||||
"scrollable": True,
|
||||
"bounds": "[50, 300][1030, 2100]", # width=980, height=1800
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
sx, sy, ex, ey, dur = compute_smart_swipe_coordinates(
|
||||
direction="down",
|
||||
ui_hierarchy=ui_hierarchy,
|
||||
width=1080,
|
||||
height=2400,
|
||||
)
|
||||
# c_left=50, c_top=300, c_width=980, c_height=1800
|
||||
# sx = ex = 50 + 980 * 0.6 = 638
|
||||
# sy = 300 + 1800 * 0.3 = 840
|
||||
# ey = 300 + 1800 * 0.7 = 1560
|
||||
assert sx == 638
|
||||
assert ex == 638
|
||||
assert sy == 840
|
||||
assert ey == 1560
|
||||
Reference in new issue
Block a user