test: add comprehensive unit, integration, and e2e test suites

Introduce complete test coverage for the ARTEMIS platform across multiple layers:
Unit tests: Add tests for agent nodes (Planner, Operator, Validator, Diagnoser, FlashRunner, Explorer), tools, utils (OCR/XML fusion, UI hierarchy), and data engine.
Integration tests: Add live tool verification, reactive runner, IME, and standalone workflow tests.
E2E tests: Add end-to-end scenario tests for media playback and failure recovery flows.
Fixtures & mocks: Add mock inputs (audio, video, screenshot) and test utilities.
This commit is contained in:
chiwanko11 committed 2026-08-13 12:18:32 -07:00
1 parent c1803fcb86
commit 5d2283deaa
115 files changed
+22813

No files matched your search

+98
View File
@@ -0,0 +1,98 @@
# 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.agents.checker.checker import CheckerResult, run_async_check
from artemis.context import ArtemisContext
import pytest
@pytest.mark.asyncio
async def test_run_async_check(tmp_path):
notes_dir = tmp_path / "notes"
notes_dir.mkdir(parents=True, exist_ok=True)
# Put a dummy task plan file
(notes_dir / "task_plan.md").write_text("Plan")
# Put a dummy verification_chat file
import hashlib
subgoal_text = "Open WhatsApp"
s_hash = hashlib.md5(subgoal_text.encode("utf-8")).hexdigest()
(notes_dir / f"verification_chat_{s_hash}.json").write_text("[]")
mock_ctx = MagicMock(spec=ArtemisContext)
mock_ctx.data_engine = MagicMock()
mock_ctx.data_engine.base_dir = str(tmp_path)
mock_ctx.data_engine.get_agent_friendly_steps.return_value = []
mock_ctx.execution_setup = None
# Mock controller for screen data
mock_controller = MagicMock()
mock_controller.take_screenshot = AsyncMock(return_value="dummy_screenshot_base64")
from artemis.controllers.device_controller import ScreenDataResponse
mock_screen_data = ScreenDataResponse(
base64="dummy_screenshot_base64",
elements=[],
width=1080,
height=2400,
platform="android",
)
mock_controller.get_screen_data = AsyncMock(return_value=mock_screen_data)
# Mock LLM response with with_structured_output
mock_llm = MagicMock()
mock_response = MagicMock()
mock_response.content = "Task completed successfully."
mock_response.tool_calls = []
async def mock_astream(*args, **kwargs):
yield mock_response
mock_llm.astream.side_effect = mock_astream
mock_llm.bind_tools.return_value = mock_llm
mock_structured_llm = MagicMock()
mock_structured_llm.ainvoke = AsyncMock(
return_value=CheckerResult(success=True, reason="Task completed successfully.")
)
mock_llm.with_structured_output.return_value = mock_structured_llm
with (
patch(
"artemis.agents.checker.checker.UnifiedMobileController",
return_value=mock_controller,
),
patch("artemis.agents.checker.checker.get_llm", return_value=mock_llm),
patch(
"artemis.utils.task_tree.build_plan_and_history",
return_value="Task tree",
),
):
result = await run_async_check(
mock_ctx,
subgoal_text=subgoal_text,
subgoal_hash=s_hash,
raw_perception_data={"screenshot_b64": "dummy", "width": 1080, "height": 2400},
latest_ui_hierarchy=[],
)
assert result["status"] == "success"
assert result["reason"] == "Task completed successfully."
assert mock_llm.astream.called
assert mock_llm.with_structured_output.called
+308
View File
@@ -0,0 +1,308 @@
# 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, Mock, patch
from langchain_core.messages import AIMessage, ToolMessage
from langgraph.types import Command
from artemis.agents.diagnoser.diagnoser import Diagnoser
from artemis.constants import VALIDATOR_MESSAGES_KEY
from artemis.context import ArtemisContext
from artemis.graph.state import State
import pytest
@pytest.fixture
def mock_context():
ctx = Mock(spec=ArtemisContext)
ctx.llm_config = Mock()
ctx.data_engine = Mock()
ctx.data_engine.base_dir = "/tmp"
ctx.data_engine.get_agent_friendly_steps.return_value = []
# Mock execution_setup
mock_setup = Mock()
mock_setup.video_recording_tools_enabled = True
ctx.execution_setup = mock_setup
return ctx
@pytest.fixture
def mock_state():
state = Mock(spec=State)
state.initial_goal = "Test goal"
state.latest_screenshot = None
state.latest_ui_hierarchy = []
return state
@patch("artemis.agents.diagnoser.diagnoser.get_llm")
@patch("artemis.agents.diagnoser.diagnoser.get_video_analyzer_tool")
@patch("artemis.agents.diagnoser.diagnoser.get_analyze_logs_tool")
@pytest.mark.asyncio
async def test_diagnostic_agent_simple_call(
mock_get_log_tool,
mock_get_video_tool,
mock_get_llm,
mock_context,
mock_state,
):
"""Test that DiagnosticAgent calls LLM and returns result."""
mock_llm = Mock()
mock_llm.bind_tools.return_value = mock_llm
mock_response = AIMessage(content="I diagnosed the issue: it is a network error.")
async def mock_astream(*args, **kwargs):
yield mock_response
mock_llm.astream.side_effect = mock_astream
mock_get_llm.return_value = mock_llm
# Mock tools
mock_video_tool = Mock()
mock_video_tool.name = "video_analyzer"
mock_get_video_tool.return_value = mock_video_tool
mock_log_tool = Mock()
mock_log_tool.name = "analyze_logs"
mock_get_log_tool.return_value = mock_log_tool
agent = Diagnoser(mock_context)
result = await agent.run("Diagnose why the app crashed", mock_state)
assert result == "I diagnosed the issue: it is a network error."
@patch("artemis.agents.diagnoser.diagnoser.get_llm")
@patch("artemis.agents.diagnoser.diagnoser.get_video_analyzer_tool")
@patch("artemis.agents.diagnoser.diagnoser.get_analyze_logs_tool")
@pytest.mark.asyncio
async def test_diagnostic_agent_tool_calls(
mock_get_log_tool,
mock_get_video_tool,
mock_get_llm,
mock_context,
mock_state,
):
"""Test that DiagnosticAgent handles tool calls."""
mock_llm = Mock()
mock_llm.bind_tools.return_value = mock_llm
# Simulate response with tool calls
mock_response = AIMessage(
content="I need to check logs and video.",
tool_calls=[
{
"name": "analyze_logs",
"args": {"specific_query": "Check for crashes"},
"id": "call_1",
},
{
"name": "video_analyzer",
"args": {
"time_description": "around 10s",
"purpose": "Check UI",
},
"id": "call_2",
},
],
)
# Second call returns final answer
mock_final_response = AIMessage(content="Diagnosis complete.")
responses = [mock_response, mock_final_response]
call_count = 0
async def mock_astream(*args, **kwargs):
nonlocal call_count
if call_count < len(responses):
yield responses[call_count]
call_count += 1
mock_llm.astream.side_effect = mock_astream
mock_get_llm.return_value = mock_llm
# Mock tools
mock_log_tool = AsyncMock()
mock_log_tool.name = "analyze_logs"
mock_log_tool.args = {"specific_query": None, "state": None}
mock_log_tool.ainvoke.return_value = "Logs show no crash."
mock_get_log_tool.return_value = mock_log_tool
mock_video_tool = AsyncMock()
mock_video_tool.name = "video_analyzer"
mock_video_tool.args = {
"time_description": None,
"purpose": None,
"state": None,
}
mock_command = Command(
update={
VALIDATOR_MESSAGES_KEY: [
ToolMessage(content="Video shows UI loading.", tool_call_id="call_2")
]
}
)
mock_video_tool.ainvoke.return_value = mock_command
mock_get_video_tool.return_value = mock_video_tool
agent = Diagnoser(mock_context)
result = await agent.run("Diagnose issue", mock_state)
assert result == "Diagnosis complete."
mock_log_tool.ainvoke.assert_called_once_with(
{"specific_query": "Check for crashes", "state": mock_state}
)
mock_video_tool.ainvoke.assert_called_once()
@patch("artemis.agents.diagnoser.diagnoser.get_llm")
@patch("artemis.agents.diagnoser.diagnoser.get_video_analyzer_tool")
@patch("artemis.agents.diagnoser.diagnoser.get_analyze_logs_tool")
@pytest.mark.asyncio
async def test_diagnostic_agent_no_video_tool_when_disabled(
mock_get_log_tool,
mock_get_video_tool,
mock_get_llm,
mock_context,
mock_state,
):
"""Test that Diagnoser does not bind video tool when disabled."""
# Set video recording to False
mock_context.execution_setup.video_recording_tools_enabled = False
mock_llm = Mock()
mock_llm.bind_tools.return_value = mock_llm
mock_response = AIMessage(content="Diagnosis complete.")
async def mock_astream(*args, **kwargs):
yield mock_response
mock_llm.astream.side_effect = mock_astream
mock_get_llm.return_value = mock_llm
# Mock tools
mock_video_tool = Mock()
mock_video_tool.name = "video_analyzer"
mock_get_video_tool.return_value = mock_video_tool
mock_log_tool = Mock()
mock_log_tool.name = "analyze_logs"
mock_get_log_tool.return_value = mock_log_tool
agent = Diagnoser(mock_context)
await agent.run("Diagnose issue", mock_state)
# Verify bind_tools was called
mock_llm.bind_tools.assert_called_once()
bound_tools = mock_llm.bind_tools.call_args[1]["tools"]
# Verify video_analyzer is NOT in bound tools
tool_names = [t.name for t in bound_tools if hasattr(t, "name")]
assert "video_analyzer" not in tool_names
# Verify log tool IS in bound tools (should still be there)
assert "analyze_logs" in tool_names
@patch("artemis.agents.diagnoser.diagnoser.get_llm")
@patch("artemis.tools.video_tool.VideoAnalyzer")
@pytest.mark.asyncio
async def test_diagnoser_background_job_lifecycle(
mock_video_analyzer_class, mock_get_llm, mock_state
):
# Setup context
ctx = ArtemisContext.model_construct(
device=Mock(),
llm_config=Mock(),
execution_setup=Mock(video_recording_tools_enabled=True),
)
ctx.data_engine = None
# The Mock VideoAnalyzer agent running inside the background job
mock_agent_instance = AsyncMock()
async def mock_agent_run(time_desc, purpose):
return "Video shows step successfully completed.", "success"
mock_agent_instance.run.side_effect = mock_agent_run
mock_video_analyzer_class.return_value = mock_agent_instance
# Prepare LLM responses
mock_llm = Mock()
mock_llm.bind_tools.return_value = mock_llm
# Turn 1: LLM requests tool call
mock_response_1 = AIMessage(
content="Checking the video first.",
tool_calls=[
{
"name": "video_analyzer",
"args": {
"time_description": "from 0s to 5s",
"purpose": "check UI",
},
"id": "call_video_1",
}
],
)
# Turn 2: LLM gets the initial job start message, continues, then background job finishes, and result is injected.
mock_response_2 = AIMessage(content="Final complete diagnosis.")
responses = [mock_response_1, mock_response_2]
call_count = 0
async def mock_astream(*args, **kwargs):
nonlocal call_count
if call_count < len(responses):
yield responses[call_count]
call_count += 1
mock_llm.astream.side_effect = mock_astream
mock_get_llm.return_value = mock_llm
# Mock asanitize_update on mock_state
async def mock_asanitize_update(ctx, update, agent):
return update
mock_state.asanitize_update = mock_asanitize_update
# Instantiate the Diagnoser
agent = Diagnoser(ctx)
# We want to use the actual video analyzer tool so it triggers the background task
from artemis.tools.video_tool import get_video_analyzer_tool
real_video_tool = get_video_analyzer_tool(ctx, role="diagnoser")
with patch(
"artemis.tools.log_tool.get_analyze_logs_tool",
return_value=Mock(name="analyze_logs"),
):
with patch(
"artemis.tools.video_tool.get_video_analyzer_tool",
return_value=real_video_tool,
):
result = await agent.run("Diagnose video issue", mock_state)
assert result == "Final complete diagnosis."
# Verify job registry
assert len(ctx.background_jobs) == 1
job_id = list(ctx.background_jobs.keys())[0]
job = ctx.background_jobs[job_id]
assert job["tool_call_id"] == "call_video_1"
assert job["status"] == "completed"
assert job["result"] == "Video shows step successfully completed."
assert job["consumed"] is True
@@ -0,0 +1,343 @@
# 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, Mock, patch
from adbutils import AdbClient
from langchain_core.messages import AIMessage, ToolMessage
from artemis.agents.diagnoser.diagnoser import Diagnoser
from artemis.context import ArtemisContext
from artemis.graph.state import State
import pytest
@pytest.fixture(scope="session", autouse=True)
def ensure_device_connected():
try:
client = AdbClient()
if not client.device_list():
pytest.skip("No active Android device/emulator found. Skipping diagnostic tool tests.")
except Exception as e:
pytest.skip(f"ADB Client connection failed: {e}. Skipping tests.")
@pytest.fixture
def mock_context():
ctx = Mock(spec=ArtemisContext)
ctx.llm_config = Mock()
ctx.data_engine = Mock()
ctx.data_engine.base_dir = "/tmp"
ctx.data_engine.get_agent_friendly_steps.return_value = []
# Mock execution_setup
mock_setup = Mock()
mock_setup.video_recording_tools_enabled = True
ctx.execution_setup = mock_setup
return ctx
@pytest.fixture
def mock_state():
state = Mock(spec=State)
state.initial_goal = "Test goal"
state.latest_screenshot = None
state.latest_ui_hierarchy = []
return state
@patch("artemis.agents.diagnoser.diagnoser.get_llm")
@patch("artemis.tools.video_tool.get_video_analyzer_tool")
@patch("artemis.tools.log_tool.get_analyze_logs_tool")
@patch("artemis.agents.diagnoser.diagnoser.get_ui_hierarchy_tool")
@pytest.mark.asyncio
async def test_maestro_uninstall_timeout(
mock_get_ui_tool,
mock_get_log_tool,
mock_get_video_tool,
mock_get_llm,
mock_context,
mock_state,
):
"""Test Maestro Uninstall Timeout / Blocker Failure."""
# Mock LLM calls
mock_llm = Mock()
mock_llm.bind_tools.return_value = mock_llm
mock_response_1 = AIMessage(
content="I need to check the UI hierarchy.",
tool_calls=[
{
"name": "get_ui_hierarchy",
"args": {},
"id": "call_hierarchy",
}
],
)
mock_response_2 = AIMessage(
content=(
"Diagnoser identifies the blocker failure and recommends restarting"
" ADB/uninstalling conflicting tools manually."
)
)
responses = [mock_response_1, mock_response_2]
call_count = 0
async def mock_astream(*args, **kwargs):
nonlocal call_count
if call_count < len(responses):
yield responses[call_count]
call_count += 1
mock_llm.astream.side_effect = mock_astream
mock_get_llm.return_value = mock_llm
# Mock tool
mock_ui_tool = AsyncMock()
mock_ui_tool.name = "get_ui_hierarchy"
mock_ui_tool.args = {}
mock_ui_tool.ainvoke.return_value = "Error: pm list packages timed out"
mock_get_ui_tool.return_value = mock_ui_tool
# Run agent
agent = Diagnoser(mock_context)
result = await agent.run("Diagnose why the app crashed", mock_state)
assert "recommends restarting ADB" in result
assert "uninstalling conflicting tools" in result
# Assert get_ui_hierarchy was called
mock_ui_tool.ainvoke.assert_called_once()
# Assert that the ToolMessage was sent back to LLM
assert len(mock_llm.astream.call_args_list) == 2
sent_messages = mock_llm.astream.call_args_list[1][0][0]
tool_message = sent_messages[-2]
assert isinstance(tool_message, ToolMessage)
assert tool_message.tool_call_id == "call_hierarchy"
assert tool_message.content == "Error: pm list packages timed out"
assert tool_message.status == "error"
@patch("artemis.agents.diagnoser.diagnoser.get_llm")
@patch("artemis.tools.video_tool.get_video_analyzer_tool")
@patch("artemis.tools.log_tool.get_analyze_logs_tool")
@patch("artemis.agents.diagnoser.diagnoser.get_ui_hierarchy_tool")
@pytest.mark.asyncio
async def test_uiautomator2_connection_lost(
mock_get_ui_tool,
mock_get_log_tool,
mock_get_video_tool,
mock_get_llm,
mock_context,
mock_state,
):
"""Test UIAutomator2 Daemon Connection Lost."""
# Mock LLM
mock_llm = Mock()
mock_llm.bind_tools.return_value = mock_llm
mock_response_1 = AIMessage(
content="Let me check the UI hierarchy.",
tool_calls=[
{
"name": "get_ui_hierarchy",
"args": {},
"id": "call_hierarchy_lost",
}
],
)
mock_response_2 = AIMessage(content="Diagnoser reports connection loss with UIAutomator2.")
responses = [mock_response_1, mock_response_2]
call_count = 0
async def mock_astream(*args, **kwargs):
nonlocal call_count
if call_count < len(responses):
yield responses[call_count]
call_count += 1
mock_llm.astream.side_effect = mock_astream
mock_get_llm.return_value = mock_llm
# Mock tool to throw ConnectionError
mock_ui_tool = AsyncMock()
mock_ui_tool.name = "get_ui_hierarchy"
mock_ui_tool.args = {}
mock_ui_tool.ainvoke.side_effect = ConnectionError(
"UIAutomator2 connection lost, reconnecting..."
)
mock_get_ui_tool.return_value = mock_ui_tool
# Run agent
agent = Diagnoser(mock_context)
result = await agent.run("Diagnose crash", mock_state)
assert "reports connection loss" in result
# Assert that tool error is wrapped and returned in the ToolMessage
assert len(mock_llm.astream.call_args_list) == 2
sent_messages = mock_llm.astream.call_args_list[1][0][0]
tool_message = sent_messages[-2]
assert isinstance(tool_message, ToolMessage)
assert tool_message.tool_call_id == "call_hierarchy_lost"
assert "UIAutomator2 connection lost, reconnecting..." in tool_message.content
assert tool_message.status == "error"
# Verify that the trace span recorded the failure
record_trace_calls = mock_context.data_engine.record_trace.call_args_list
tool_trace_calls = [c for c in record_trace_calls if c[1].get("name") == "get_ui_hierarchy"]
failed_trace = next((c for c in tool_trace_calls if c[1].get("status") == "failed"), None)
assert failed_trace is not None
assert "UIAutomator2 connection lost, reconnecting..." in failed_trace[1]["payload"]["error"]
@patch("artemis.agents.diagnoser.diagnoser.get_llm")
@patch("artemis.tools.video_tool.get_video_analyzer_tool")
@patch("artemis.tools.log_tool.get_analyze_logs_tool")
@patch("artemis.agents.diagnoser.diagnoser.get_ui_hierarchy_tool")
@pytest.mark.asyncio
async def test_ui_hierarchy_dump_stall(
mock_get_ui_tool,
mock_get_log_tool,
mock_get_video_tool,
mock_get_llm,
mock_context,
mock_state,
):
"""Test UI Hierarchy Dump Stall (UI Thread Frozen)."""
# Mock LLM
mock_llm = Mock()
mock_llm.bind_tools.return_value = mock_llm
mock_response_1 = AIMessage(
content="Let me fetch the UI hierarchy to see what's on screen.",
tool_calls=[
{
"name": "get_ui_hierarchy",
"args": {},
"id": "call_hierarchy_stall",
}
],
)
mock_response_2 = AIMessage(
content=(
"Diagnoser identifies frozen UI and suggests rendering-safe"
" recovery actions (like ADB wake commands or a wait/retry cycle)."
)
)
responses = [mock_response_1, mock_response_2]
call_count = 0
async def mock_astream(*args, **kwargs):
nonlocal call_count
if call_count < len(responses):
yield responses[call_count]
call_count += 1
mock_llm.astream.side_effect = mock_astream
mock_get_llm.return_value = mock_llm
# Mock tool
mock_ui_tool = AsyncMock()
mock_ui_tool.name = "get_ui_hierarchy"
mock_ui_tool.args = {}
mock_ui_tool.ainvoke.return_value = (
"Error retrieving UI hierarchy: dump_hierarchy timed out after 3 seconds"
)
mock_get_ui_tool.return_value = mock_ui_tool
# Run agent
agent = Diagnoser(mock_context)
result = await agent.run("Diagnose stall", mock_state)
assert "rendering-safe recovery actions" in result
# Assert ToolMessage was passed back with correct error details
assert len(mock_llm.astream.call_args_list) == 2
sent_messages = mock_llm.astream.call_args_list[1][0][0]
tool_message = sent_messages[-2]
assert isinstance(tool_message, ToolMessage)
assert tool_message.tool_call_id == "call_hierarchy_stall"
assert "dump_hierarchy timed out after 3 seconds" in tool_message.content
assert tool_message.status == "error"
@patch("artemis.agents.diagnoser.diagnoser.get_llm")
@patch("artemis.tools.video_tool.get_video_analyzer_tool")
@patch("artemis.tools.log_tool.get_analyze_logs_tool")
@patch("artemis.agents.diagnoser.diagnoser.get_ui_hierarchy_tool")
@pytest.mark.asyncio
async def test_device_offline(
mock_get_ui_tool,
mock_get_log_tool,
mock_get_video_tool,
mock_get_llm,
mock_context,
mock_state,
):
"""Test Device Offline (ADB Transport Lost)."""
# Mock LLM
mock_llm = Mock()
mock_llm.bind_tools.return_value = mock_llm
mock_response_1 = AIMessage(
content="Let's grab the UI hierarchy.",
tool_calls=[
{
"name": "get_ui_hierarchy",
"args": {},
"id": "call_hierarchy_offline",
}
],
)
mock_response_2 = AIMessage(
content=("The device is offline. Diagnoser flags the offline device state.")
)
responses = [mock_response_1, mock_response_2]
call_count = 0
async def mock_astream(*args, **kwargs):
nonlocal call_count
if call_count < len(responses):
yield responses[call_count]
call_count += 1
mock_llm.astream.side_effect = mock_astream
mock_get_llm.return_value = mock_llm
# Mock tool
mock_ui_tool = AsyncMock()
mock_ui_tool.name = "get_ui_hierarchy"
mock_ui_tool.args = {}
mock_ui_tool.ainvoke.side_effect = RuntimeError("device offline")
mock_get_ui_tool.return_value = mock_ui_tool
# Run agent
agent = Diagnoser(mock_context)
result = await agent.run("Diagnose offline", mock_state)
assert "flags the offline device state" in result
# Assert ToolMessage was passed back with correct error details
assert len(mock_llm.astream.call_args_list) == 2
sent_messages = mock_llm.astream.call_args_list[1][0][0]
tool_message = sent_messages[-2]
assert isinstance(tool_message, ToolMessage)
assert tool_message.tool_call_id == "call_hierarchy_offline"
assert "device offline" in tool_message.content
assert tool_message.status == "error"
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,127 @@
# 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 Explorer and Multi-Model Grounding Engine."""
import json
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
from langchain_core.messages import AIMessage
from artemis.agents.explorer.explorer import (
UNIVERSAL_EXPLORER_TOOLS,
Explorer,
)
from artemis.config import settings
from artemis.context import ArtemisContext
from artemis.graph.state import State
@pytest.fixture
def mock_context(tmp_path):
ctx = Mock(spec=ArtemisContext)
ctx.llm_config = Mock()
mock_llm_cfg = Mock()
mock_llm_cfg.model = "claude-3-7-sonnet"
mock_llm_cfg.temperature = 0.1
ctx.llm_config.explorer = mock_llm_cfg
ctx.device = Mock()
ctx.device.device_width = 1080
ctx.device.device_height = 2400
ctx.agent_config = Mock()
ctx.agent_config.blacklisted_tools = {}
ctx.data_engine = Mock()
ctx.data_engine.base_dir = tmp_path
return ctx
def test_explorer_universal_engine_detection(mock_context):
"""Test that Explorer routes to Universal Engine when non-Gemini model is configured."""
with patch.object(settings, "GOOGLE_API_KEY", None):
agent = Explorer(mock_context)
assert agent.use_native_gemini is False
assert agent.model_name == "claude-3-7-sonnet"
def test_universal_explorer_tool_schemas():
"""Verify schema integrity of Universal Explorer tool declarations."""
assert len(UNIVERSAL_EXPLORER_TOOLS) == 6
names = [t["function"]["name"] for t in UNIVERSAL_EXPLORER_TOOLS]
assert "ask_perception_tool" in names
assert "detect_objects" in names
assert "get_ocr_list" in names
assert "ask_image_processor" in names
assert "inspect_region" in names
assert "submit_answer" in names
@pytest.mark.asyncio
async def test_run_universal_explorer_success(mock_context, tmp_path):
"""Test universal Explorer tool invocation and submit_answer coordinate parsing."""
agent = Explorer(mock_context)
agent.use_native_gemini = False
fake_screenshot = tmp_path / "screenshot.jpg"
fake_screenshot.write_bytes(b"fake_image_bytes")
# Mock tool call response for submit_answer
msg_response = AIMessage(
content="Submitting grounded elements.",
tool_calls=[
{
"name": "submit_answer",
"args": {
"candidates": [
{
"label": "D1",
"coords": [500, 600],
"description": "Settings button",
}
],
"fallback_message": "",
},
"id": "exp_call_1",
}
],
)
mock_llm = MagicMock()
mock_bound = MagicMock()
mock_bound.ainvoke = AsyncMock(return_value=msg_response)
mock_llm.bind_tools.return_value = mock_bound
with patch(
"artemis.agents.explorer.explorer.get_llm",
return_value=mock_llm,
):
mock_state = Mock(spec=State)
res_str = await agent._run_universal(
query="find settings icon",
context_feedback="",
screenshot_path=str(fake_screenshot),
state=mock_state,
minimal_list="",
version="pro",
prompt_template="You are an explorer.",
max_iterations=3,
)
res = json.loads(res_str)
assert len(res["candidates"]) == 1
assert res["candidates"][0]["label"] == "D1"
assert res["candidates"][0]["coords"] == [500, 600]
+213
View File
@@ -0,0 +1,213 @@
# 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 Failure Analyzer."""
import base64
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
from langchain_core.messages import AIMessage
from artemis.agents.validator.failure_analyzer import (
FailureAnalyzer,
PixelTargetDisappearedStrategy,
TargetDisappearedStrategy,
ValidationErrorCategory,
VALIDATOR_TOOLS_DECLARATION,
)
from artemis.context import ArtemisContext
class DummyState:
def __init__(self, latest_screenshot=None):
self.latest_screenshot = latest_screenshot
self.indexed_points = []
self.indexed_elements = []
@pytest.fixture
def mock_context(tmp_path):
ctx = Mock(spec=ArtemisContext)
ctx.llm_config = Mock()
mock_llm_cfg = Mock()
mock_llm_cfg.model = "gemini-2.5-pro"
mock_llm_cfg.temperature = 0.1
ctx.llm_config.get_agent.return_value = mock_llm_cfg
ctx.device = Mock()
ctx.device.device_width = 1080
ctx.device.device_height = 2400
ctx.data_engine = Mock()
ctx.data_engine.base_dir = tmp_path
ctx.data_engine.get_agent_friendly_steps.return_value = []
ctx.adb_client = None
ctx.driver = Mock()
return ctx
@pytest.mark.asyncio
async def test_failure_analyzer_cannot_fix(mock_context, tmp_path):
"""Test that FailureAnalyzer natively executes and handles 'cannot_fix' report."""
screenshot_file = tmp_path / "shot.jpg"
screenshot_file.write_bytes(b"dummy_bytes")
state = DummyState(latest_screenshot=str(screenshot_file))
failed_action = {"action": "click", "coordinates": [100, 200]}
error_msg = "Element not found"
analyzer = FailureAnalyzer(mock_context)
mock_llm = MagicMock()
mock_response = AIMessage(
content="I see the error.",
tool_calls=[
{
"name": "report_failure_analysis",
"args": {
"status": "cannot_fix",
"analysis": "The element was definitely not there.",
},
"id": "call_1",
}
],
)
mock_bound = MagicMock()
mock_bound.ainvoke = AsyncMock(return_value=mock_response)
mock_llm.bind_tools.return_value = mock_bound
with (
patch("artemis.agents.validator.failure_analyzer.get_llm", return_value=mock_llm),
patch(
"artemis.agents.validator.failure_analyzer.UnifiedMobileController"
) as mock_controller_cls,
):
mock_controller = Mock()
mock_controller_cls.return_value = mock_controller
mock_device_data = Mock()
mock_device_data.base64 = base64.b64encode(b"mocked_screenshot").decode()
mock_device_data.elements = []
mock_device_data.width = 1080
mock_device_data.height = 2400
mock_controller.get_screen_data = AsyncMock(return_value=mock_device_data)
mock_controller.get_ui_elements = AsyncMock(return_value=mock_device_data.elements)
result = await analyzer.analyze(
state,
failed_action,
error_msg,
pre_screenshot=base64.b64encode(b"dummy").decode(),
post_screenshot=base64.b64encode(b"dummy").decode(),
)
assert result["status"] == "cannot_fix"
assert "not there" in result["analysis"]
assert result["new_remaining_actions"] == []
def test_report_failure_analysis_tool_declaration():
"""Test that report_failure_analysis tool declaration is correct and matches."""
report_tool = next(
(t for t in VALIDATOR_TOOLS_DECLARATION if t.name == "report_failure_analysis"),
None,
)
assert report_tool is not None
props = report_tool.parameters.get("properties", {})
assert "analysis" in props
assert "single paragraph" in props["analysis"]["description"]
@pytest.mark.asyncio
async def test_failure_analyzer_strategy_routing(mock_context):
"""Test that FailureAnalyzer selects appropriate strategies based on error category."""
analyzer = FailureAnalyzer(mock_context)
strat_disappeared = analyzer._select_strategy(ValidationErrorCategory.TARGET_DISAPPEARED)
assert isinstance(strat_disappeared, TargetDisappearedStrategy)
assert "target_disappeared" in strat_disappeared.get_prompt_template_name()
strat_pixel = analyzer._select_strategy(ValidationErrorCategory.PIXEL_TARGET_DISAPPEARED)
assert isinstance(strat_pixel, PixelTargetDisappearedStrategy)
assert "pixel_target_disappeared" in strat_pixel.get_prompt_template_name()
@pytest.mark.asyncio
async def test_failure_analyzer_click_sequence(mock_context, tmp_path):
"""Test click_sequence execution in FailureAnalyzer."""
screenshot_file = tmp_path / "shot.jpg"
screenshot_file.write_bytes(b"dummy_bytes")
state = DummyState(latest_screenshot=str(screenshot_file))
analyzer = FailureAnalyzer(mock_context)
# Turn 1: Call click_sequence tool with direct coordinates
call_step_1 = AIMessage(
content="Let me click in sequence.",
tool_calls=[
{
"name": "click_sequence",
"args": {"sequence": [[100, 200], [300, 400]]},
"id": "call_seq_1",
}
],
)
# Turn 2: Report failure fixed
call_step_2 = AIMessage(
content="Repaired successfully.",
tool_calls=[
{
"name": "report_failure_analysis",
"args": {
"status": "fixed",
"analysis": "Sequence clicked successfully.",
},
"id": "call_report_2",
}
],
)
mock_llm = MagicMock()
mock_bound = MagicMock()
mock_bound.ainvoke = AsyncMock(side_effect=[call_step_1, call_step_2])
mock_llm.bind_tools.return_value = mock_bound
with (
patch("artemis.agents.validator.failure_analyzer.get_llm", return_value=mock_llm),
patch(
"artemis.agents.validator.failure_analyzer.UnifiedMobileController"
) as mock_controller_cls,
):
mock_controller = Mock()
mock_controller_cls.return_value = mock_controller
mock_device_data = Mock()
mock_device_data.base64 = base64.b64encode(b"mocked_screenshot").decode()
mock_device_data.elements = []
mock_device_data.width = 1080
mock_device_data.height = 2400
mock_controller.get_screen_data = AsyncMock(return_value=mock_device_data)
mock_controller.get_ui_elements = AsyncMock(return_value=mock_device_data.elements)
mock_controller.tap_at = AsyncMock(return_value=Mock(error=None))
result = await analyzer.analyze(
state,
{"action": "click", "coordinates": [100, 200]},
"Element not found",
pre_screenshot=base64.b64encode(b"dummy").decode(),
post_screenshot=base64.b64encode(b"dummy").decode(),
error_category=ValidationErrorCategory.TARGET_DISAPPEARED,
)
assert result["status"] == "fixed"
assert mock_controller.tap_at.call_count == 2
@@ -0,0 +1,460 @@
# 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.
import base64
from unittest.mock import AsyncMock, Mock, patch
from artemis.agents.validator.failure_analyzer import (
_exec_click,
_exec_click_sequence,
_exec_input_text,
_exec_long_press,
_exec_manage_app,
_exec_press_key,
_exec_swipe,
_exec_wait_for_delay,
)
from artemis.context import ArtemisContext
import pytest
class DummyState:
def __init__(self, latest_screenshot=None):
self.latest_screenshot = latest_screenshot
self.indexed_points = []
self.indexed_elements = []
@pytest.fixture
def mock_context(tmp_path):
ctx = Mock(spec=ArtemisContext)
ctx.device = Mock()
ctx.device.device_width = 1080
ctx.device.device_height = 2400
ctx.data_engine = Mock()
ctx.data_engine.base_dir = tmp_path
ctx.data_engine.get_or_create_image.return_value = "dummy_image_name"
ctx.data_engine.get_image_path.return_value = tmp_path / "dummy_image_name.jpg"
return ctx
@pytest.fixture
def mock_controller():
controller = Mock()
mock_device_data = Mock()
mock_device_data.base64 = base64.b64encode(b"mocked_screenshot").decode()
mock_device_data.elements = []
mock_device_data.width = 1080
mock_device_data.height = 2400
controller.get_screen_data = AsyncMock(return_value=mock_device_data)
controller.get_ui_elements = AsyncMock(return_value=mock_device_data.elements)
controller.tap_at = AsyncMock(return_value=Mock(error=None))
controller.type_text = AsyncMock(return_value=True)
controller.erase_text = AsyncMock(return_value=True)
controller.swipe_coords = AsyncMock(return_value=None)
controller.press_enter = AsyncMock(return_value=True)
controller.go_back = AsyncMock(return_value=True)
controller.go_home = AsyncMock(return_value=True)
controller.press_key = AsyncMock(return_value=True)
controller.terminate_app = AsyncMock(return_value=True)
return controller
# ==========================================
# SLIDE 1: click
# ==========================================
@pytest.mark.asyncio
async def test_click_happy_path(mock_context, mock_controller):
state = DummyState()
outcome, img_bytes, shot_path, xml_list = await _exec_click(
target=[500, 600],
state=state,
controller=mock_controller,
ctx=mock_context,
times=2,
delay_ms=150,
)
assert "Clicked at [500, 600]" in outcome
assert mock_controller.tap_at.call_count == 1
mock_controller.tap_at.assert_any_call(540, 1440, times=2, delay_ms=150)
@pytest.mark.asyncio
async def test_click_input_validation(mock_context, mock_controller):
state = DummyState()
# Coordinates holding non-integer values
outcome, _, _, _ = await _exec_click(
target=["invalid", "invalid"],
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Error during click:" in outcome
@pytest.mark.asyncio
async def test_click_edge_cases(mock_context, mock_controller):
state = DummyState()
# Underflow/overflow coordinates should be clamped to valid boundary
outcome, _, _, _ = await _exec_click(
target=[-10, 1050],
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Clicked at [-10, 1050]" in outcome
mock_controller.tap_at.assert_called_with(0, 2399, times=1, delay_ms=100)
@pytest.mark.asyncio
async def test_click_graceful_failure(mock_context, mock_controller):
state = DummyState()
mock_controller.tap_at.return_value = Mock(error="Click failed connection lost")
outcome, _, _, _ = await _exec_click(
target=[500, 600],
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Error executing click: Click failed connection lost" in outcome
# ==========================================
# SLIDE 2: click_sequence
# ==========================================
@pytest.mark.asyncio
async def test_click_sequence_happy_path(mock_context, mock_controller):
state = DummyState()
state.indexed_points = [[540, 1200]] # Mapped from index 1 (1-indexed)
outcome, img_bytes, shot_path, xml_list = await _exec_click_sequence(
sequence=[1, [300, 400]],
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Sequence clicked successfully" in outcome
assert mock_controller.tap_at.call_count == 2
mock_controller.tap_at.assert_any_call(540, 1200)
mock_controller.tap_at.assert_any_call(324, 960)
@pytest.mark.asyncio
async def test_click_sequence_input_validation(mock_context, mock_controller):
state = DummyState()
state.indexed_points = [[540, 1200], [270, 600]]
# Subsequent elements can also be integers (mixed array of indexes and coordinates)
outcome, _, _, _ = await _exec_click_sequence(
sequence=[1, 2, [300, 400]],
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Sequence clicked successfully" in outcome
assert mock_controller.tap_at.call_count == 3
mock_controller.tap_at.assert_any_call(540, 1200)
mock_controller.tap_at.assert_any_call(270, 600)
mock_controller.tap_at.assert_any_call(324, 960)
# Reset indexed_points since _exec_click_sequence overwrites them
# via _capture_screenshot_and_parse_ui
state.indexed_points = [[540, 1200], [270, 600]]
# Index out of bounds
outcome, _, _, _ = await _exec_click_sequence(
sequence=[5, [300, 400]],
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Error during click sequence: Index 5 is out of range" in outcome
@pytest.mark.asyncio
async def test_click_sequence_edge_cases(mock_context, mock_controller):
state = DummyState()
# Coordinates in sequence outside range should be clamped
outcome, _, _, _ = await _exec_click_sequence(
sequence=[[0, 0], [1200, -50]],
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Sequence clicked successfully" in outcome
mock_controller.tap_at.assert_any_call(0, 0)
mock_controller.tap_at.assert_any_call(1079, 0)
@pytest.mark.asyncio
async def test_click_sequence_graceful_failure(mock_context, mock_controller):
state = DummyState()
mock_controller.tap_at.side_effect = [
Mock(error=None),
Mock(error="Connection error on step 2"),
]
outcome, _, _, _ = await _exec_click_sequence(
sequence=[[100, 200], [300, 400]],
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Error executing click at" in outcome
assert "Connection error on step 2" in outcome
@pytest.mark.asyncio
async def test_click_sequence_stringified_formats(mock_context, mock_controller):
state = DummyState()
state.indexed_points = [[540, 1200]]
# Test stringified coordinate formats commonly generated by LLMs:
# 1. string representation of lists: ["[500, 280]", "[884, 362]"]
outcome, _, _, _ = await _exec_click_sequence(
sequence=["[500, 280]", "[884, 362]"],
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Sequence clicked successfully" in outcome
mock_controller.tap_at.assert_any_call(540, 672)
mock_controller.tap_at.assert_any_call(954, 868)
# 2. comma-separated strings: ["500,280", "884,362"]
outcome, _, _, _ = await _exec_click_sequence(
sequence=["500,280", "884,362"],
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Sequence clicked successfully" in outcome
# 3. whole sequence as a string
outcome, _, _, _ = await _exec_click_sequence(
sequence="[[500, 280], [884, 362]]",
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Sequence clicked successfully" in outcome
# ==========================================
# SLIDE 3: long_press
# ==========================================
@pytest.mark.asyncio
async def test_long_press_happy_path(mock_context, mock_controller):
state = DummyState()
outcome, img_bytes, shot_path, xml_list = await _exec_long_press(
target=[500, 500],
duration=1500,
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Long pressed at [500, 500]" in outcome
mock_controller.tap_at.assert_called_once_with(
540, 1200, long_press=True, long_press_duration=1500
)
@pytest.mark.asyncio
async def test_long_press_input_validation(mock_context, mock_controller):
state = DummyState()
# Invalid target type
outcome, _, _, _ = await _exec_long_press(
target="not-a-list",
duration=1500,
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Error during long press:" in outcome
# ==========================================
# SLIDE 4: input_text
# ==========================================
@pytest.mark.asyncio
async def test_input_text_happy_path(mock_context, mock_controller):
state = DummyState()
outcome, img_bytes, shot_path, xml_list = await _exec_input_text(
text="hello",
target=[500, 500],
clear_exist=True,
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Executed typing 'hello'." in outcome
mock_controller.tap_at.assert_called_once_with(540, 1200)
mock_controller.erase_text.assert_called_once()
mock_controller.type_text.assert_called_once_with("hello")
@pytest.mark.asyncio
async def test_input_text_input_validation(mock_context, mock_controller):
state = DummyState()
outcome, _, _, _ = await _exec_input_text(
text="hello",
target="invalid",
clear_exist=True,
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Error during input text:" in outcome
# ==========================================
# SLIDE 5: swipe
# ==========================================
@pytest.mark.asyncio
async def test_swipe_happy_path_direction(mock_context, mock_controller):
state = DummyState()
outcome, img_bytes, shot_path, xml_list = await _exec_swipe(
action="up",
duration=400,
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Swipe completed successfully" in outcome
mock_controller.swipe_coords.assert_called_once_with(648, 1920, 648, 480, 400)
@pytest.mark.asyncio
async def test_swipe_happy_path_coords(mock_context, mock_controller):
state = DummyState()
outcome, img_bytes, shot_path, xml_list = await _exec_swipe(
action=[100, 900, 100, 100],
duration=400,
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Swipe completed successfully" in outcome
mock_controller.swipe_coords.assert_called_once_with(108, 2160, 108, 240, 400)
@pytest.mark.asyncio
async def test_swipe_input_validation(mock_context, mock_controller):
state = DummyState()
outcome, _, _, _ = await _exec_swipe(
action="invalid_direction",
duration=400,
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Error during swipe: Invalid direction:" in outcome
# ==========================================
# SLIDE 6: press_key
# ==========================================
@pytest.mark.asyncio
async def test_press_key_happy_path(mock_context, mock_controller):
state = DummyState()
outcome, img_bytes, shot_path, xml_list = await _exec_press_key(
key="BACK",
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Executed key press 'BACK'." in outcome
mock_controller.go_back.assert_called_once()
@pytest.mark.asyncio
async def test_press_key_input_validation(mock_context, mock_controller):
state = DummyState()
outcome, _, _, _ = await _exec_press_key(
key="INVALID_KEY",
state=state,
controller=mock_controller,
ctx=mock_context,
)
# The current execution doesn't raise error on unknown keys, just logs error
assert "Error executing key press 'INVALID_KEY'." in outcome
# ==========================================
# SLIDE 7: manage_app
# ==========================================
@pytest.mark.asyncio
async def test_manage_app_happy_path(mock_context, mock_controller):
state = DummyState()
with (
patch("artemis.agents.validator.tool_declarations.find_package") as mock_find_package,
patch("artemis.agents.validator.tool_declarations.launch_app_with_retries") as mock_launch,
):
mock_find_package.return_value = "com.google.android.youtube"
mock_launch.return_value = (True, "")
outcome, img_bytes, shot_path, xml_list = await _exec_manage_app(
action="launch",
app_name="YouTube",
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Launched app 'YouTube' (com.google.android.youtube) successfully." in outcome
mock_find_package.assert_called_once_with(mock_context, "YouTube", use_fallback=False)
mock_launch.assert_called_once_with(mock_context, "com.google.android.youtube")
@pytest.mark.asyncio
async def test_manage_app_input_validation(mock_context, mock_controller):
state = DummyState()
with patch("artemis.agents.validator.tool_declarations.find_package") as mock_find_package:
mock_find_package.return_value = None
outcome, _, _, _ = await _exec_manage_app(
action="launch",
app_name="NonExistentApp",
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Error finding package for app: NonExistentApp" in outcome
# ==========================================
# SLIDE 8: wait_for_delay
# ==========================================
@pytest.mark.asyncio
async def test_wait_for_delay_happy_path(mock_context, mock_controller):
state = DummyState()
outcome, img_bytes, shot_path, xml_list = await _exec_wait_for_delay(
time_in_ms=100,
state=state,
controller=mock_controller,
ctx=mock_context,
)
assert "Waited for 100ms successfully." in outcome
+113
View File
@@ -0,0 +1,113 @@
# 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 FlashRunner."""
from unittest.mock import Mock, patch
import pytest
from langchain_core.messages import HumanMessage, ToolMessage
from artemis.agents.flash.runner import FlashRunner
from artemis.context import ArtemisContext
@pytest.fixture
def mock_context():
ctx = Mock(spec=ArtemisContext)
ctx.llm_config = Mock()
mock_llm_cfg = Mock()
mock_llm_cfg.model = "gemini-2.5-flash"
mock_llm_cfg.temperature = 0.1
ctx.llm_config.get_agent.return_value = mock_llm_cfg
ctx.device = Mock()
ctx.device.device_width = 1080
ctx.device.device_height = 2400
ctx.data_engine = None
ctx.adb_client = None
ctx.driver = Mock()
return ctx
def test_flash_runner_tools_initialization(mock_context):
"""Verify FlashRunner gathers correct active tools."""
with patch("artemis.controllers.unified_controller.get_driver"):
runner = FlashRunner(mock_context, goal="Test Open Settings")
tools = runner._get_tools()
tool_names = [t.name for t in tools]
assert "click" in tool_names
assert "click_sequence" in tool_names
assert "swipe" in tool_names
assert "input_text" in tool_names
assert "ask_explorer" in tool_names
assert "report_task_status" in tool_names
# Validator's report_failure_analysis should NOT be in FlashRunner
assert "report_failure_analysis" not in tool_names
def test_flash_runner_screenshot_pruning(mock_context):
"""Verify that earlier screenshot blocks are pruned, keeping only the latest."""
with patch("artemis.controllers.unified_controller.get_driver"):
runner = FlashRunner(mock_context, goal="Test Pruning")
messages = [
HumanMessage(
content=[
{"type": "text", "text": "Turn 1 goal"},
{
"type": "image_url",
"image_url": {"url": "data:image/jpeg;base64,SCREENSHOT_1"},
},
]
),
ToolMessage(
tool_call_id="tc1",
name="click",
content=[
{"type": "text", "text": "Clicked button"},
{
"type": "image_url",
"image_url": {"url": "data:image/jpeg;base64,SCREENSHOT_2"},
},
],
),
ToolMessage(
tool_call_id="tc2",
name="swipe",
content=[
{"type": "text", "text": "Swiped up"},
{
"type": "image_url",
"image_url": {"url": "data:image/jpeg;base64,SCREENSHOT_3"},
},
],
),
]
runner._prune_intermediate_screenshots(messages)
# Check Turn 1: image should be removed
assert len(messages[0].content) == 1
assert messages[0].content[0]["type"] == "text"
# Check Turn 2 (Tool 1): image should be removed
assert len(messages[1].content) == 1
assert messages[1].content[0]["type"] == "text"
# Check Turn 3 (Tool 2 - Latest): image MUST be preserved
assert len(messages[2].content) == 2
assert messages[2].content[0]["type"] == "text"
assert messages[2].content[1]["type"] == "image_url"
assert messages[2].content[1]["image_url"]["url"] == "data:image/jpeg;base64,SCREENSHOT_3"
@@ -0,0 +1,289 @@
# 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 VisualStepSummarizer and Context Compressor in Flash profile."""
import asyncio
from unittest.mock import AsyncMock, Mock
import pytest
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from artemis.agents.flash.context_compressor import compress_flash_messages
from artemis.agents.flash.summarizer import VisualStepSummarizer
from artemis.context import ArtemisContext
from artemis.sdk.builders import Builders
@pytest.fixture
def mock_context():
ctx = Mock(spec=ArtemisContext)
ctx.data_engine = None
return ctx
@pytest.mark.asyncio
async def test_summarizer_dispatch_and_caching(mock_context):
"""Verify non-blocking dispatch and summary generation."""
summarizer = VisualStepSummarizer(mock_context)
mock_llm_response = AIMessage(
content="[Step 1 State: Tapped 'Login' button. Home view opened.]"
)
summarizer._llm = Mock()
summarizer._llm.ainvoke = AsyncMock(return_value=mock_llm_response)
summarizer.dispatch(
step_number=1,
action_name="click",
action_args={"target": [500, 600]},
pre_img_bytes=b"fake_pre",
post_img_bytes=b"fake_post",
exec_outcome="Clicked successfully",
)
# Await background task flush
await summarizer.flush()
assert summarizer.has_summary(1)
summary = summarizer.get_summary(1)
assert "[Step 1 State: Tapped 'Login' button. Home view opened.]" in summary
# Verify memory optimization: image buffers are released
assert summarizer._step_inputs[1]["pre_img_bytes"] is None
assert summarizer._step_inputs[1]["post_img_bytes"] is None
@pytest.mark.asyncio
async def test_summarizer_retrigger_stalled_steps(mock_context):
"""Verify that earlier stalled steps are re-triggered when subsequent steps arrive."""
summarizer = VisualStepSummarizer(mock_context)
attempt_count = 0
async def mock_ainvoke(messages):
nonlocal attempt_count
attempt_count += 1
# Fail the first attempt (Step 1 first call)
if attempt_count == 1:
raise asyncio.CancelledError("Simulated timeout")
# Check messages for step number
msg_str = str(messages)
if "Step 1" in msg_str:
return AIMessage(content="[Step 1 State: Recovered Step 1 summary]")
return AIMessage(content="[Step 2 State: Step 2 summary]")
summarizer._llm = Mock()
summarizer._llm.ainvoke = AsyncMock(side_effect=mock_ainvoke)
# Dispatch step 1 (fails on first attempt)
summarizer.dispatch(
step_number=1,
action_name="click",
action_args={"target": [100, 100]},
pre_img_bytes=b"pre1",
post_img_bytes=b"post1",
exec_outcome="Outcome 1",
)
await summarizer.flush()
assert not summarizer.has_summary(1)
# Dispatch step 2 (should detect stalled Step 1 and re-trigger)
summarizer.dispatch(
step_number=2,
action_name="swipe",
action_args={"action": "up"},
pre_img_bytes=b"pre2",
post_img_bytes=b"post2",
exec_outcome="Outcome 2",
)
await summarizer.flush()
assert summarizer.has_summary(1)
assert summarizer.has_summary(2)
assert "Recovered Step 1" in summarizer.get_summary(1)
def test_context_compressor_with_ready_summaries(mock_context):
"""Verify that ready summaries replace past image blocks and prune outdated XML."""
summarizer = VisualStepSummarizer(mock_context)
summarizer._summaries[1] = "[Step 1 State: Tapped Search bar. Keyboard appeared.]"
summarizer._summaries[2] = "[Step 2 State: Typed 'flights'. Suggestions displayed.]"
messages = [
HumanMessage(
content=[
{"type": "text", "text": "Task: Search flights"},
{"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,IMG_1"}},
{"type": "text", "text": "--- UI Element List ---\nOld Tree 1"},
]
),
ToolMessage(
tool_call_id="tc1",
name="click",
content=[
{"type": "text", "text": "Action 'click' completed."},
{"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,IMG_2"}},
{"type": "text", "text": "--- UI Element List ---\nOld Tree 2"},
],
),
ToolMessage(
tool_call_id="tc2",
name="input_text",
content=[
{"type": "text", "text": "Action 'input_text' completed."},
{"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,IMG_3"}},
{"type": "text", "text": "--- UI Element List ---\nLive Latest Tree 3"},
],
),
]
compress_flash_messages(messages, summarizer=summarizer, prune_history_xml=True)
# Initial HumanMessage: Historical initial image pruned, task text preserved
assert len(messages[0].content) == 1
assert messages[0].content[0]["text"] == "Task: Search flights"
# Step 1 (ToolMessage 1): Image replaced with Step 1 summary, old XML pruned
assert len(messages[1].content) == 2
assert messages[1].content[0]["text"] == "Action 'click' completed."
assert messages[1].content[1]["text"] == "[Step 1 State: Tapped Search bar. Keyboard appeared.]"
# Step 2 (ToolMessage 2 - Latest live screen): Image and latest live XML MUST be preserved
assert len(messages[2].content) == 3
assert messages[2].content[0]["text"] == "Action 'input_text' completed."
assert messages[2].content[1]["type"] == "image_url"
assert "Live Latest Tree 3" in messages[2].content[2]["text"]
def test_context_compressor_graceful_fallback(mock_context):
"""Verify that when a summary is not ready yet, it cleanly falls back to standard pruned state."""
summarizer = VisualStepSummarizer(mock_context)
# Step 1 has no summary in cache
messages = [
HumanMessage(
content=[
{"type": "text", "text": "Task: Search"},
{"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,IMG_1"}},
]
),
ToolMessage(
tool_call_id="tc1",
name="click",
content=[
{"type": "text", "text": "Action 'click' completed."},
{"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,IMG_2"}},
],
),
]
compress_flash_messages(messages, summarizer=summarizer, prune_history_xml=True)
# Turn 1 should simply have image omitted (standard fallback)
assert len(messages[0].content) == 1
assert messages[0].content[0]["text"] == "Task: Search"
# Turn 2 (Latest) preserves image
assert len(messages[1].content) == 2
assert messages[1].content[1]["type"] == "image_url"
def test_flash_config_and_builder():
"""Verify Flash profile configuration model and SDK builder fluent API."""
cfg = Builders.AgentConfig.with_flash_config(
max_turns=25,
explorer_mode="flash",
step_summarizer=True,
step_summarizer_model="gemini-2.5-flash-lite",
prune_history_xml=True,
).build()
assert cfg.flash.max_turns == 25
assert cfg.flash.explorer_mode == "flash"
assert cfg.flash.step_summarizer.enabled is True
assert cfg.flash.step_summarizer.model == "gemini-2.5-flash-lite"
assert cfg.flash.step_summarizer.prune_history_xml is True
@pytest.mark.asyncio
async def test_data_engine_integration_with_step_summarizer(mock_context):
"""Verify that generated summaries sync into DataEngine and are visible to get_agent_friendly_steps."""
mock_de = Mock()
mock_de.update_step_summary = Mock()
mock_context.data_engine = mock_de
summarizer = VisualStepSummarizer(mock_context)
summarizer._llm = Mock()
summarizer._llm.ainvoke = AsyncMock(
return_value=AIMessage(content="[Step 1 State: Input text confirmed.]")
)
summarizer.dispatch(
step_number=1,
action_name="input_text",
action_args={"text": "hello"},
pre_img_bytes=b"pre",
post_img_bytes=b"post",
exec_outcome="Executed typing",
)
await summarizer.flush()
mock_de.update_step_summary.assert_called_once_with(1, "[Step 1 State: Input text confirmed.]")
def test_draw_action_overlay_rendering():
"""Verify that draw_action_overlay_on_image correctly renders markers on images."""
import io
from PIL import Image
from artemis.utils.visualization import draw_action_overlay_on_image
# Create a blank RGB image in memory
img = Image.new("RGB", (1080, 2400), color="white")
buf = io.BytesIO()
img.save(buf, format="JPEG")
raw_bytes = buf.getvalue()
# 1. Click overlay
annotated_click = draw_action_overlay_on_image(
image_bytes=raw_bytes,
action_name="click",
action_args={"target": [540, 1200]},
)
assert len(annotated_click) > 0
assert annotated_click != raw_bytes
# 2. Swipe overlay
annotated_swipe = draw_action_overlay_on_image(
image_bytes=raw_bytes,
action_name="swipe",
action_args={"start": [500, 1800], "end": [500, 600]},
)
assert len(annotated_swipe) > 0
assert annotated_swipe != raw_bytes
# 3. Click Sequence overlay
annotated_seq = draw_action_overlay_on_image(
image_bytes=raw_bytes,
action_name="click_sequence",
action_args={"sequence": [[200, 300], [400, 500], [600, 700]]},
)
assert len(annotated_seq) > 0
assert annotated_seq != raw_bytes
# 4. Fail-safe on corrupt bytes
corrupt_result = draw_action_overlay_on_image(
image_bytes=b"invalid_bytes",
action_name="click",
action_args={"target": [100, 100]},
)
assert corrupt_result == b"invalid_bytes"
+377
View File
@@ -0,0 +1,377 @@
# 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.
import json
from pathlib import Path
from unittest.mock import MagicMock, patch
from langchain_core.messages import ToolMessage
from artemis.agents.history_analyzer.history_analyzer import HistoryAnalyzer
from artemis.context import ArtemisContext
import pytest
@pytest.mark.asyncio
async def test_history_analyzer_no_history():
# Mock context with a data engine that has no steps
mock_ctx = MagicMock(spec=ArtemisContext)
mock_ctx.data_engine = MagicMock()
mock_ctx.data_engine.get_agent_friendly_steps.return_value = []
analyzer = HistoryAnalyzer(mock_ctx)
result = await analyzer.run("What happened?")
assert result == "No history recorded for this session yet."
@pytest.mark.asyncio
async def test_history_analyzer_simple_query_no_tool_call():
# Mock context and data engine
mock_ctx = MagicMock(spec=ArtemisContext)
mock_ctx.data_engine = MagicMock()
mock_ctx.data_engine.base_dir = "/tmp/fake_traces"
steps = [
{
"step_number": 1,
"relative_time": "1.2s",
"summary": "Opened the settings app",
},
{
"step_number": 2,
"relative_time": "4.5s",
"summary": "Toggled the wifi switch",
},
]
mock_ctx.data_engine.get_agent_friendly_steps.return_value = steps
# Mock LLM to return a direct text response (no tool calls)
mock_response = MagicMock()
mock_response.content = "First you opened settings, then you toggled wifi."
mock_response.tool_calls = []
mock_llm = MagicMock()
mock_llm.bind_tools.return_value = mock_llm
async def mock_astream(*args, **kwargs):
yield mock_response
mock_llm.astream.side_effect = mock_astream
with (
patch(
"artemis.agents.history_analyzer.history_analyzer.get_llm",
return_value=mock_llm,
),
patch("pathlib.Path.exists", return_value=False), # Force system prompt fallback
):
analyzer = HistoryAnalyzer(mock_ctx)
result = await analyzer.run("Summarize the session")
assert result == "First you opened settings, then you toggled wifi."
assert mock_llm.astream.called
@pytest.mark.asyncio
async def test_history_analyzer_detailed_query_with_tool_call():
# Mock context and data engine with full details
mock_ctx = MagicMock(spec=ArtemisContext)
mock_ctx.data_engine = MagicMock()
mock_ctx.data_engine.base_dir = "/tmp/fake_traces"
steps = [
{
"step_number": 1,
"relative_time": "1.2s",
"summary": "Opened the settings app",
"action_taken": {"action": "tap", "coordinates": [500, 600]},
"operator_raw_thinking": "Need to open settings to configure wifi",
"last_execution_result": {"status": "success"},
},
{
"step_number": 2,
"relative_time": "4.5s",
"summary": "Toggled the wifi switch",
"action_taken": {"action": "tap", "coordinates": [100, 200]},
"operator_raw_thinking": "Toggle the switch to turn wifi on",
"last_execution_result": {"status": "success"},
},
]
mock_ctx.data_engine.get_agent_friendly_steps.return_value = steps
# Turn 1 LLM response: Tool call to get_step_details for step 2
mock_response_turn_1 = MagicMock()
mock_response_turn_1.content = ""
mock_tool_call = {
"name": "get_step_details",
"args": {"start_step": 2, "end_step": 2},
"id": "call_123456",
}
mock_response_turn_1.tool_calls = [mock_tool_call]
# Turn 2 LLM response: Final answer using the details
mock_response_turn_2 = MagicMock()
mock_response_turn_2.content = (
"In step 2, the operator was thinking: 'Toggle the switch to turn wifi on'."
)
mock_response_turn_2.tool_calls = []
mock_llm = MagicMock()
mock_llm.bind_tools.return_value = mock_llm
responses = [mock_response_turn_1, mock_response_turn_2]
call_count = 0
async def mock_astream(*args, **kwargs):
nonlocal call_count
if call_count < len(responses):
yield responses[call_count]
call_count += 1
mock_llm.astream.side_effect = mock_astream
with (
patch(
"artemis.agents.history_analyzer.history_analyzer.get_llm",
return_value=mock_llm,
),
patch("pathlib.Path.exists", return_value=False), # Force fallback prompt
):
analyzer = HistoryAnalyzer(mock_ctx)
result = await analyzer.run("What was the operator thinking in step 2?")
assert (
result == "In step 2, the operator was thinking: 'Toggle the switch to turn wifi on'."
)
assert mock_llm.astream.call_count == 2
# Verify the tool was invoked with correct parameters internally
tool_msg = mock_llm.astream.call_args_list[1][0][0][
-2
] # ToolMessage is second to last after final response is appended
assert isinstance(tool_msg, ToolMessage)
assert tool_msg.tool_call_id == "call_123456"
# Parse the JSON in the ToolMessage content
tool_result = json.loads(tool_msg.content)
assert len(tool_result) == 1
assert tool_result[0]["step_number"] == 2
assert tool_result[0]["operator_raw_thinking"] == "Toggle the switch to turn wifi on"
assert tool_result[0]["action_taken"] == {
"action": "tap",
"coordinates": [100, 200],
}
@pytest.mark.asyncio
async def test_history_analyzer_read_note_tool_call():
mock_ctx = MagicMock(spec=ArtemisContext)
mock_ctx.data_engine = MagicMock()
mock_ctx.data_engine.base_dir = "/tmp/fake_traces"
steps = [{"step_number": 1, "relative_time": "1.0s", "summary": "Initiated"}]
mock_ctx.data_engine.get_agent_friendly_steps.return_value = steps
# Turn 1 LLM response: Tool call to list_notes
mock_response_turn_1 = MagicMock()
mock_response_turn_1.content = ""
mock_tool_call_list = {"name": "list_notes", "args": {}, "id": "call_list"}
mock_response_turn_1.tool_calls = [mock_tool_call_list]
# Turn 2 LLM response: Tool call to read_note for tactical_plan
mock_response_turn_2 = MagicMock()
mock_response_turn_2.content = ""
mock_tool_call_read = {
"name": "read_note",
"args": {"key": "tactical_plan"},
"id": "call_read",
}
mock_response_turn_2.tool_calls = [mock_tool_call_read]
# Turn 3 LLM response: Final answer
mock_response_turn_3 = MagicMock()
mock_response_turn_3.content = "The plan was to toggle wifi."
mock_response_turn_3.tool_calls = []
mock_llm = MagicMock()
mock_llm.bind_tools.return_value = mock_llm
responses = [
mock_response_turn_1,
mock_response_turn_2,
mock_response_turn_3,
]
call_count = 0
async def mock_astream(*args, **kwargs):
nonlocal call_count
if call_count < len(responses):
yield responses[call_count]
call_count += 1
mock_llm.astream.side_effect = mock_astream
# Mock the file system read using patch
fake_plan_content = "Goal: Toggle wifi"
def mock_exists(path):
# Let Path("/tmp/fake_traces/notes").exists() and notes/tactical_plan.md exist
return True
def mock_read_text(self, encoding="utf-8"):
return fake_plan_content
def mock_glob(self, pattern):
return [Path("/tmp/fake_traces/notes/tactical_plan.md")]
with (
patch(
"artemis.agents.history_analyzer.history_analyzer.get_llm",
return_value=mock_llm,
),
patch("pathlib.Path.exists", mock_exists),
patch("pathlib.Path.read_text", mock_read_text),
patch("pathlib.Path.glob", mock_glob),
):
analyzer = HistoryAnalyzer(mock_ctx)
result = await analyzer.run("What is the saved tactical plan?")
assert result == "The plan was to toggle wifi."
assert mock_llm.astream.call_count == 3
# Verify list_notes response
list_msg = mock_llm.astream.call_args_list[1][0][0][
3
] # Index 3 contains the ToolMessage for list_notes
assert isinstance(list_msg, ToolMessage)
assert list_msg.tool_call_id == "call_list"
assert list_msg.content == "Here are all the notes:\n- tactical_plan (1 lines)"
# Verify read_note response
read_msg = mock_llm.astream.call_args_list[2][0][0][
5
] # Index 5 contains the ToolMessage for read_note
assert isinstance(read_msg, ToolMessage)
assert read_msg.tool_call_id == "call_read"
assert (
read_msg.content == "Successfully read note 'tactical_plan'. 'tactical_plan' note"
f" content:\n{fake_plan_content}"
)
def test_history_analyzer_robust_tools_behavior():
mock_ctx = MagicMock(spec=ArtemisContext)
mock_ctx.data_engine = MagicMock()
mock_ctx.data_engine.base_dir = "/tmp/fake_traces"
steps = [
{
"step_number": 1,
"relative_time": "1.0s",
"summary": "Action 1",
"action_taken": {"action": "tap"},
}
]
analyzer = HistoryAnalyzer(mock_ctx)
# 1. Test get_step_details with string inputs
details_tool = analyzer._get_step_details_tool(steps)
result_details = details_tool.invoke({"start_step": "1", "end_step": "1"})
parsed_result = json.loads(result_details)
assert len(parsed_result) == 1
assert parsed_result[0]["step_number"] == 1
# 2. Test read_note with .md suffix
from artemis.tools.scratchpad import get_read_note_tool_pure
read_tool = get_read_note_tool_pure(analyzer.ctx)
def mock_exists(path):
return "tactical_plan.md" in str(path)
with (
patch("pathlib.Path.exists", mock_exists),
patch("pathlib.Path.read_text", return_value="Tactical plan content"),
):
result_read = read_tool.invoke({"key": "tactical_plan.md"})
assert (
result_read == "Successfully read note 'tactical_plan.md'. 'tactical_plan.md'"
" note content:\nTactical plan content"
)
@pytest.mark.asyncio
async def test_history_analyzer_integration_with_task_tree():
mock_ctx = MagicMock(spec=ArtemisContext)
mock_ctx.data_engine = MagicMock()
mock_ctx.data_engine.base_dir = "/tmp/fake_traces"
steps = [
{
"step_id": "step_1",
"step_number": 1,
"relative_time": "1.0s",
"summary": "Step 1",
},
]
mock_ctx.data_engine.get_agent_friendly_steps.return_value = steps
mock_response = MagicMock()
mock_response.content = "Answer"
mock_response.tool_calls = []
mock_llm = MagicMock()
mock_llm.bind_tools.return_value = mock_llm
async def mock_astream(*args, **kwargs):
yield mock_response
mock_llm.astream.side_effect = mock_astream
fake_plan = """- [/] Active Subgoal
- [ ] Pending Subgoal"""
mock_path = MagicMock()
mock_path.exists.return_value = True
mock_path.read_text.return_value = fake_plan
# Mock build_plan_and_history
with (
patch(
"artemis.agents.history_analyzer.history_analyzer.get_llm",
return_value=mock_llm,
),
patch(
"artemis.agents.history_analyzer.history_analyzer.get_note_file_path",
return_value=mock_path,
),
patch(
"artemis.agents.history_analyzer.history_analyzer.build_plan_and_history"
) as mock_build,
):
mock_build.return_value = "Mocked operation history"
analyzer = HistoryAnalyzer(mock_ctx)
await analyzer.run("query")
# Verify build_plan_and_history call args
mock_build.assert_called_once()
args, kwargs = mock_build.call_args
assert args[0] == fake_plan
assert args[1] == steps
import hashlib
expected_hash = hashlib.md5(b"Active Subgoal").hexdigest()
assert args[2] == expected_hash
assert kwargs.get("last_n_detailed") == 1
assert kwargs.get("min_summaries") == len(steps)
File diff suppressed because it is too large. Load diff
+508
View File
@@ -0,0 +1,508 @@
# 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, Mock, patch
from pydantic import BaseModel
import pytest
from artemis.agents.outputter.outputter import outputter
from artemis.config import LLM, OutputConfig # noqa: E402
from artemis.context import ArtemisContext # noqa: E402
from artemis.utils.logger import get_logger # noqa: E402
logger = get_logger(__name__)
class _CyFunctionDetectorMeta(type):
def __instancecheck__(self, instance):
name = type(instance).__name__
return (
name
in (
"cyfunction",
"cython_function_or_method",
"builtin_function_or_method",
)
or "cyfunction" in name.lower()
)
class CyFunctionDetector(metaclass=_CyFunctionDetectorMeta):
pass
class MockPydanticSchema(BaseModel):
model_config = {"ignored_types": (CyFunctionDetector,)}
color: str
price: float
currency_symbol: str
website_url: str
mock_dict = {
"color": "green",
"price": 20,
"currency_symbol": "$",
"website_url": "http://superwebsite.fr",
}
class DummyState:
def __init__(self, messages, initial_goal, operator_raw_data=None):
self.messages = messages
self.initial_goal = initial_goal
self.operator_raw_data = operator_raw_data
@pytest.fixture
def mock_context():
"""Create a properly mocked context with all required fields."""
ctx = Mock(spec=ArtemisContext)
ctx.llm_config = {
"planner": LLM(provider="openai", model="gpt-5-nano"),
"operator": LLM(provider="openai", model="gpt-5-nano"),
"validator": LLM(provider="openai", model="gpt-5-nano"),
}
ctx.device = Mock()
ctx.data_engine = None
return ctx
@pytest.fixture
def mock_state():
"""Create a mock state with test data."""
return DummyState(
messages=[],
initial_goal="Find a green product on my website",
)
def setup_mock_llm(mock_get_llm, react_response_content="Paris", structured_response=None):
mock_llm = Mock()
mock_llm_fallback = Mock()
# Mock bind_tools
mock_llm_with_tools = Mock()
mock_llm_with_tools.ainvoke = AsyncMock()
mock_react_response = Mock()
mock_react_response.content = react_response_content
mock_react_response.tool_calls = []
mock_llm_with_tools.ainvoke.return_value = mock_react_response
mock_llm.bind_tools.return_value = mock_llm_with_tools
mock_llm_fallback.bind_tools.return_value = mock_llm_with_tools
# Mock with_structured_output
mock_structured_llm = Mock()
mock_structured_llm.ainvoke = AsyncMock()
if structured_response is not None:
mock_structured_llm.ainvoke.return_value = structured_response
mock_llm.with_structured_output.return_value = mock_structured_llm
mock_llm_fallback.with_structured_output.return_value = mock_structured_llm
def get_llm_side_effect(ctx, name, is_utils=False, use_fallback=False):
if use_fallback:
return mock_llm_fallback
return mock_llm
mock_get_llm.side_effect = get_llm_side_effect
return mock_llm, mock_llm_with_tools, mock_structured_llm
@patch("artemis.agents.outputter.outputter.get_llm")
@pytest.mark.asyncio
async def test_outputter_with_pydantic_model(mock_get_llm, mock_context, mock_state):
"""Test outputter with Pydantic model output."""
expected_structured = MockPydanticSchema(
color="green",
price=20,
currency_symbol="$",
website_url="http://superwebsite.fr",
)
setup_mock_llm(
mock_get_llm,
react_response_content="Raw details about green product",
structured_response=expected_structured,
)
config = OutputConfig(
structured_output=MockPydanticSchema,
output_description=None,
)
result = await outputter(ctx=mock_context, output_config=config, graph_output=mock_state)
assert isinstance(result, dict)
assert result.get("color") == "green"
@patch("artemis.agents.outputter.outputter.get_llm")
@pytest.mark.asyncio
async def test_outputter_with_dict(mock_get_llm, mock_context, mock_state):
"""Test outputter with dictionary output."""
expected_dict = {
"color": "green",
"price": 20,
"currency_symbol": "$",
"website_url": "http://superwebsite.fr",
}
setup_mock_llm(
mock_get_llm,
react_response_content="Raw details",
structured_response=expected_dict,
)
config = OutputConfig(
structured_output=mock_dict,
output_description=None,
)
result = await outputter(ctx=mock_context, output_config=config, graph_output=mock_state)
assert isinstance(result, dict)
assert result.get("color") == "green"
assert result.get("price") == 20
assert result.get("currency_symbol") == "$"
assert result.get("website_url") == "http://superwebsite.fr"
@patch("artemis.agents.outputter.outputter.get_llm")
@pytest.mark.asyncio
async def test_outputter_with_natural_language_output(mock_get_llm, mock_context, mock_state):
"""Test outputter with natural language description output (returns JSON string)."""
expected_json = (
'{"color": "green", "price": 20, "currency_symbol": "$", "website_url":'
' "http://superwebsite.fr"}'
)
setup_mock_llm(mock_get_llm, react_response_content=expected_json)
config = OutputConfig(
structured_output=None,
output_description=(
"A JSON object with a color, a price, a currency_symbol and a website_url key"
),
)
result = await outputter(ctx=mock_context, output_config=config, graph_output=mock_state)
assert isinstance(result, dict)
assert result.get("color") == "green"
assert result.get("price") == 20
assert result.get("currency_symbol") == "$"
assert result.get("website_url") == "http://superwebsite.fr"
@patch("artemis.agents.outputter.outputter.get_llm")
@pytest.mark.asyncio
async def test_outputter_with_history_and_image(mock_get_llm, mock_context):
"""Test outputter with explicit history and image."""
expected_json = '{"answer": "Paris"}'
_, mock_llm_with_tools, _ = setup_mock_llm(mock_get_llm, react_response_content=expected_json)
config = OutputConfig(
structured_output=None,
output_description="The capital of France",
)
class StateWithRawData:
def __init__(self, messages, initial_goal, operator_raw_data):
self.messages = messages
self.initial_goal = initial_goal
self.operator_raw_data = operator_raw_data
state = StateWithRawData(
messages=[],
initial_goal="What is the capital of France?",
operator_raw_data={"screenshot_b64": "mock_b64_data"},
)
await outputter(
ctx=mock_context,
output_config=config,
graph_output=state,
plan_and_history="Plan: Find capital\nStep 1: Opened browser",
)
call_args = mock_llm_with_tools.ainvoke.call_args
assert call_args is not None
messages = call_args[0][0]
assert (
"Your sole objective is to verify whether the user's initial goal was"
" achieved" in messages[0].content
)
human_content = messages[1].content
assert isinstance(human_content, list)
assert human_content[0]["type"] == "text"
assert "Step 1: Opened browser" in human_content[0]["text"]
assert "Plan: Find capital" in human_content[0]["text"]
assert human_content[1]["type"] == "image_url"
assert human_content[1]["image_url"]["url"] == "data:image/jpeg;base64,mock_b64_data"
@patch("artemis.agents.outputter.outputter.get_llm")
@pytest.mark.asyncio
async def test_outputter_builds_concise_history(mock_get_llm):
"""Test that the outputter dynamically builds clean Summary ➡️ Action history."""
expected_json = '{"answer": "Success"}'
_, mock_llm_with_tools, _ = setup_mock_llm(mock_get_llm, react_response_content=expected_json)
config = OutputConfig(
structured_output=None,
output_description="Test output",
)
ctx = Mock(spec=ArtemisContext)
ctx.llm_config = {
"planner": LLM(provider="openai", model="gpt-5-nano"),
"operator": LLM(provider="openai", model="gpt-5-nano"),
"validator": LLM(provider="openai", model="gpt-5-nano"),
}
ctx.device = Mock()
mock_data_engine = Mock()
steps = [
{
"step_id": "step_1",
"step_number": 1,
"relative_time": "2.5s",
"summary": "Tapped search button",
"action_taken": [{"action": "click", "target_text": "Search"}],
"last_execution_result": {"status": "success"},
}
]
mock_data_engine.get_agent_friendly_steps.return_value = steps
mock_data_engine.base_dir = "/tmp/mock_session"
ctx.data_engine = mock_data_engine
class StateWithRawData:
def __init__(self, messages, initial_goal, operator_raw_data):
self.messages = messages
self.initial_goal = initial_goal
self.operator_raw_data = operator_raw_data
state = StateWithRawData(
messages=[],
initial_goal="Search for something",
operator_raw_data=None,
)
# Test Case 1: Fallback path (no plan file)
from pathlib import Path
original_exists = Path.exists
def mock_exists_fallback(self, *args, **kwargs):
if self.name == "task_plan.md":
return False
return original_exists(self, *args, **kwargs)
with patch("pathlib.Path.exists", mock_exists_fallback):
await outputter(
ctx=ctx,
output_config=config,
graph_output=state,
)
call_args = mock_llm_with_tools.ainvoke.call_args
assert call_args is not None
messages = call_args[0][0]
human_content_fallback = messages[1].content[0]["text"]
assert "--- Execution History ---" in human_content_fallback
assert "- *Step 1 (Start: 2.5s): Tapped search button*" in human_content_fallback
assert "*Action*: Tapped 'Search' at None" in human_content_fallback
# Reset mock call history
mock_llm_with_tools.ainvoke.reset_mock()
# Test Case 2: Standard path (plan file exists)
def mock_exists_standard(self, *args, **kwargs):
if self.name == "task_plan.md":
return True
return original_exists(self, *args, **kwargs)
original_read_text = Path.read_text
def mock_read_text(self, *args, **kwargs):
if self.name == "task_plan.md":
return "- [ ] Goal 1\n - [/] Subgoal 1.1"
return original_read_text(self, *args, **kwargs)
with (
patch("pathlib.Path.exists", mock_exists_standard),
patch("pathlib.Path.read_text", mock_read_text),
):
await outputter(
ctx=ctx,
output_config=config,
graph_output=state,
)
call_args = mock_llm_with_tools.ainvoke.call_args
assert call_args is not None
messages = call_args[0][0]
human_content_standard = messages[1].content[0]["text"]
assert "--- Execution History ---" in human_content_standard
assert "- *Step 1 (Start: 2.5s): Tapped search button*" in human_content_standard
assert "*Action*: Tapped 'Search' at None" in human_content_standard
@patch("artemis.tools.video_tool.VideoAnalyzer")
@patch("artemis.agents.outputter.outputter.get_llm")
@pytest.mark.asyncio
async def test_outputter_executes_video_analyzer_tool(
mock_get_llm, mock_video_analyzer, mock_context, mock_state
):
"""Test that the outputter ReAct loop can successfully invoke the video_analyzer tool."""
# Setup mock VideoAnalyzer
mock_agent_instance = Mock()
mock_agent_instance.run = AsyncMock(
return_value=("The user played a video for 5 seconds.", "success")
)
mock_video_analyzer.return_value = mock_agent_instance
# Setup mock LLM with two turns
mock_llm = Mock()
mock_llm_fallback = Mock()
mock_llm_with_tools = Mock()
mock_llm_with_tools.ainvoke = AsyncMock()
# Turn 1: LLM calls video_analyzer tool
tool_call = {
"name": "video_analyzer",
"args": {
"time_description": "from 5s to 10s",
"purpose": "verify video played",
},
"id": "call_123",
}
msg_turn1 = Mock()
msg_turn1.content = ""
msg_turn1.tool_calls = [tool_call]
# Turn 2: LLM returns final answer
msg_turn2 = Mock()
msg_turn2.content = "Video played successfully."
msg_turn2.tool_calls = []
mock_llm_with_tools.ainvoke.side_effect = [msg_turn1, msg_turn2]
mock_llm.bind_tools.return_value = mock_llm_with_tools
mock_llm_fallback.bind_tools.return_value = mock_llm_with_tools
def get_llm_side_effect(ctx, name, is_utils=False, use_fallback=False):
if use_fallback:
return mock_llm_fallback
return mock_llm
mock_get_llm.side_effect = get_llm_side_effect
config = OutputConfig(
structured_output=None,
output_description="Verify video",
)
result = await outputter(ctx=mock_context, output_config=config, graph_output=mock_state)
# Assert VideoAnalyzer was called correctly
mock_video_analyzer.assert_called_once_with(mock_context)
mock_agent_instance.run.assert_called_once_with("from 5s to 10s", "verify video played")
# Assert LLM's ainvoke was called twice
assert mock_llm_with_tools.ainvoke.call_count == 2
# Assert tool message was appended (messages list is mutated in place,
# so it eventually has 5 elements)
sent_messages = mock_llm_with_tools.ainvoke.call_args_list[1][0][0]
assert len(sent_messages) == 5
# Verify the sequence of messages
assert sent_messages[2] == msg_turn1
tool_message = sent_messages[3]
from langchain_core.messages import ToolMessage
assert isinstance(tool_message, ToolMessage)
assert tool_message.content == "The user played a video for 5 seconds."
assert tool_message.status == "success"
assert sent_messages[4] == msg_turn2
# Assert final result
assert result == "Video played successfully."
@patch("artemis.agents.outputter.outputter.get_llm")
@pytest.mark.asyncio
async def test_outputter_executes_save_note_tool(mock_get_llm, mock_context, mock_state):
"""Test that the outputter ReAct loop can successfully invoke the save_note tool."""
# Setup mock LLM with two turns
mock_llm = Mock()
mock_llm_fallback = Mock()
mock_llm_with_tools = Mock()
mock_llm_with_tools.ainvoke = AsyncMock()
# Turn 1: LLM calls save_note tool
tool_call = {
"name": "save_note",
"args": {"key": "verification_code", "content": "123456"},
"id": "call_save_123",
}
msg_turn1 = Mock()
msg_turn1.content = ""
msg_turn1.tool_calls = [tool_call]
# Turn 2: LLM returns final answer
msg_turn2 = Mock()
msg_turn2.content = "Verification code is 123456."
msg_turn2.tool_calls = []
mock_llm_with_tools.ainvoke.side_effect = [msg_turn1, msg_turn2]
mock_llm.bind_tools.return_value = mock_llm_with_tools
mock_llm_fallback.bind_tools.return_value = mock_llm_with_tools
def get_llm_side_effect(ctx, name, is_utils=False, use_fallback=False):
if use_fallback:
return mock_llm_fallback
return mock_llm
mock_get_llm.side_effect = get_llm_side_effect
config = OutputConfig(
structured_output=None,
output_description="Save verification code to note",
)
# Setup mock data_engine on context to support saving notes
mock_data_engine = Mock()
mock_data_engine.base_dir = "/tmp/mock_session"
mock_context.data_engine = mock_data_engine
# Patch save_note_content utility to avoid writing to disk
with patch("artemis.tools.scratchpad.save_note_content") as mock_save_content:
result = await outputter(ctx=mock_context, output_config=config, graph_output=mock_state)
mock_save_content.assert_called_once_with(
"/tmp/mock_session", "verification_code", "123456"
)
# Assert final result
assert result == "Verification code is 123456."
@@ -0,0 +1,102 @@
# 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.
import json
from unittest.mock import MagicMock, mock_open, patch
from artemis.agents.outputter.tools import (
get_search_history_tool,
get_step_details_tool,
get_step_screenshot_tool,
)
from artemis.context import ArtemisContext
import pytest
@pytest.fixture
def mock_context():
ctx = MagicMock(spec=ArtemisContext)
ctx.data_engine = MagicMock()
ctx.data_engine.global_base_dir = "/tmp/fake_traces"
return ctx
def test_get_step_details_tool():
steps = [
{
"step_number": 1,
"relative_time": "1s",
"summary": "Step 1",
"action_taken": "tap",
},
{
"step_number": 2,
"relative_time": "2s",
"summary": "Step 2",
"action_taken": "swipe",
},
]
tool = get_step_details_tool(steps)
# Test valid range
result = tool.invoke({"start_step": 1, "end_step": 1})
parsed = json.loads(result)
assert len(parsed) == 1
assert parsed[0]["step_number"] == 1
assert parsed[0]["action_taken"] == "tap"
# Test invalid range
result_invalid = tool.invoke({"start_step": 3, "end_step": 4})
assert "No steps found" in result_invalid
def test_get_step_screenshot_tool(mock_context):
steps = [
{"step_number": 1, "pre_image_name": "img1", "post_image_name": "img2"},
]
tool = get_step_screenshot_tool(mock_context, steps)
# Mock open and exists
with (
patch("pathlib.Path.exists", return_value=True),
patch("builtins.open", mock_open(read_data=b"fake_image_bytes")),
):
result = tool.invoke({"step_number": 1})
assert isinstance(result, list)
assert result[0]["type"] == "text"
assert result[1]["type"] == "image_url"
assert "data:image/jpeg;base64" in result[1]["image_url"]["url"]
def test_search_history_tool(mock_context):
tool = get_search_history_tool(mock_context)
# Mock steps returned by storage
mock_step_1 = MagicMock()
mock_step_1.step_number = 1
mock_step_1.pre_image_name = "img1"
mock_step_1.post_image_name = None
mock_context.data_engine.storage.get_steps.return_value = [mock_step_1]
# Mock search_ui_by_hash
mock_context.data_engine.storage.search_ui_by_hash.return_value = [
{"type": "ocr", "matched_text": "verification code: 1234", "score": 0.9}
]
result = tool.invoke({"query": "verification"})
parsed = json.loads(result)
assert len(parsed) == 1
assert parsed[0]["step_number"] == 1
assert parsed[0]["matches"][0]["matched_text"] == "verification code: 1234"
+89
View File
@@ -0,0 +1,89 @@
# 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 Mock, patch
from artemis.agents.planner.planner import PlannerNode
from artemis.context import DevicePlatform, ArtemisContext
import pytest
class DummyState:
def __init__(
self,
initial_goal,
needs_replan=False,
operator_raw_data=None,
operator_replan_reason=None,
):
self.initial_goal = initial_goal
self.needs_replan = needs_replan
self.operator_raw_data = operator_raw_data
self.operator_replan_reason = operator_replan_reason
async def asanitize_update(self, ctx, update, agent):
return update
@pytest.fixture
def mock_context():
ctx = Mock(spec=ArtemisContext)
ctx.device = Mock()
ctx.device.mobile_platform = DevicePlatform.ANDROID
ctx.adb_client = Mock()
ctx.ui_adb_client = Mock()
mock_screen_data = Mock()
mock_screen_data.base64 = "YmFzZTY0"
mock_screen_data.elements = []
mock_screen_data.width = 1080
mock_screen_data.height = 2400
ctx.ui_adb_client.get_screen_data.return_value = mock_screen_data
ctx.data_engine = None
return ctx
@pytest.mark.asyncio
async def test_planner_initial_plan(mock_context):
state = DummyState(
initial_goal="Test Goal",
operator_raw_data={"screenshot_b64": "dummy_base64_string"},
)
node = PlannerNode(mock_context)
mock_llm = Mock()
mock_response = Mock()
mock_response.content = "- [ ] Subgoal 1"
mock_response.tool_calls = []
async def mock_astream(*args, **kwargs):
yield mock_response
mock_llm.astream.side_effect = mock_astream
mock_llm.bind_tools = Mock(return_value=mock_llm)
with patch("artemis.agents.planner.planner.get_llm", return_value=mock_llm):
await node(state)
assert mock_llm.astream.called
args, kwargs = mock_llm.astream.call_args
messages = args[0]
system_msg = messages[0].content
assert "## Role" in system_msg
human_msg_content = messages[1].content
assert isinstance(human_msg_content, list)
assert human_msg_content[0]["type"] == "text"
assert "Goal: Test Goal" in human_msg_content[0]["text"]
+112
View File
@@ -0,0 +1,112 @@
# 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.
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
from artemis.agents.operator.operator import OperatorNode
from artemis.context import ArtemisContext
from artemis.graph.graph import execution_check_node
from artemis.graph.state import State
import pytest
@pytest.mark.asyncio
async def test_snapshot_rollback_flow(tmp_path):
# Setup dummy directories
base_dir = tmp_path / "trace"
notes_dir = base_dir / "notes"
notes_dir.mkdir(parents=True)
task_plan_path = notes_dir / "task_plan.md"
task_plan_path.write_text("Initial Plan", encoding="utf-8")
# Mock Context
mock_ctx = MagicMock(spec=ArtemisContext)
mock_ctx.data_engine = MagicMock()
mock_ctx.data_engine.base_dir = base_dir
mock_ctx.execution_setup = None
# Mock State
mock_state = MagicMock(spec=State)
mock_state.subagent_calls = []
mock_state.asanitize_update = AsyncMock(return_value={"status": "success"})
mock_state.initial_goal = "Test goal"
mock_state.short_term_memory = ""
mock_state.operator_raw_data = {
"screenshot_b64": "YmFzZTY0",
"xml_hierarchy": "<hierarchy></hierarchy>",
"ocr_results": [],
}
mock_state.current_step_id = "test-step-id"
# Mock Controller and LLM to let Operator run minimally
mock_controller = MagicMock()
mock_controller.take_screenshot = AsyncMock(return_value="YmFzZTY0")
mock_device_data = MagicMock()
mock_device_data.elements = []
mock_controller.get_screen_data = AsyncMock(return_value=mock_device_data)
mock_llm = MagicMock()
mock_response = MagicMock()
mock_response.tool_calls = []
async def mock_astream(*args, **kwargs):
yield mock_response
mock_llm.astream.side_effect = mock_astream
mock_llm.bind_tools.return_value = mock_llm
# 1. Run Operator to trigger snapshot
with patch("artemis.agents.operator.operator.get_llm", return_value=mock_llm):
node = OperatorNode(mock_ctx)
await node(mock_state)
# Verify snapshot was created
snapshot_dir = notes_dir.with_name("notes_snapshot")
assert snapshot_dir.exists()
assert (snapshot_dir / "task_plan.md").read_text(encoding="utf-8") == "Initial Plan"
assert mock_ctx.task_plan_snapshot == snapshot_dir
# 2. Simulate Operator optimistic modification
task_plan_path.write_text("Modified Plan", encoding="utf-8")
# 3. Run execution_check_node with failure to trigger rollback
future = asyncio.Future()
future.set_result({"status": "failed"})
mock_ctx.checker_task = future
await execution_check_node(mock_state, mock_ctx)
# Verify rollback restored the file
assert task_plan_path.read_text(encoding="utf-8") == "Initial Plan"
# 4. Run Operator again to create new snapshot (simulate next attempt)
# First modify it again to simulate operator working on restored plan
task_plan_path.write_text("Modified Plan 2", encoding="utf-8")
with patch("artemis.agents.operator.operator.get_llm", return_value=mock_llm):
await node(mock_state)
assert (snapshot_dir / "task_plan.md").read_text(encoding="utf-8") == "Modified Plan 2"
# 5. Run execution_check_node with success to trigger commit (cleanup)
future_success = asyncio.Future()
future_success.set_result({"status": "success"})
mock_ctx.checker_task = future_success
await execution_check_node(mock_state, mock_ctx)
# Verify snapshot was deleted
assert not snapshot_dir.exists()
+216
View File
@@ -0,0 +1,216 @@
# 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.
import json
from unittest.mock import MagicMock, Mock, patch
from artemis.agents.summarizer.summarizer import SummarizerNode
from artemis.context import ArtemisContext
import pytest
class DummyState:
def __init__(
self,
structured_decisions=None,
operator_raw_thinking=None,
operator_native_thinking=None,
last_execution_result=None,
current_step_id=None,
):
self.structured_decisions = structured_decisions
self.operator_raw_thinking = operator_raw_thinking
self.operator_native_thinking = operator_native_thinking
self.last_execution_result = last_execution_result
self.current_step_id = current_step_id
@pytest.fixture
def mock_context():
ctx = Mock(spec=ArtemisContext)
ctx.device = None
ctx.data_engine = Mock()
ctx.background_tasks = []
return ctx
@pytest.mark.asyncio
async def test_summarizer_success(mock_context):
"""Test that SummarizerNode successfully generates summary from raw thinking and execution result."""
decisions = json.dumps([{"action": "tap", "coordinates": [500, 600]}])
thinking = "The operator clicked the search button."
result = {
"status": "success",
"executed_actions": [{"action": "tap", "coordinates": [500, 600]}],
}
state = DummyState(
structured_decisions=decisions,
operator_raw_thinking=thinking,
last_execution_result=result,
current_step_id="12345678-1234-5678-1234-567812345678",
)
# Mock LLM response
mock_response = Mock()
mock_response.content = "Step completed successfully."
mock_llm = MagicMock()
async def mock_astream(*args, **kwargs):
yield mock_response
mock_llm.astream.side_effect = mock_astream
with (
patch(
"artemis.agents.summarizer.summarizer.get_llm", return_value=mock_llm
) as mock_get_llm,
patch("asyncio.create_task") as mock_create_task,
):
# Run the node call
node = SummarizerNode(mock_context)
await node(state)
# Verify task was registered in context
assert len(mock_context.background_tasks) == 1
assert mock_context.background_tasks[0] == mock_create_task.return_value
# Verify background task was spawned
assert mock_create_task.called
background_coroutine = mock_create_task.call_args[0][0]
# Await the background task synchronously for test verification
await background_coroutine
# Verify LLM was invoked with proper structured messages
assert mock_llm.astream.called
invoked_messages = mock_llm.astream.call_args[0][0]
system_msg = invoked_messages[0].content
human_msg = invoked_messages[1].content
assert "Summarizer Agent" in system_msg
assert "Decisions made" in human_msg
assert "The operator clicked the search button." in human_msg
# Verify Data Engine updated the step summary
from uuid import UUID
mock_context.data_engine.update_step_summary.assert_called_once_with(
UUID("12345678-1234-5678-1234-567812345678"),
"Step completed successfully.",
)
@pytest.mark.asyncio
async def test_summarizer_with_native_thinking(mock_context):
"""Test that SummarizerNode successfully generates summary from both explicit and native thoughts."""
decisions = json.dumps([{"action": "tap", "coordinates": [500, 600]}])
raw_thinking = "Explicit monologue."
native_thinking = "Silent core reasoning."
result = {"status": "success"}
state = DummyState(
structured_decisions=decisions,
operator_raw_thinking=raw_thinking,
operator_native_thinking=native_thinking,
last_execution_result=result,
current_step_id="12345678-1234-5678-1234-567812345678",
)
mock_response = Mock()
mock_response.content = "Summary result."
mock_llm = MagicMock()
async def mock_astream(*args, **kwargs):
yield mock_response
mock_llm.astream.side_effect = mock_astream
with (
patch(
"artemis.agents.summarizer.summarizer.get_llm", return_value=mock_llm
) as mock_get_llm,
patch("asyncio.create_task") as mock_create_task,
):
node = SummarizerNode(mock_context)
await node(state)
# Verify task was registered in context
assert len(mock_context.background_tasks) == 1
assert mock_context.background_tasks[0] == mock_create_task.return_value
background_coroutine = mock_create_task.call_args[0][0]
await background_coroutine
assert mock_llm.astream.called
invoked_messages = mock_llm.astream.call_args[0][0]
human_msg = invoked_messages[1].content
assert "Operator explicit thoughts:" in human_msg
assert "Explicit monologue." in human_msg
assert "Operator native thoughts:" in human_msg
assert "Silent core reasoning." in human_msg
@pytest.mark.asyncio
async def test_summarizer_with_chronological_trace(mock_context):
"""Test that SummarizerNode generates a prompt containing the chronological step trace when available."""
step_id = "12345678-1234-5678-1234-567812345678"
state = DummyState(
current_step_id=step_id,
)
mock_step_data = {
"step_id": step_id,
"step_number": 1,
"relative_time": "5.0s",
"action_taken": [{"action": "click", "target_text": "Search"}],
"last_execution_result": {"status": "success"},
"interleaved_events": [
{"type": "thought", "content": "Let's click search."},
],
}
mock_context.data_engine.get_agent_friendly_steps.return_value = [mock_step_data]
# Mock LLM response
mock_response = Mock()
mock_response.content = "Summary using chronological trace."
mock_llm = MagicMock()
async def mock_astream(*args, **kwargs):
yield mock_response
mock_llm.astream.side_effect = mock_astream
with (
patch("artemis.agents.summarizer.summarizer.get_llm", return_value=mock_llm),
patch("asyncio.create_task") as mock_create_task,
):
node = SummarizerNode(mock_context)
await node(state)
# Verify task was registered in context
assert len(mock_context.background_tasks) == 1
assert mock_context.background_tasks[0] == mock_create_task.return_value
background_coroutine = mock_create_task.call_args[0][0]
await background_coroutine
assert mock_llm.astream.called
invoked_messages = mock_llm.astream.call_args[0][0]
human_msg = invoked_messages[1].content
assert "--- Execution History ---" in human_msg
assert "Let's click search." in human_msg
+913
View File
@@ -0,0 +1,913 @@
# 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.
import json
from unittest.mock import AsyncMock, MagicMock, Mock, patch
from langchain_core.messages import AIMessage
from artemis.agents.validator.failure_analyzer import ValidationErrorCategory
from artemis.agents.validator.validator import ValidatorNode
from artemis.context import ArtemisContext
import pytest
class DummyState:
def __init__(
self,
structured_decisions,
validator_messages=[],
current_step_id=None,
latest_screenshot=None,
):
self.structured_decisions = structured_decisions
self.validator_messages = validator_messages
self.current_step_id = current_step_id
self.latest_screenshot = latest_screenshot
async def asanitize_update(self, ctx, update, agent):
return update
@pytest.fixture
def mock_context(tmp_path):
ctx = Mock(spec=ArtemisContext)
ctx.llm_config = Mock()
ctx.data_engine = Mock()
ctx.data_engine.base_dir = tmp_path
ctx.data_engine.get_relative_time.return_value = 1.0
return ctx
@pytest.fixture
def temp_screenshot(tmp_path):
p = tmp_path / "screenshot.png"
p.write_bytes(b"dummy_data")
return str(p)
@pytest.fixture
def mock_mcp():
with (
patch("mcp.client.stdio.stdio_client") as mock_stdio,
patch("artemis.agents.validator.validator.ClientSession") as mock_session_cls,
):
mock_ctx = AsyncMock()
mock_ctx.__aenter__.return_value = (Mock(), Mock())
mock_stdio.return_value = mock_ctx
mock_session = AsyncMock()
mock_session.initialize = AsyncMock()
mock_session_cls.return_value = mock_session
yield mock_session
@pytest.mark.asyncio
async def test_validator_success(mock_mcp, mock_context, temp_screenshot):
"""Test that ValidatorNode executes actions successfully."""
# Mock MCP calls: screenshot, tap, screenshot
mock_mcp.call_tool.side_effect = [
Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")]), # pre-screenshot
Mock(content=[Mock(text="Success")]), # tap action
Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")]), # post-screenshot
]
decisions = json.dumps([{"action": "tap", "coordinates": [105, 205]}])
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
with patch("artemis.utils.image_diff.check_ui_change", return_value=True):
node = ValidatorNode(mock_context)
result = await node(state)
assert "last_execution_result" in result
report = result["last_execution_result"]
assert "execution" in report
assert len(report["execution"]) == 1
assert report["execution"][0]["action"] == "tap"
assert "attempts" not in report["execution"][0]
@pytest.mark.asyncio
async def test_validator_failure_analysis(mock_mcp, mock_context, temp_screenshot):
"""Test that ValidatorNode triggers failure analysis on error and handles
cannot_fix via tool.
"""
tap_count = 0
def mock_call_tool(name, args):
nonlocal tap_count
if name == "take_screenshot":
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
if name == "tap":
tap_count += 1
return Mock(content=[Mock(text="Error: Element not found")])
return Mock(content=[Mock(text="Success")])
mock_mcp.call_tool.side_effect = mock_call_tool
decisions = json.dumps([{"action": "tap", "coordinates": [105, 205]}])
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
node = ValidatorNode(mock_context)
with (
patch("artemis.agents.validator.failure_analyzer.FailureAnalyzer.analyze") as mock_analyze,
patch("artemis.utils.image_diff.check_ui_change", return_value=False),
):
mock_analyze.return_value = {
"status": "cannot_fix",
"analysis": "Analysis: Coordinates are likely wrong.",
}
result = await node(state)
assert "last_execution_result" in result
report = result["last_execution_result"]
assert "execution" in report
assert len(report["execution"]) == 1
assert report["execution"][0]["action"] == "tap"
assert report["execution"][0]["attempts"] == [
"Error: Element not found",
"Error: Element not found",
]
assert report["execution"][0]["repair"] == "Analysis: Coordinates are likely wrong."
@pytest.mark.asyncio
async def test_validator_repair_success(mock_mcp, mock_context, temp_screenshot):
"""Test that ValidatorNode terminates successfully and clears remaining actions
after successful repair.
"""
tap_count = 0
def mock_call_tool(name, args):
nonlocal tap_count
if name == "take_screenshot":
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
if name == "tap":
tap_count += 1
if tap_count <= 2: # First tap fails
return Mock(content=[Mock(text="Error: Element not found")])
return Mock(content=[Mock(text="Success")])
return Mock(content=[Mock(text="Success")])
mock_mcp.call_tool.side_effect = mock_call_tool
# Two actions planned initially
decisions = json.dumps(
[
{"action": "tap", "coordinates": [105, 205]},
{"action": "tap", "coordinates": [205, 305]},
]
)
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
node = ValidatorNode(mock_context)
with (
patch("artemis.agents.validator.failure_analyzer.FailureAnalyzer.analyze") as mock_analyze,
patch("artemis.utils.image_diff.check_ui_change", return_value=False),
):
mock_analyze.return_value = {
"status": "fixed",
"analysis": "Fixed by updating actions",
}
result = await node(state)
assert "last_execution_result" in result
report = result["last_execution_result"]
assert "execution" in report
assert len(report["execution"]) == 1 # Only the first failed action is in report
assert report["execution"][0]["coordinates"] == [105, 205]
assert report["execution"][0]["attempts"] == [
"Error: Element not found",
"Error: Element not found",
]
assert report["execution"][0]["repair"] == "Fixed by updating actions"
@pytest.mark.asyncio
async def test_validator_wait_for_delay(mock_mcp, mock_context, temp_screenshot):
"""Test that wait_for_delay executes successfully."""
def mock_call(name, args):
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
mock_mcp.call_tool.side_effect = mock_call
decisions = json.dumps([{"action": "wait_for_delay", "time_in_ms": 10}])
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
with patch("artemis.utils.image_diff.check_ui_change", return_value=False):
node = ValidatorNode(mock_context)
result = await node(state)
assert "last_execution_result" in result
report = result["last_execution_result"]
assert "execution" in report
assert len(report["execution"]) == 1
assert report["execution"][0]["action"] == "wait_for_delay"
@pytest.mark.asyncio
async def test_validator_focus_and_clear_text_no_ui_change(mock_mcp, mock_context, temp_screenshot):
"""Test that focus_and_clear_text executes successfully even if no UI change is detected."""
def mock_call(name, args):
if name == "focus_and_clear_text":
return Mock(content=[Mock(text="Success")])
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
mock_mcp.call_tool.side_effect = mock_call
decisions = json.dumps([{"action": "focus_and_clear_text", "coordinates": [400, 500]}])
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
with patch("artemis.utils.image_diff.check_ui_change", return_value=False):
node = ValidatorNode(mock_context)
result = await node(state)
assert "last_execution_result" in result
report = result["last_execution_result"]
assert "execution" in report
assert len(report["execution"]) == 1
assert report["execution"][0]["action"] == "focus_and_clear_text"
@pytest.mark.asyncio
async def test_validator_silent_failure_treated_as_success(mock_mcp, mock_context, temp_screenshot):
"""Test that silent failure (exec succeeds but no UI change) is treated as
success and does not trigger FailureAnalyzer.
"""
def mock_call_tool(name, args):
if name == "take_screenshot":
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
if name == "tap":
return Mock(content=[Mock(text="Success")])
return Mock(content=[Mock(text="Success")])
mock_mcp.call_tool.side_effect = mock_call_tool
decisions = json.dumps([{"action": "tap", "coordinates": [105, 205]}])
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
node = ValidatorNode(mock_context)
with (
patch("artemis.agents.validator.validator.VALIDATOR_POLL_TIMEOUT", 0.1),
patch("artemis.agents.validator.validator.VALIDATOR_POLL_INTERVAL", 0.01),
patch("artemis.agents.validator.failure_analyzer.FailureAnalyzer.analyze") as mock_analyze,
patch("artemis.utils.image_diff.check_ui_change", return_value=False), # No UI change
):
result = await node(state)
# Assert FailureAnalyzer was NOT called
mock_analyze.assert_not_called()
# Assert the action is considered executed successfully
assert "last_execution_result" in result
report = result["last_execution_result"]
assert "execution" in report
assert len(report["execution"]) == 1
assert report["execution"][0]["action"] == "tap"
assert "attempts" not in report["execution"][0]
@pytest.mark.asyncio
async def test_validator_records_skipped_actions_on_cannot_fix(
mock_mcp, mock_context, temp_screenshot
):
"""Test that ValidatorNode marks unexecuted actions as Skipped when repair fails."""
def mock_call_tool(name, args):
if name == "take_screenshot":
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
return Mock(content=[Mock(text="Error: Connection lost")])
mock_mcp.call_tool.side_effect = mock_call_tool
# Two actions planned
decisions = json.dumps(
[
{"action": "tap", "coordinates": [100, 200]},
{"action": "press_key", "keycode": "ENTER"},
]
)
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
node = ValidatorNode(mock_context)
with (
patch("artemis.agents.validator.failure_analyzer.FailureAnalyzer.analyze") as mock_analyze,
patch("artemis.utils.image_diff.check_ui_change", return_value=False),
):
mock_analyze.return_value = {
"status": "cannot_fix",
"analysis": "Cannot fix: System crashed.",
}
result = await node(state)
assert "last_execution_result" in result
report = result["last_execution_result"]
assert "execution" in report
assert len(report["execution"]) == 2
# First action failed
assert report["execution"][0]["action"] == "tap"
assert report["execution"][0]["attempts"] == [
"Error: Connection lost",
"Error: Connection lost",
]
assert report["execution"][0]["repair"] == "Cannot fix: System crashed."
# Second action was skipped
assert report["execution"][1]["action"] == "press_key"
assert report["execution"][1]["attempts"] == ["Skipped"]
@pytest.mark.asyncio
async def test_validator_pre_execution_validation_and_repair(
mock_mcp, mock_context, temp_screenshot
):
"""Test that pre-execution validation fails when element changes, and is
repaired by FailureAnalyzer.
"""
mock_live_elements = [
{
"text": "Sign Up",
"bounds": "[100,200][300,300]",
"resource-id": "btn_signup",
}
]
def mock_call_tool(name, args):
if name == "take_screenshot":
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
if name == "get_ui_hierarchy":
return Mock(content=[Mock(text=json.dumps(mock_live_elements))])
if name == "tap":
return Mock(content=[Mock(text="Success")])
return Mock(content=[Mock(text="Success")])
mock_mcp.call_tool.side_effect = mock_call_tool
decisions = json.dumps(
[
{
"action": "tap",
"coordinates": [200, 250],
"target_text": "Login",
"target_bounds": [100, 200, 300, 300],
"target_resource_id": "btn_login",
}
]
)
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
node = ValidatorNode(mock_context)
with (
patch("artemis.agents.validator.failure_analyzer.FailureAnalyzer.analyze") as mock_analyze,
patch("artemis.utils.image_diff.check_ui_change", return_value=True),
patch.object(
ValidatorNode,
"_validate_action_precondition_pixel",
new_callable=AsyncMock,
) as mock_pixel,
):
mock_pixel.return_value = (
False,
ValidationErrorCategory.PIXEL_TARGET_DISAPPEARED,
"Pixel check failed",
)
mock_analyze.return_value = {
"status": "fixed",
"analysis": ("Pre-exec failure matched and resolved by switching to Sign Up button."),
}
result = await node(state)
mock_analyze.assert_called_once()
args, kwargs = mock_analyze.call_args
assert "Pre-execution validation failed" in args[2]
assert "Login" in args[2]
assert kwargs.get("error_category") == ValidationErrorCategory.TARGET_OCCUPIED
assert "last_execution_result" in result
report = result["last_execution_result"]
assert len(report["execution"]) == 1
assert report["execution"][0]["action"] == "tap"
assert "Pre-execution validation failed" in report["execution"][0]["attempts"][0]
assert (
report["execution"][0]["repair"]
== "Pre-exec failure matched and resolved by switching to Sign Up"
" button."
)
@pytest.mark.asyncio
async def test_validator_pre_execution_validation_self_healing(
mock_mcp, mock_context, temp_screenshot
):
"""Test that pre-execution validation succeeds and self-heals when element shifts slightly."""
mock_live_elements = [
{
"text": "Login",
"bounds": "[100,220][300,320]",
"resource-id": "btn_login",
}
]
tapped_coordinates = None
def mock_call_tool(name, args):
nonlocal tapped_coordinates
if name == "take_screenshot":
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
if name == "get_ui_hierarchy":
return Mock(content=[Mock(text=json.dumps(mock_live_elements))])
if name == "tap":
tapped_coordinates = args.get("coordinates")
return Mock(content=[Mock(text="Success")])
return Mock(content=[Mock(text="Success")])
mock_mcp.call_tool.side_effect = mock_call_tool
decisions = json.dumps(
[
{
"action": "tap",
"coordinates": [200, 250],
"target_text": "Login",
"target_bounds": [100, 200, 300, 300],
"target_resource_id": "btn_login",
}
]
)
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
node = ValidatorNode(mock_context)
with patch("artemis.utils.image_diff.check_ui_change", return_value=True):
await node(state)
# Center of [100, 220][300, 320] is [200, 270]
assert tapped_coordinates == [200, 270]
@pytest.mark.asyncio
async def test_validator_pre_execution_validation_anonymous_occupant(
mock_mcp, mock_context, temp_screenshot
):
"""Test that pre-execution validation correctly flags TARGET_OCCUPIED
when coordinates are blocked by an anonymous clickable view.
"""
# The live screen contains an anonymous clickable view covering
# target coordinates [200, 250]
mock_live_elements = [
{
"text": "",
"bounds": "[150,200][250,300]",
"clickable": True,
"resource-id": "",
}
]
def mock_call_tool(name, args):
if name == "take_screenshot":
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
if name == "get_ui_hierarchy":
return Mock(content=[Mock(text=json.dumps(mock_live_elements))])
return Mock(content=[Mock(text="Success")])
mock_mcp.call_tool.side_effect = mock_call_tool
decisions = json.dumps(
[
{
"action": "tap",
"coordinates": [200, 250],
"target_text": "Login",
"target_bounds": [100, 200, 300, 300],
"target_resource_id": "btn_login",
}
]
)
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
node = ValidatorNode(mock_context)
with (
patch("artemis.agents.validator.failure_analyzer.FailureAnalyzer.analyze") as mock_analyze,
patch("artemis.utils.image_diff.check_ui_change", return_value=True),
patch.object(
ValidatorNode,
"_validate_action_precondition_pixel",
new_callable=AsyncMock,
) as mock_pixel,
):
mock_pixel.return_value = (
False,
ValidationErrorCategory.PIXEL_TARGET_DISAPPEARED,
"Pixel check failed",
)
mock_analyze.return_value = {
"status": "fixed",
"analysis": "Pre-exec failure bypassed.",
}
await node(state)
# Assert that the validator correctly classified this as TARGET_OCCUPIED
# (due to the clickable anonymous view blocking the click)
mock_analyze.assert_called_once()
args, kwargs = mock_analyze.call_args
assert kwargs.get("error_category") == ValidationErrorCategory.TARGET_OCCUPIED
assert "occupied/intercepted by a different element: interactive anonymous element" in args[2]
@pytest.mark.asyncio
async def test_validator_pixel_validation_success(mock_mcp, mock_context, temp_screenshot):
"""Test that the pixel safety net succeeds when Gemini reports target is present."""
mock_mcp.call_tool.side_effect = [
Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")]), # live screenshot for pixel check
Mock(content=[Mock(text="Success")]), # tap action
Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")]), # post-screenshot
]
decisions = json.dumps([{"action": "tap", "coordinates": [100, 200]}])
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
mock_agent_cfg = Mock()
mock_agent_cfg.provider = "google"
mock_agent_cfg.model = "gemini-3.5-flash-lite"
mock_context.llm_config.get_agent.return_value = mock_agent_cfg
mock_response = Mock()
mock_response.text = json.dumps(
{
"is_present": True,
"reason": "Button is clearly visible and clickable.",
"confidence": 0.95,
}
)
mock_llm = MagicMock()
mock_llm.ainvoke = AsyncMock(return_value=AIMessage(content=mock_response.text))
# Mock crop and draw helpers to return dummy bytes
with (
patch(
"artemis.utils.visualization.crop_and_annotate_target",
return_value=b"dummy_bytes",
),
patch("artemis.agents.validator.validator.get_llm", return_value=mock_llm),
patch("artemis.utils.image_diff.check_ui_change", return_value=True),
):
node = ValidatorNode(mock_context)
result = await node(state)
assert "last_execution_result" in result
report = result["last_execution_result"]
assert len(report["execution"]) == 1
assert "attempts" not in report["execution"][0]
@pytest.mark.asyncio
async def test_validator_pixel_validation_failure(mock_mcp, mock_context, temp_screenshot):
"""Test that the pixel safety net fails and triggers FailureAnalyzer
when Gemini reports target is missing.
"""
def mock_call_tool(name, args):
if name == "take_screenshot":
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
return Mock(content=[Mock(text="Success")])
mock_mcp.call_tool.side_effect = mock_call_tool
decisions = json.dumps([{"action": "tap", "coordinates": [100, 200]}])
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
mock_agent_cfg = Mock()
mock_agent_cfg.provider = "google"
mock_agent_cfg.model = "gemini-3.5-flash-lite"
mock_context.llm_config.get_agent.return_value = mock_agent_cfg
mock_response = Mock()
mock_response.text = json.dumps(
{
"is_present": False,
"reason": "The button has completely disappeared from the screen.",
"confidence": 0.90,
}
)
mock_llm = MagicMock()
mock_llm.ainvoke = AsyncMock(return_value=AIMessage(content=mock_response.text))
with (
patch(
"artemis.utils.visualization.crop_and_annotate_target",
return_value=b"dummy_bytes",
),
patch("artemis.agents.validator.validator.get_llm", return_value=mock_llm),
patch("artemis.utils.image_diff.check_ui_change", return_value=False),
patch("artemis.agents.validator.failure_analyzer.FailureAnalyzer.analyze") as mock_analyze,
):
mock_analyze.return_value = {
"status": "cannot_fix",
"analysis": "Pixel safety net flagged missing target.",
}
node = ValidatorNode(mock_context)
await node(state)
mock_analyze.assert_called_once()
args, kwargs = mock_analyze.call_args
assert "Pixel-level validation failed" in args[2]
assert kwargs.get("error_category") == ValidationErrorCategory.PIXEL_TARGET_DISAPPEARED
@pytest.mark.asyncio
@patch("artemis.agents.validator.validator.find_package")
@patch("artemis.utils.app_launch_utils.launch_app_with_retries")
async def test_validator_launch_app_passes_use_fallback_mcp(
mock_launch_app, mock_find_package, mock_mcp, mock_context, temp_screenshot
):
"""Test that ValidatorNode calls find_package with use_fallback=False
for launch_app action in MCP mode.
"""
mock_find_package.return_value = "com.example.app"
mock_launch_app.return_value = (True, None)
# Mock MCP calls: screenshot (pre), screenshot (post)
mock_mcp.call_tool.side_effect = [
Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")]), # pre-screenshot
Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")]), # post-screenshot
]
decisions = json.dumps([{"action": "launch_app", "app_name": "My App"}])
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
with patch("artemis.utils.image_diff.check_ui_change", return_value=True):
node = ValidatorNode(mock_context)
result = await node(state)
mock_find_package.assert_called_once_with(mock_context, "My App", use_fallback=False)
mock_launch_app.assert_called_once_with(mock_context, "com.example.app")
assert "last_execution_result" in result
@pytest.mark.asyncio
@patch("artemis.agents.validator.validator.find_package")
@patch("artemis.utils.app_launch_utils.launch_app_with_retries")
async def test_validator_launch_app_failure_no_retry(
mock_launch_app, mock_find_package, mock_mcp, mock_context, temp_screenshot
):
"""Test that ValidatorNode does NOT retry launch_app action on failure."""
mock_find_package.return_value = "com.example.app"
mock_launch_app.return_value = (False, "Error: Force close")
def mock_call_tool(name, args):
if name == "take_screenshot":
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
return Mock(content=[Mock(text="Success")])
mock_mcp.call_tool.side_effect = mock_call_tool
decisions = json.dumps([{"action": "launch_app", "app_name": "My App"}])
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
with (
patch("artemis.agents.validator.validator.VALIDATOR_POLL_TIMEOUT", 0.05),
patch("artemis.agents.validator.validator.VALIDATOR_POLL_INTERVAL", 0.01),
patch("artemis.agents.validator.failure_analyzer.FailureAnalyzer.analyze") as mock_analyze,
patch("artemis.utils.image_diff.check_ui_change", return_value=False),
):
mock_analyze.return_value = {
"status": "cannot_fix",
"analysis": "Cannot launch app.",
}
node = ValidatorNode(mock_context)
result = await node(state)
# Verify find_package was called only once
mock_find_package.assert_called_once_with(mock_context, "My App", use_fallback=False)
# Verify launch_app_with_retries was called only once
# (no retry by ValidatorNode itself since it relies on launch_app_with_retries)
mock_launch_app.assert_called_once_with(mock_context, "com.example.app")
# Verify attempts list has only 1 element
assert "last_execution_result" in result
report = result["last_execution_result"]
assert len(report["execution"]) == 1
assert report["execution"][0]["action"] == "launch_app"
assert report["execution"][0]["attempts"] == ["Error: Force close"]
@pytest.mark.asyncio
async def test_validator_pre_execution_validation_disappeared_not_shifted(
mock_mcp, mock_context, temp_screenshot
):
"""Test that a disappeared small button matching text in a giant container
is classified as TARGET_DISAPPEARED, not TARGET_SHIFTED.
"""
# Live hierarchy only contains a giant background/container
# containing the text "Sign Up"
mock_live_elements = [
{
"text": "Sign Up",
"bounds": "[0,0][1080,2400]",
"resource-id": "root_container",
}
]
def mock_call_tool(name, args):
if name == "take_screenshot":
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
if name == "get_ui_hierarchy":
return Mock(content=[Mock(text=json.dumps(mock_live_elements))])
return Mock(content=[Mock(text="Success")])
mock_mcp.call_tool.side_effect = mock_call_tool
decisions = json.dumps(
[
{
"action": "tap",
"coordinates": [200, 250],
"target_text": "Sign Up",
"target_bounds": [100, 200, 300, 300],
"target_resource_id": "btn_signup",
}
]
)
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
node = ValidatorNode(mock_context)
with (
patch("artemis.agents.validator.failure_analyzer.FailureAnalyzer.analyze") as mock_analyze,
patch("artemis.utils.image_diff.check_ui_change", return_value=True),
patch.object(
ValidatorNode,
"_validate_action_precondition_pixel",
new_callable=AsyncMock,
) as mock_pixel,
):
mock_pixel.return_value = (
False,
ValidationErrorCategory.PIXEL_TARGET_DISAPPEARED,
"Pixel check failed",
)
mock_analyze.return_value = {
"status": "cannot_fix",
"analysis": "Sign Up button disappeared.",
}
await node(state)
# Assert that the validator correctly classified this as TARGET_DISAPPEARED
# (since size mismatch prevents shift matching)
mock_analyze.assert_called_once()
args, kwargs = mock_analyze.call_args
assert kwargs.get("error_category") == ValidationErrorCategory.TARGET_DISAPPEARED
@pytest.mark.asyncio
async def test_validator_pre_execution_validation_ocr_direct_to_pixel(
mock_mcp, mock_context, temp_screenshot
):
"""Test that OCR-derived elements bypass XML validation and
directly use pixel-based validation.
"""
mock_mcp.call_tool.side_effect = [
Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")]), # live screenshot for pixel check
Mock(content=[Mock(text="Success")]), # tap action
Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")]), # post-screenshot
]
decisions = json.dumps(
[
{
"action": "tap",
"coordinates": [200, 250],
"target_text": "Login",
"target_bounds": [100, 200, 300, 300],
"target_class": "android.view.View [OCR]",
}
]
)
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
node = ValidatorNode(mock_context)
with (
patch.object(
ValidatorNode,
"_validate_action_precondition",
new_callable=AsyncMock,
) as mock_xml,
patch.object(
ValidatorNode,
"_validate_action_precondition_pixel",
new_callable=AsyncMock,
) as mock_pixel,
patch("artemis.utils.image_diff.check_ui_change", return_value=True),
):
mock_pixel.return_value = (True, ValidationErrorCategory.NONE, "")
result = await node(state)
# Assert XML validation was NOT called
mock_xml.assert_not_called()
# Assert Pixel validation WAS called
mock_pixel.assert_called_once()
assert "last_execution_result" in result
report = result["last_execution_result"]
assert report["status"] == "success"
@pytest.mark.asyncio
async def test_validator_pre_execution_xml_failure_not_overridden_when_pixel_bypassed(
mock_mcp, mock_context, temp_screenshot
):
"""Test that when XML validation fails AND Pixel validation is bypassed
(PIXEL_BYPASSED), the XML failure is NOT overridden.
"""
mock_live_elements = [
{
"text": "Sign Up",
"bounds": "[100,200][300,300]",
"resource-id": "btn_signup",
}
]
def mock_call_tool(name, args):
if name == "take_screenshot":
return Mock(content=[Mock(text="ZHVtbXlfZGF0YQ==")])
if name == "get_ui_hierarchy":
return Mock(content=[Mock(text=json.dumps(mock_live_elements))])
return Mock(content=[Mock(text="Success")])
mock_mcp.call_tool.side_effect = mock_call_tool
decisions = json.dumps(
[
{
"action": "tap",
"coordinates": [200, 250],
"target_text": "Login",
"target_bounds": [100, 200, 300, 300],
"target_resource_id": "btn_login",
}
]
)
state = DummyState(structured_decisions=decisions, latest_screenshot=temp_screenshot)
node = ValidatorNode(mock_context)
with (
patch("artemis.agents.validator.failure_analyzer.FailureAnalyzer.analyze") as mock_analyze,
patch("artemis.utils.image_diff.check_ui_change", return_value=True),
patch.object(
ValidatorNode,
"_validate_action_precondition_pixel",
new_callable=AsyncMock,
) as mock_pixel,
):
# Pixel check was bypassed due to VLM error or low confidence
mock_pixel.return_value = (
True,
ValidationErrorCategory.PIXEL_BYPASSED,
"",
)
mock_analyze.return_value = {
"status": "cannot_fix",
"analysis": ("XML validation failed and pixel bypass did not override."),
}
await node(state)
# FailureAnalyzer SHOULD be called because XML validation failed
# and PIXEL_BYPASSED did not override it
mock_analyze.assert_called_once()
args, kwargs = mock_analyze.call_args
assert "Pre-execution validation failed" in args[2]
+456
View File
@@ -0,0 +1,456 @@
# 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.
import os
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
from artemis.agents.video_analyzer.video_analyzer import VideoAnalyzer
from artemis.context import ArtemisContext
import pytest
@pytest.mark.asyncio
async def test_video_analyzer_run():
# Mock context
mock_ctx = MagicMock(spec=ArtemisContext)
mock_ctx.data_engine = None
mock_ctx.llm_config = MagicMock()
mock_ctx.llm_config.utils.video_analyzer = MagicMock()
mock_ctx.llm_config.utils.video_analyzer.model = "gemini-3.6-flash"
# Mock Gemini Client
mock_client = MagicMock()
mock_client.aio = MagicMock()
mock_client.aio.models = MagicMock()
mock_chunk = MagicMock()
mock_chunk.text = "Final analysis: The action succeeded."
mock_chunk.function_calls = []
async def mock_stream():
yield mock_chunk
mock_client.aio.models.generate_content_stream = AsyncMock(return_value=mock_stream())
# Mock file reading for prompts
mock_prompt = "# Dynamic Video Analyzer..."
with (
patch(
"artemis.agents.video_analyzer.video_analyzer.genai.Client",
return_value=mock_client,
),
patch(
"pathlib.Path.exists",
autospec=True,
side_effect=lambda self: not str(self).endswith(".jpg"),
),
patch("pathlib.Path.read_text", return_value=mock_prompt),
patch(
"artemis.agents.video_analyzer.video_analyzer.cleanup_abandoned_gemini_files",
AsyncMock(),
),
):
analyzer = VideoAnalyzer(mock_ctx)
result, status = await analyzer.run(
time_description="from 10s to 15s", purpose="Verify action"
)
assert result == "Final analysis: The action succeeded."
assert status == "success"
assert mock_client.aio.models.generate_content_stream.called
@pytest.mark.asyncio
async def test_video_analyzer_preserves_thought_signature():
# Mock context
mock_ctx = MagicMock(spec=ArtemisContext)
mock_ctx.data_engine = None
mock_ctx.llm_config = MagicMock()
mock_ctx.llm_config.utils.video_analyzer = MagicMock()
mock_ctx.llm_config.utils.video_analyzer.model = "gemini-3.6-flash"
# Mock MobileDeviceController
mock_controller = MagicMock()
mock_controller.extract_segment_metadata = AsyncMock(
return_value=MagicMock(
success=True,
video_path="/tmp/video.mp4",
actual_start_relative_time=10.0,
)
)
from google.genai import types
dummy_signature = b"dummy_thought_signature_bytes_56789"
# Turn 1: Model calls spawn_sub_agent
spawn_call = types.FunctionCall(
name="spawn_sub_agent",
args={
"start_time": 10.0,
"end_time": 15.0,
"specific_query": "Verify login screen",
},
)
spawn_part = types.Part(
function_call=spawn_call,
thought_signature=dummy_signature,
)
mock_chunk_1 = MagicMock()
mock_chunk_1.text = ""
mock_chunk_1.function_calls = [spawn_call]
mock_chunk_1.candidates = [
types.Candidate(content=types.Content(role="model", parts=[spawn_part]))
]
# Turn 2: Model returns final answer
mock_chunk_2 = MagicMock()
mock_chunk_2.text = "Final analysis: The spawn succeeded and login button is visible."
mock_chunk_2.function_calls = []
mock_chunk_2.candidates = [
types.Candidate(
content=types.Content(
role="model",
parts=[types.Part.from_text(text=mock_chunk_2.text)],
)
)
]
# We mock generate_content_stream
call_count = 0
captured_contents = []
async def mock_stream_1():
yield mock_chunk_1
async def mock_stream_2():
yield mock_chunk_2
async def mock_generate_content_stream(model, contents, config):
nonlocal call_count
has_submit_answer = False
if config.tools:
for t in config.tools:
fd_list = getattr(t, "function_declarations", None)
if fd_list:
for fd in fd_list:
if fd.name == "submit_answer":
has_submit_answer = True
break
if not has_submit_answer:
captured_contents.append(list(contents)) # copy history contents passed to call
call_count += 1
if call_count == 1:
return mock_stream_1()
else:
return mock_stream_2()
else:
submit_call = types.FunctionCall(
name="submit_answer",
args={
"confidence_score": 0.9,
"summary": "Sub-agent verification finished",
"analysis": "Analysis here",
"verification_timestamp_secs": 10.0,
},
)
mock_sub_chunk = MagicMock()
mock_sub_chunk.text = ""
mock_sub_chunk.function_calls = [submit_call]
async def sub_stream():
yield mock_sub_chunk
return sub_stream()
# Mock Gemini Client
mock_client = MagicMock()
mock_client.aio = MagicMock()
mock_client.aio.files = MagicMock()
mock_file = MagicMock()
mock_file.name = "dummy_cloud_file_123"
mock_file.uri = "https://gemini-file-api/dummy_cloud_file_123"
mock_file.mime_type = "video/mp4"
mock_client.aio.files.upload = AsyncMock(return_value=mock_file)
mock_f_state = MagicMock()
mock_f_state.state.name = "ACTIVE"
mock_client.aio.files.get = AsyncMock(return_value=mock_f_state)
mock_client.aio.files.delete = AsyncMock()
mock_client.aio.models = MagicMock()
mock_client.aio.models.generate_content_stream = AsyncMock(
side_effect=mock_generate_content_stream
)
# Mock file reading for prompts
mock_prompt = "# Dynamic Video Analyzer..."
with (
patch(
"artemis.agents.video_analyzer.video_analyzer.genai.Client",
return_value=mock_client,
),
patch(
"artemis.agents.video_analyzer.video_analyzer.get_controller",
return_value=mock_controller,
),
patch(
"pathlib.Path.exists",
autospec=True,
side_effect=lambda self: not str(self).endswith(".jpg"),
),
patch(
"pathlib.Path.stat",
return_value=os.stat_result((0o100644, 0, 0, 0, 0, 0, 1024 * 1024, 0, 0, 0)),
),
patch("pathlib.Path.read_text", return_value=mock_prompt),
patch(
"artemis.agents.video_analyzer.video_analyzer.cleanup_abandoned_gemini_files",
AsyncMock(),
),
patch(
"artemis.agents.video_analyzer.video_analyzer.compress_video_for_api",
AsyncMock(return_value=Path("/tmp/video.mp4")),
),
):
analyzer = VideoAnalyzer(mock_ctx)
result, status = await analyzer.run(
time_description="from 10s to 15s", purpose="Verify login"
)
assert status == "success"
assert "spawn succeeded" in result
assert call_count == 2
# Verify that the second model call's history contains the model turn with the thought_signature
second_call_history = captured_contents[1]
assert len(second_call_history) >= 4
model_turn = second_call_history[1]
assert model_turn.role == "model"
assert len(model_turn.parts) == 1
assert model_turn.parts[0].function_call.name == "spawn_sub_agent"
assert model_turn.parts[0].thought_signature == dummy_signature
@pytest.mark.asyncio
async def test_video_analyzer_sub_agent_confidence_validation():
# Mock context
mock_ctx = MagicMock(spec=ArtemisContext)
mock_ctx.data_engine = None
mock_ctx.llm_config = MagicMock()
mock_ctx.llm_config.utils.video_analyzer = MagicMock()
mock_ctx.llm_config.utils.video_analyzer.model = "gemini-3.6-flash"
# Mock MobileDeviceController
mock_controller = MagicMock()
mock_controller.extract_segment_metadata = AsyncMock(
return_value=MagicMock(
success=True,
video_path="/tmp/video.mp4",
actual_start_relative_time=10.0,
)
)
from google.genai import types
# Main agent responses
# Turn 1: Calls spawn_sub_agent
spawn_call = types.FunctionCall(
name="spawn_sub_agent",
args={
"start_time": 10.0,
"end_time": 15.0,
"specific_query": "Verify login screen",
},
)
mock_main_chunk_1 = MagicMock()
mock_main_chunk_1.text = ""
mock_main_chunk_1.function_calls = [spawn_call]
# Turn 2: Returns final answer
mock_main_chunk_2 = MagicMock()
mock_main_chunk_2.text = "Final blackboard status checked."
mock_main_chunk_2.function_calls = []
# Sub-agent responses
# Turn 1: Calls submit_answer with invalid confidence_score (string)
invalid_submit_call = types.FunctionCall(
name="submit_answer",
args={
"timeline_events": [
{
"confidence_score": "HIGH",
"transcription": "Invalid confidence score call",
"verification_timestamp_secs": 10.0,
}
],
"summary": "Invalid confidence score call",
"analysis": "Some analysis",
},
)
sub_dummy_signature = b"sub_dummy_signature_9999"
sub_part_1 = types.Part(
function_call=invalid_submit_call,
thought_signature=sub_dummy_signature,
)
mock_sub_chunk_1 = MagicMock()
mock_sub_chunk_1.text = ""
mock_sub_chunk_1.function_calls = [invalid_submit_call]
mock_sub_chunk_1.candidates = [
types.Candidate(content=types.Content(role="model", parts=[sub_part_1]))
]
# Turn 2: Calls submit_answer with valid confidence_score
valid_submit_call = types.FunctionCall(
name="submit_answer",
args={
"timeline_events": [
{
"confidence_score": 0.8,
"transcription": "Valid confidence score call",
"verification_timestamp_secs": 10.0,
}
],
"summary": "Valid confidence score call",
"analysis": "Some analysis",
},
)
mock_sub_chunk_2 = MagicMock()
mock_sub_chunk_2.text = ""
mock_sub_chunk_2.function_calls = [valid_submit_call]
mock_sub_chunk_2.candidates = [
types.Candidate(
content=types.Content(role="model", parts=[types.Part(function_call=valid_submit_call)])
)
]
# Stream generator mocks
main_calls = 0
sub_calls = 0
captured_sub_contents = []
async def mock_main_stream():
nonlocal main_calls
if main_calls == 0:
main_calls += 1
yield mock_main_chunk_1
else:
yield mock_main_chunk_2
async def mock_sub_stream():
nonlocal sub_calls
if sub_calls == 0:
sub_calls += 1
yield mock_sub_chunk_1
else:
yield mock_sub_chunk_2
async def mock_generate_content_stream(model, contents, config):
system_instruction = config.system_instruction
if "Dynamic Video Analyzer" in str(system_instruction):
return mock_main_stream()
else:
captured_sub_contents.append(list(contents))
return mock_sub_stream()
# Mock Gemini Client
mock_client = MagicMock()
mock_client.aio = MagicMock()
mock_client.aio.files = MagicMock()
mock_file = MagicMock()
mock_file.name = "dummy_cloud_file_123"
mock_file.uri = "https://gemini-file-api/dummy_cloud_file_123"
mock_file.mime_type = "video/mp4"
mock_client.aio.files.upload = AsyncMock(return_value=mock_file)
mock_f_state = MagicMock()
mock_f_state.state.name = "ACTIVE"
mock_client.aio.files.get = AsyncMock(return_value=mock_f_state)
mock_client.aio.files.delete = AsyncMock()
mock_client.aio.models = MagicMock()
mock_client.aio.models.generate_content_stream = AsyncMock(
side_effect=mock_generate_content_stream
)
# Mock file reading for prompts
mock_main_prompt = "# Dynamic Video Analyzer..."
mock_sub_prompt = "# Sub Agent..."
def mock_read_text(self, encoding="utf-8"):
if "video_analyzer.md" in str(self):
return mock_main_prompt
elif "video_sub_agent.md" in str(self):
return mock_sub_prompt
return ""
with (
patch(
"artemis.agents.video_analyzer.video_analyzer.genai.Client",
return_value=mock_client,
),
patch(
"artemis.agents.video_analyzer.video_analyzer.get_controller",
return_value=mock_controller,
),
patch(
"pathlib.Path.exists",
autospec=True,
side_effect=lambda self: not str(self).endswith(".jpg"),
),
patch(
"pathlib.Path.stat",
return_value=os.stat_result((0o100644, 0, 0, 0, 0, 0, 1024 * 1024, 0, 0, 0)),
),
patch("pathlib.Path.read_text", mock_read_text),
patch("pathlib.Path.mkdir"),
patch("pathlib.Path.touch"),
patch("pathlib.Path.is_dir", return_value=True),
patch(
"artemis.agents.video_analyzer.video_analyzer.cleanup_abandoned_gemini_files",
AsyncMock(),
),
patch(
"artemis.agents.video_analyzer.video_analyzer.compress_video_for_api",
AsyncMock(return_value=Path("/tmp/video.mp4")),
),
patch(
"asyncio.create_subprocess_exec",
AsyncMock(return_value=MagicMock(communicate=AsyncMock(return_value=(b"", b"")))),
),
):
analyzer = VideoAnalyzer(mock_ctx)
result, status = await analyzer.run(
time_description="from 10s to 15s", purpose="Verify login"
)
assert status == "success"
assert sub_calls == 1
assert len(analyzer.blackboard_entries) == 1
entry = analyzer.blackboard_entries[0]
assert entry["confidence_score"] == 0.8
assert entry["summary"] == "Valid confidence score call"
# Verify that sub-agent turn 2 received the model turn with the thought_signature
assert len(captured_sub_contents) >= 2
sub_turn_2_contents = captured_sub_contents[1]
model_turns = [c for c in sub_turn_2_contents if c.role == "model"]
assert len(model_turns) == 1
assert model_turns[0].parts[0].thought_signature == sub_dummy_signature
@@ -0,0 +1,234 @@
# 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 VideoAnalyzer dual-path execution engine."""
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
from langchain_core.messages import AIMessage
from artemis.agents.video_analyzer.video_analyzer import (
UNIVERSAL_MAIN_TOOLS,
UNIVERSAL_SUBMIT_ANSWER_TOOL,
VideoAnalyzer,
)
from artemis.context import ArtemisContext
@pytest.fixture
def mock_context(tmp_path):
ctx = Mock(spec=ArtemisContext)
ctx.llm_config = Mock()
mock_llm_cfg = Mock()
mock_llm_cfg.model = "claude-3-7-sonnet"
mock_llm_cfg.temperature = 0.2
ctx.llm_config.utils = Mock()
ctx.llm_config.utils.video_analyzer = mock_llm_cfg
ctx.device = Mock()
ctx.device.device_width = 1080
ctx.device.device_height = 2400
ctx.data_engine = Mock()
ctx.data_engine.base_dir = tmp_path
return ctx
def test_video_analyzer_universal_engine_detection(mock_context):
"""Test that VideoAnalyzer routes to Universal Engine when non-Gemini model is set."""
from artemis.config import settings
with patch.object(settings, "GOOGLE_API_KEY", None):
agent = VideoAnalyzer(mock_context)
assert agent.use_native_gemini is False
assert agent.model_name == "claude-3-7-sonnet"
def test_universal_tool_schemas():
"""Verify schema structures for Universal tools."""
assert UNIVERSAL_SUBMIT_ANSWER_TOOL["type"] == "function"
assert UNIVERSAL_SUBMIT_ANSWER_TOOL["function"]["name"] == "submit_answer"
assert len(UNIVERSAL_MAIN_TOOLS) == 3
tool_names = [t["function"]["name"] for t in UNIVERSAL_MAIN_TOOLS]
assert "extract_segment_metadata" in tool_names
assert "spawn_sub_agent" in tool_names
assert "analyze_audio_only" in tool_names
@pytest.mark.asyncio
async def test_exec_single_chunk_universal(mock_context, tmp_path):
"""Test keyframe extraction and LangChain tool invocation in universal sub-agent."""
agent = VideoAnalyzer(mock_context)
agent.use_native_gemini = False
fake_video = tmp_path / "compressed.mp4"
fake_video.write_bytes(b"video_bytes")
raw_video = tmp_path / "raw.mp4"
raw_video.write_bytes(b"raw_bytes")
fake_keyframes = [
(0.0, b"frame0"),
(1.0, b"frame1"),
(2.0, b"frame2"),
]
mock_llm = MagicMock()
mock_response = AIMessage(
content="Sub-agent analyzed frames.",
tool_calls=[
{
"name": "submit_answer",
"args": {
"summary": "Detected settings button tap.",
"analysis": "At 1.0s user clicked Settings icon.",
"timeline_events": [
{
"start_time": 0.5,
"end_time": 1.5,
"transcription": "Tap on Settings",
"confidence_score": 0.95,
"verification_timestamp_secs": 1.0,
}
],
},
"id": "sub_call_1",
}
],
)
mock_bound = MagicMock()
mock_bound.ainvoke = AsyncMock(return_value=mock_response)
mock_llm.bind_tools.return_value = mock_bound
with (
patch(
"artemis.agents.video_analyzer.video_analyzer.extract_keyframes_from_video",
return_value=fake_keyframes,
),
patch(
"artemis.agents.video_analyzer.video_analyzer.get_llm",
return_value=mock_llm,
),
):
result = await agent._exec_single_chunk_universal(
compressed_path=fake_video,
raw_path=raw_video,
start_time=0.0,
end_time=3.0,
actual_start=0.0,
prompt_with_context="Find when settings was opened.",
specific_query="settings icon",
)
assert "Detected settings button tap" in result
assert "At 1.0s user clicked Settings icon" in result
assert len(agent.blackboard_entries) == 1
assert agent.blackboard_entries[0]["summary"] == "Tap on Settings"
assert agent.blackboard_entries[0]["confidence_score"] == 0.95
@pytest.mark.asyncio
async def test_exec_analyze_audio_universal(mock_context, tmp_path):
"""Test audio analysis in universal mode."""
agent = VideoAnalyzer(mock_context)
agent.use_native_gemini = False
fake_audio = tmp_path / "audio.aac"
fake_audio.write_bytes(b"audio_bytes")
mock_llm = MagicMock()
mock_response = AIMessage(
content="",
tool_calls=[
{
"name": "submit_answer",
"args": {
"summary": "Heard notification sound.",
"analysis": "Notification bell sounded at 2.0s.",
},
"id": "audio_call_1",
}
],
)
mock_bound = MagicMock()
mock_bound.ainvoke = AsyncMock(return_value=mock_response)
mock_llm.bind_tools.return_value = mock_bound
with patch(
"artemis.agents.video_analyzer.video_analyzer.get_llm",
return_value=mock_llm,
):
result = await agent._exec_analyze_audio_universal(
audio_path=fake_audio,
start_time=1.0,
end_time=4.0,
actual_start=1.0,
prompt_with_context="Listen for notification sound.",
specific_query="notification bell",
)
assert "Heard notification sound" in result
assert len(agent.blackboard_entries) == 1
assert agent.blackboard_entries[0]["summary"] == "Heard notification sound."
@pytest.mark.asyncio
async def test_run_universal_multi_turn_reasoning(mock_context):
"""Test main reasoning loop of VideoAnalyzer in universal mode with tool dispatch."""
agent = VideoAnalyzer(mock_context)
agent.use_native_gemini = False
# Turn 1: Call extract_segment_metadata
msg_turn1 = AIMessage(
content="Let me check the video metadata first.",
tool_calls=[
{
"name": "extract_segment_metadata",
"args": {"start_time": 0.0, "end_time": 10.0},
"id": "main_call_1",
}
],
)
# Turn 2: Final response
msg_turn2 = AIMessage(
content="The video shows the app launched successfully at 3.5s.",
tool_calls=[],
)
mock_llm = MagicMock()
mock_bound = MagicMock()
mock_bound.ainvoke = AsyncMock(side_effect=[msg_turn1, msg_turn2])
mock_llm.bind_tools.return_value = mock_bound
with (
patch(
"artemis.agents.video_analyzer.video_analyzer.get_llm",
return_value=mock_llm,
),
patch.object(
agent,
"exec_extract_segment_metadata",
new_callable=AsyncMock,
return_value='{"duration_seconds": 10.0, "file_size_mb": 2.5}',
),
):
outcome, status = await agent._run_universal(
time_description="from 0s to 10s",
purpose="Verify app launch",
system_prompt="You are a video analyzer.",
)
assert status == "success"
assert "launched successfully at 3.5s" in outcome
assert agent.exec_extract_segment_metadata.call_count == 1