Artemis-Eapil/tests/unit/agents/test_flash_acceptance.py

import asyncio
import json
from unittest.mock import AsyncMock, MagicMock
from langchain_core.messages import AIMessage

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()


@pytest.mark.asyncio
async def test_complete_checker_answer_is_persisted_without_reformatting(tmp_path, monkeypatch):
    from artemis.agents.checker import checker

    ctx = context(tmp_path)
    ctx.exploration_id = None
    ctx.on_execution_review = None
    result = CaseTestResult(
        status="passed",
        reason="双端布局探索和报告达标；不代表网站无缺陷。",
        evidence=["步骤 18、27：双端真实页底；报告列出待核实风险。"],
    )
    answer = AIMessage(
        content="最终裁定\n```json\n" + json.dumps({"test_result": result.model_dump()}) + "\n```"
    )
    llm = MagicMock()
    monkeypatch.setattr(checker, "get_llm", lambda **_: llm)
    monkeypatch.setattr(checker, "acomplete", AsyncMock(return_value=answer))
    monkeypatch.setattr(checker, "build_checker_tools", lambda *args: [])
    monkeypatch.setattr(checker, "_capture_final_screen", AsyncMock(return_value=(None, "footer")))
    monkeypatch.setattr(checker, "_load_steps", lambda *args: [])

    outcome = await flash.settle_flash_outcome(
        ctx, "如实报告布局风险，不要求网站无缺陷", {"status": "completed"}
    )

    llm.with_structured_output.assert_not_called()
    assert outcome["test_result"] == result.model_dump()
    saved = json.loads((tmp_path / "run_outcome.json").read_text(encoding="utf-8"))
    assert saved["test_result"] == result.model_dump()