428 lines
17 KiB
Python
428 lines
17 KiB
Python
import asyncio
|
|
import json
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
|
|
from apps.admin_console.services.automation_store import AutomationStore
|
|
from artemis.config.exploration import ExplorationConfig
|
|
from artemis.exploration.demonstrations import Demonstrations
|
|
from artemis.exploration.store import ExplorationConflict, ExplorationStore
|
|
from artemis.runtime import task_control
|
|
|
|
|
|
@pytest.fixture
|
|
def demo_session(tmp_path, monkeypatch):
|
|
repository = ExplorationStore(AutomationStore(tmp_path / "history.db"))
|
|
monkeypatch.setattr("artemis.exploration.runtime.explorations", repository)
|
|
monkeypatch.setattr("artemis.exploration.demo_runtime.explorations", repository)
|
|
monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path / "traces")
|
|
session = repository.create("验证操作结果", ExplorationConfig(), {}, "create")
|
|
run = repository.reserve_run(
|
|
session["id"], "start", kind="exploration", goal="验证", task_config={}
|
|
)
|
|
repository.set_run_state(run["id"], "running")
|
|
return repository, session, run
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"window_class", ["Chrome_WidgetWin_1", "Progman", "WorkerW", "Shell_TrayWnd"]
|
|
)
|
|
def test_window_demo_bounds_use_foreground_hwnd_not_virtual_desktop(monkeypatch, window_class):
|
|
from artemis.exploration import demo_runtime
|
|
|
|
user32 = Mock()
|
|
user32.GetForegroundWindow.return_value = 2**40 + 123
|
|
user32.IsIconic.return_value = False
|
|
user32.SetThreadDpiAwarenessContext.return_value = 17
|
|
|
|
def class_name(hwnd, buffer, size):
|
|
buffer.value = window_class
|
|
return len(window_class)
|
|
|
|
def bounds(hwnd, pointer):
|
|
rect = pointer._obj
|
|
rect.left, rect.top, rect.right, rect.bottom = -1200, 50, -400, 650
|
|
return True
|
|
|
|
def title(hwnd, buffer, size):
|
|
buffer.value = "示范浏览器"
|
|
return len(buffer.value)
|
|
|
|
user32.GetClassNameW.side_effect = class_name
|
|
user32.GetWindowRect.side_effect = bounds
|
|
user32.GetWindowTextW.side_effect = title
|
|
monkeypatch.setattr(
|
|
demo_runtime.ctypes, "windll", SimpleNamespace(user32=user32), raising=False
|
|
)
|
|
monkeypatch.setattr(demo_runtime, "os", SimpleNamespace(name="nt"))
|
|
result = demo_runtime.desktop_source("window")
|
|
user32.GetSystemMetrics.assert_not_called()
|
|
assert user32.SetThreadDpiAwarenessContext.call_count == 2
|
|
if window_class != "Chrome_WidgetWin_1":
|
|
assert result is None
|
|
return
|
|
assert result.target_id == str(2**40 + 123)
|
|
assert result.label == "示范浏览器"
|
|
assert result.window == {"left": -1200, "top": 50, "width": 800, "height": 600}
|
|
assert result.screen is None
|
|
assert (result.width, result.height) == (800, 600)
|
|
|
|
|
|
def test_demo_command_retries_do_not_extend_twice_or_finish_other_sessions(
|
|
demo_session, monkeypatch
|
|
):
|
|
repository, session, run = demo_session
|
|
now = [100.0]
|
|
monkeypatch.setattr("artemis.exploration.demonstrations.time.time", lambda: now[0])
|
|
demos = Demonstrations(repository)
|
|
demo = demos.request(session["id"], "both", "request")
|
|
assert demos.request(session["id"], "both", "request")["id"] == demo["id"]
|
|
with pytest.raises(ExplorationConflict):
|
|
demos.request(session["id"], "android", "second")
|
|
demos.transition(demo["id"], "recording")
|
|
for _ in range(2):
|
|
demos.command(session["id"], demo["id"], "extend", "extend", seconds=60)
|
|
assert demos.get(demo["id"])["deadline"] == 1060
|
|
other = repository.create("另一个目标", ExplorationConfig(), {}, "other")
|
|
with pytest.raises(KeyError):
|
|
demos.command(other["id"], demo["id"], "finish", "finish")
|
|
now[0] = 1061
|
|
with pytest.raises(ExplorationConflict):
|
|
demos.command(session["id"], demo["id"], "finish", "finish")
|
|
demos.transition(demo["id"], "timed_out", media=[{"saved": True}])
|
|
assert repository.get(session["id"])["active_demonstration_id"] is None
|
|
assert repository.run(run["id"])["outcome"] is None
|
|
assert demos.transition(demo["id"], "completed")["status"] == "timed_out"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_demo_safe_point_keeps_lease_and_restores_capture_after_success(
|
|
demo_session, monkeypatch
|
|
):
|
|
from artemis.exploration.runtime import ExplorationControl
|
|
|
|
repository, session, run = demo_session
|
|
demo = Demonstrations(repository).request(session["id"], "android", "demo")
|
|
lock = SimpleNamespace(device_id="test-device", release=Mock())
|
|
control = ExplorationControl(run["id"], lock)
|
|
control.context = SimpleNamespace()
|
|
stop, resume = AsyncMock(), AsyncMock()
|
|
|
|
async def perform(ctx, observed_control, record):
|
|
assert record["id"] == demo["id"]
|
|
stop.assert_awaited_once()
|
|
lock.release.assert_not_called()
|
|
observed_control._state("demonstrating")
|
|
assert task_control.read_state(run["id"])["status"] == "demonstrating"
|
|
return ""
|
|
|
|
monkeypatch.setattr("artemis.exploration.demo_runtime.perform_demonstration", perform)
|
|
task_control.request_pause(run["id"], reason="demonstration")
|
|
assert await control.checkpoint(on_pause=stop, on_resume=resume)
|
|
lock.release.assert_not_called()
|
|
resume.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_demo_timeout_saves_media_then_releases_lease_without_failing(
|
|
demo_session, monkeypatch
|
|
):
|
|
from artemis.exploration.runtime import ExplorationControl
|
|
|
|
repository, session, run = demo_session
|
|
demo = Demonstrations(repository).request(session["id"], "android", "demo")
|
|
lock = SimpleNamespace(device_id="test-device", release=Mock())
|
|
control = ExplorationControl(run["id"], lock)
|
|
control.context = SimpleNamespace()
|
|
|
|
async def perform(*_):
|
|
Demonstrations(repository).transition(
|
|
demo["id"], "timed_out", media=[{"file": "saved.mp4"}]
|
|
)
|
|
return "demonstration_timed_out"
|
|
|
|
monkeypatch.setattr("artemis.exploration.demo_runtime.perform_demonstration", perform)
|
|
task_control.request_pause(run["id"], reason="demonstration")
|
|
waiting = asyncio.create_task(control.checkpoint())
|
|
for _ in range(100):
|
|
await asyncio.sleep(0.01)
|
|
if waiting.done():
|
|
waiting.result()
|
|
if repository.get(session["id"])["status"] == "paused":
|
|
break
|
|
try:
|
|
assert lock.release.called
|
|
assert repository.get(session["id"])["status"] == "paused"
|
|
assert repository.run(run["id"])["outcome"] is None
|
|
assert Demonstrations(repository).get(demo["id"])["media"]
|
|
finally:
|
|
waiting.cancel()
|
|
await asyncio.gather(waiting, return_exceptions=True)
|
|
|
|
|
|
def test_system_demo_summary_cannot_answer_pending_user_question(demo_session):
|
|
from artemis.exploration.runtime import ExplorationControl
|
|
|
|
repository, session, run = demo_session
|
|
control = ExplorationControl(run["id"], Mock())
|
|
question = repository.ask(run["id"], "预期是哪个状态？", "开启", "q")
|
|
task_control.add_instruction(run["id"], "视频看到了点击", source="demonstration")
|
|
assert not control.question_answered()
|
|
repository.answer_question(session["id"], question["id"])
|
|
assert control.question_answered()
|
|
|
|
|
|
def test_legacy_demo_source_survives_recent_message_window_without_copying_media(demo_session):
|
|
from artemis.exploration.runtime import conversation_message
|
|
|
|
repository, session, run = demo_session
|
|
demo = {
|
|
"id": "old-demo",
|
|
"run_id": run["id"],
|
|
"exploration_id": session["id"],
|
|
"source": "computer",
|
|
"status": "completed",
|
|
"media": [{"kind": "computer", "segments": [{"file": "private/path/recording.mp4"}]}],
|
|
}
|
|
repository.database.put("exploration_demo", demo["id"], demo)
|
|
repository.database.put(
|
|
"exploration_demo", "failed-demo", {**demo, "id": "failed-demo", "status": "failed"}
|
|
)
|
|
repository.add_message(
|
|
session["id"],
|
|
"打开浏览器看首页",
|
|
"legacy-summary",
|
|
role="assistant",
|
|
metadata={"demonstration_id": demo["id"], "kind": "demonstration"},
|
|
)
|
|
for i in range(4):
|
|
repository.add_message(session["id"], "最新指导" * 2000, f"guidance-{i}")
|
|
ctx = SimpleNamespace(exploration_id=session["id"], exploration_run_id=run["id"])
|
|
payload = json.loads(conversation_message(ctx).content.split("\n", 1)[1])
|
|
assert not any(m.get("demonstration_id") for m in payload["messages"])
|
|
assert payload["demonstrations"] == [
|
|
{
|
|
"id": "old-demo",
|
|
"run_id": run["id"],
|
|
"source": "computer",
|
|
"recorded_sources": ["computer"],
|
|
}
|
|
]
|
|
assert "private/path" not in str(payload)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("needs_user", [True, False])
|
|
@pytest.mark.parametrize(
|
|
("source", "recorded_sources"),
|
|
[("computer", ["computer"]), ("android", ["android"]), ("both", ["android", "computer"])],
|
|
)
|
|
async def test_demo_source_survives_clarification_and_execution_restart(
|
|
demo_session, monkeypatch, needs_user, source, recorded_sources
|
|
):
|
|
from apps.admin_console.services.exploration_service import ExplorationService
|
|
from artemis.exploration import demo_runtime
|
|
from artemis.exploration.runtime import conversation_message
|
|
|
|
repository, session, run = demo_session
|
|
demos = Demonstrations(repository)
|
|
demo = demos.request(session["id"], source, "demo")
|
|
demos.transition(demo["id"], "recording")
|
|
demos.command(session["id"], demo["id"], "finish", "finish")
|
|
capture = SimpleNamespace(
|
|
start=AsyncMock(),
|
|
stop=AsyncMock(),
|
|
errors=[],
|
|
media=[{"id": f"track-{kind}", "kind": kind} for kind in recorded_sources],
|
|
)
|
|
monkeypatch.setattr(demo_runtime, "DemonstrationCapture", lambda *_: capture)
|
|
monkeypatch.setattr(
|
|
demo_runtime,
|
|
"analyze_demonstration",
|
|
AsyncMock(
|
|
return_value=demo_runtime.DemonstrationAnalysis(
|
|
summary="打开浏览器，浏览首页后返回顶部。",
|
|
needs_user=needs_user,
|
|
question="目标网址是什么？" if needs_user else "",
|
|
)
|
|
),
|
|
)
|
|
|
|
async def run_interruptibly(operation, **_):
|
|
return False, await operation
|
|
|
|
control = SimpleNamespace(
|
|
session_id=run["id"],
|
|
directory=task_control.control_dir(run["id"]),
|
|
_state=Mock(),
|
|
run_interruptibly=run_interruptibly,
|
|
)
|
|
ctx = SimpleNamespace(exploration_id=session["id"], exploration_run_id=run["id"])
|
|
reason = await demo_runtime.perform_demonstration(ctx, control, demo)
|
|
assert reason == ("question" if needs_user else "")
|
|
pending = repository.get(session["id"])["pending_question"]
|
|
assert bool(pending) == needs_user
|
|
ExplorationService(repository).message(session["id"], "www.playones.com，首页看一遍就行", "url")
|
|
assert repository.get(session["id"])["pending_question"] is None
|
|
|
|
# Reopen the durable store: source must not depend on in-memory capture,
|
|
# the model repeating a platform name, or whether analysis needed an answer.
|
|
restored = ExplorationStore(repository.database)
|
|
monkeypatch.setattr("artemis.exploration.runtime.explorations", restored)
|
|
payload = json.loads(conversation_message(ctx).content.split("\n", 1)[1])
|
|
assert payload["demonstrations"] == [
|
|
{
|
|
"id": demo["id"],
|
|
"run_id": run["id"],
|
|
"source": source,
|
|
"recorded_sources": recorded_sources,
|
|
}
|
|
]
|
|
summary = next(m for m in payload["messages"] if m.get("demonstration_id") == demo["id"])
|
|
assert "打开浏览器" in summary["text"]
|
|
assert ("电脑" if source == "computer" else "安卓") in summary["text"]
|
|
instructions = task_control.consume_instructions(control.directory)["instruction"]
|
|
assert "打开浏览器" in instructions
|
|
assert "www.playones.com" in instructions
|
|
assert ("电脑" if source == "computer" else "安卓") in instructions
|
|
|
|
# Explicit user changes remain later guidance; provenance is evidence,
|
|
# never a hard global execution-target switch that would forbid mixed tests.
|
|
ExplorationService(restored).message(session["id"], "接下来切到手机核对页面", "switch")
|
|
changed = json.loads(conversation_message(ctx).content.split("\n", 1)[1])
|
|
assert changed["messages"][-1]["text"] == "接下来切到手机核对页面"
|
|
assert changed["demonstrations"] == payload["demonstrations"]
|
|
assert restored.run(run["id"])["task_config"] == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_saved_video_extracts_copy_and_preserves_source_and_offsets(tmp_path, monkeypatch):
|
|
from artemis.exploration.media import SavedVideoSource
|
|
|
|
original = tmp_path / "demo.mp4"
|
|
original.write_bytes(b"evidence")
|
|
|
|
async def render(segments, start, end, target, **kwargs):
|
|
assert (start, end) == (12, 14)
|
|
target.write_bytes(b"temporary")
|
|
return True
|
|
|
|
monkeypatch.setattr("artemis.exploration.media.render_timeline_clip", render)
|
|
source = SavedVideoSource(
|
|
tmp_path,
|
|
{
|
|
"id": "demo-computer",
|
|
"kind": "computer",
|
|
"segments": [{"file": "demo.mp4", "offset_ms": 12000, "duration_ms": 2000}],
|
|
},
|
|
)
|
|
result = await source.extract_segment_metadata(0, None)
|
|
assert result.actual_start_relative_time == 12
|
|
assert result.source_revision == "demo-computer"
|
|
result.video_path.unlink()
|
|
result.video_path.parent.rmdir()
|
|
assert original.read_bytes() == b"evidence"
|
|
with pytest.raises(ValueError):
|
|
SavedVideoSource(
|
|
tmp_path, {"segments": [{"file": "../private.mp4", "offset_ms": 0, "duration_ms": 1}]}
|
|
)
|
|
|
|
|
|
def test_saved_demo_sampling_reaches_end_of_long_interval(monkeypatch, tmp_path):
|
|
from artemis.agents.video_analyzer.universal_engine import _gather_universal_frames
|
|
|
|
sampling = Mock(return_value=[])
|
|
monkeypatch.setattr(
|
|
"artemis.agents.video_analyzer.video_analyzer.extract_frames_at_timestamps", sampling
|
|
)
|
|
analyzer = SimpleNamespace(
|
|
media_source=object(), _dense_action_offsets=lambda *args: [], max_dense_action_frames=24
|
|
)
|
|
_gather_universal_frames(analyzer, tmp_path / "video.mp4", 32.249, 92.249, 32.249)
|
|
offsets = sampling.call_args_list[0].args[1]
|
|
assert len(offsets) <= 30
|
|
assert offsets[0] == 0
|
|
assert offsets[-1] == pytest.approx(59.9)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_saved_demo_chunk_keeps_millisecond_coverage(monkeypatch, tmp_path):
|
|
from artemis.agents.video_analyzer.blackboard import VideoBlackboard
|
|
from artemis.agents.video_analyzer import chunk_runner
|
|
|
|
board = VideoBlackboard(
|
|
"demo", db_path=tmp_path / "board.db", evidence_dir=tmp_path / "evidence"
|
|
)
|
|
analyzer = SimpleNamespace(
|
|
max_retries=0,
|
|
ctx=SimpleNamespace(device=None),
|
|
blackboard=board,
|
|
model_name="test",
|
|
video_processing="static",
|
|
chunk_size_seconds=60,
|
|
_exec_single_chunk_universal=AsyncMock(return_value="actual summary"),
|
|
_parse_chunk_result=lambda _: ("已分析完整区间", "没有推断缺失操作"),
|
|
)
|
|
monkeypatch.setattr(chunk_runner, "_prepare_chunk_media", AsyncMock(return_value=(None, 1)))
|
|
monkeypatch.setattr(chunk_runner, "_build_chunk_prompt", lambda *args: "query")
|
|
await chunk_runner.exec_single_chunk(analyzer, 92.249, 137.428, "query")
|
|
assert board.missing_intervals(92.249, 137.428, "query") == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failed_render_removes_partial_temporary_clip_but_not_source(tmp_path, monkeypatch):
|
|
from artemis.exploration.media import SavedVideoSource
|
|
|
|
original = tmp_path / "recording.mp4"
|
|
original.write_bytes(b"saved")
|
|
targets = []
|
|
|
|
async def render(*args, **kwargs):
|
|
target = args[3]
|
|
targets.append(target)
|
|
target.write_bytes(b"partial")
|
|
return False
|
|
|
|
monkeypatch.setattr("artemis.exploration.media.render_timeline_clip", render)
|
|
source = SavedVideoSource(
|
|
tmp_path,
|
|
{
|
|
"id": "demo",
|
|
"kind": "android",
|
|
"segments": [{"file": original.name, "offset_ms": 0, "duration_ms": 1000}],
|
|
},
|
|
)
|
|
assert not (await source.extract_segment_metadata(0, 1)).success
|
|
assert not targets[0].parent.exists()
|
|
assert original.read_bytes() == b"saved"
|
|
|
|
|
|
def test_temporary_recording_cleanup_keeps_demonstration_playback(tmp_path):
|
|
from artemis.context import ArtemisContext
|
|
from artemis.data_engine.engine import DataEngine
|
|
from unittest.mock import MagicMock
|
|
from uuid import uuid4
|
|
import time
|
|
|
|
ctx = MagicMock(spec=ArtemisContext)
|
|
ctx.execution_setup = MagicMock(traces_path=str(tmp_path / "traces"))
|
|
ctx.device = None
|
|
engine = DataEngine(ctx)
|
|
sid = engine.start_session(goal="test")
|
|
keep = tmp_path / "demonstrations"
|
|
keep.mkdir()
|
|
demo = keep / "demo.mp4"
|
|
temporary = tmp_path / "temporary.mp4"
|
|
demo.write_bytes(b"saved")
|
|
temporary.write_bytes(b"temporary")
|
|
demo_id, temp_id = uuid4(), uuid4()
|
|
for vid, path in ((demo_id, demo), (temp_id, temporary)):
|
|
engine.record_video_start(vid, "device", path)
|
|
engine.record_video_stop(vid, "device", path, time.time() - 10, time.time())
|
|
engine.storage.discard_session_recordings(str(sid), preserve_directory=keep)
|
|
assert engine.storage.get_video_recording(demo_id).local_video_path == str(demo)
|
|
assert engine.storage.get_video_recording(temp_id).local_video_path is None
|
|
assert engine.storage.get_session(sid).video_filepath == str(demo)
|