126 lines
4.8 KiB
Python
126 lines
4.8 KiB
Python
import asyncio
|
|
import json
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from artemis.agents.checker import flash
|
|
from artemis.agents.checker.checker import CheckReport
|
|
from artemis.context import ArtemisContext, ExecutionSetup
|
|
from artemis.runtime.case_result import CaseTestResult
|
|
|
|
|
|
def context(tmp_path, **setup):
|
|
ctx = MagicMock(spec=ArtemisContext)
|
|
ctx.execution_setup = ExecutionSetup(**setup)
|
|
ctx.data_engine = MagicMock()
|
|
ctx.data_engine.base_dir = tmp_path
|
|
return ctx
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("verdict", ["passed", "failed", "inconclusive"])
|
|
async def test_flash_reuses_checker_with_original_prompt_and_final_report(
|
|
tmp_path, monkeypatch, verdict
|
|
):
|
|
result = CaseTestResult(
|
|
status=verdict, reason="提示词预期与实际对比", evidence=["步骤 3 截图显示实际状态"]
|
|
)
|
|
check = AsyncMock(return_value=CheckReport(test_result=result))
|
|
monkeypatch.setattr(flash, "run_final_check", check)
|
|
ctx = context(tmp_path)
|
|
ctx.on_execution_review = AsyncMock()
|
|
|
|
async def review_after_capture_stops(*args, **kwargs):
|
|
ctx.on_execution_review.assert_awaited_once()
|
|
return CheckReport(test_result=result)
|
|
|
|
check.side_effect = review_after_capture_stops
|
|
execution = {"status": "completed", "explanation": "操作已执行"}
|
|
data = await flash.settle_flash_outcome(ctx, "打开设置，预期看到应用入口", execution)
|
|
assert check.call_count == 1
|
|
assert check.call_args.kwargs["goal"] == "打开设置，预期看到应用入口"
|
|
assert check.call_args.kwargs["execution_report"] == execution
|
|
assert check.call_args.kwargs["check_items"] == []
|
|
assert data["task_status"] == "completed"
|
|
assert data["test_result"]["status"] == verdict
|
|
assert (
|
|
json.loads((tmp_path / "run_outcome.json").read_text(encoding="utf-8"))["test_result"]
|
|
== data["test_result"]
|
|
)
|
|
assert ctx.run_outcome == data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_acceptance_timeout_finishes_with_unknown_verdict(tmp_path, monkeypatch):
|
|
async def hanging_check(*args, **kwargs):
|
|
await asyncio.Event().wait()
|
|
|
|
monkeypatch.setattr(flash, "run_final_check", hanging_check)
|
|
data = await flash.settle_flash_outcome(
|
|
context(tmp_path, checkpoint_timeout=0.01), "任务", {"status": "completed"}
|
|
)
|
|
assert data["task_status"] == "completed"
|
|
assert data["test_result"]["status"] == "inconclusive"
|
|
assert "0.01 秒" in data["test_result"]["reason"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("setup", [{"disable_checker": True}, {"disable_final_check": True}])
|
|
async def test_disabled_acceptance_does_not_call_a_model_or_claim_pass(
|
|
tmp_path, monkeypatch, setup
|
|
):
|
|
check = AsyncMock()
|
|
monkeypatch.setattr(flash, "run_final_check", check)
|
|
data = await flash.settle_flash_outcome(
|
|
context(tmp_path, **setup), "任务", {"status": "completed"}
|
|
)
|
|
check.assert_not_called()
|
|
assert data["test_result"]["status"] == "inconclusive"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execution_failure_is_blocked_without_another_model_retry(tmp_path, monkeypatch):
|
|
check = AsyncMock()
|
|
monkeypatch.setattr(flash, "run_final_check", check)
|
|
data = await flash.settle_flash_outcome(
|
|
context(tmp_path), "任务", {"status": "failed", "explanation": "设备观测失败"}
|
|
)
|
|
check.assert_not_called()
|
|
assert data["test_result"]["status"] == "blocked"
|
|
assert data["test_result"]["reason"] == "设备观测失败"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reported_expectation_mismatch_is_checked_even_when_executor_reports_failed(
|
|
tmp_path, monkeypatch
|
|
):
|
|
check = AsyncMock(
|
|
return_value=CheckReport(
|
|
test_result=CaseTestResult(
|
|
status="failed",
|
|
reason="预期英文 Search settings，实际中文 搜索设置。",
|
|
evidence=["设置首页截图和 UI 元素 1 显示 搜索设置"],
|
|
)
|
|
)
|
|
)
|
|
monkeypatch.setattr(flash, "run_final_check", check)
|
|
data = await flash.settle_flash_outcome(
|
|
context(tmp_path),
|
|
"预期搜索框为英文 Search settings",
|
|
{"status": "failed", "explanation": "已检查，实际为中文"},
|
|
final_report_received=True,
|
|
)
|
|
check.assert_awaited_once()
|
|
assert data["test_result"]["status"] == "failed"
|
|
assert "Search settings" in data["test_result"]["reason"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancellation_of_acceptance_propagates(tmp_path, monkeypatch):
|
|
check = AsyncMock(side_effect=asyncio.CancelledError)
|
|
monkeypatch.setattr(flash, "run_final_check", check)
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await flash.settle_flash_outcome(context(tmp_path), "任务", {"status": "completed"})
|
|
assert not (tmp_path / "run_outcome.json").exists()
|