323 lines
15 KiB
Python
323 lines
15 KiB
Python
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from apps.admin_console.services.automation_store import AutomationStore
|
|
from apps.admin_console.services.exploration_service import ExplorationService
|
|
from artemis.config.exploration import ExplorationConfig
|
|
from artemis.exploration.models import (
|
|
CaseBatchRequest,
|
|
CreateExploration,
|
|
CaseDraft,
|
|
ValidateCaseRequest,
|
|
)
|
|
from artemis.exploration.store import ExplorationStore
|
|
from artemis.runtime import task_control
|
|
|
|
|
|
@pytest.fixture
|
|
def service(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path / "traces")
|
|
return ExplorationService(ExplorationStore(AutomationStore(tmp_path / "history.db")))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admission_uses_shared_queue_with_independent_model(service, monkeypatch):
|
|
enqueue = AsyncMock(return_value={"status": "queued", "tasks": []})
|
|
monkeypatch.setattr("apps.admin_console.routers.tasks.run_task", enqueue)
|
|
session = service.create(CreateExploration(goal="探索设置", request_id="create"))
|
|
await service.start(session["id"], "start")
|
|
request = enqueue.call_args.args[0]
|
|
assert request.profile == "pro"
|
|
assert request.runtime_overrides.model == "gpt-6.1-terra"
|
|
assert request.runtime_overrides.reasoning_effort == "high"
|
|
assert request.runtime_overrides.verification_level == "final"
|
|
assert request.conversation_id == session["id"]
|
|
first_id = request.session_id
|
|
await service.start(session["id"], "start")
|
|
assert enqueue.call_args.args[0].session_id == first_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validation_keeps_version_and_does_not_inherit_exploration_model(
|
|
service, monkeypatch
|
|
):
|
|
enqueue = AsyncMock(return_value={"status": "queued"})
|
|
monkeypatch.setattr("apps.admin_console.routers.tasks.run_task", enqueue)
|
|
session = service.create(
|
|
CreateExploration(
|
|
goal="探索设置",
|
|
request_id="create",
|
|
task_config={"model": "expensive", "reasoning_effort": "max"},
|
|
)
|
|
)
|
|
service.repository.save_case(session["id"], CaseDraft(name="设置", goal="复跑设置"), 0, "save")
|
|
await service.validate(session["id"], ValidateCaseRequest(request_id="validate", version=1))
|
|
request = enqueue.call_args.args[0]
|
|
assert request.profile == "flash"
|
|
assert request.runtime_overrides.model is None
|
|
assert request.runtime_overrides.reasoning_effort is None
|
|
assert request.goal == "复跑设置"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rejected_admission_releases_conversation_for_corrected_request(service, monkeypatch):
|
|
from fastapi import HTTPException
|
|
|
|
monkeypatch.setattr(
|
|
"apps.admin_console.routers.tasks.run_task",
|
|
AsyncMock(side_effect=HTTPException(409, "设备锁定")),
|
|
)
|
|
session = service.create(CreateExploration(goal="探索设置", request_id="create"))
|
|
with pytest.raises(HTTPException):
|
|
await service.start(session["id"], "start")
|
|
assert service.repository.get(session["id"])["active_run_id"] is None
|
|
assert service.repository.runs(session["id"])[0]["outcome"]["status"] == "rejected"
|
|
|
|
|
|
def test_retrying_answer_does_not_duplicate_guidance_or_answer_a_new_question(service):
|
|
session = service.create(CreateExploration(goal="探索设置", request_id="create"))
|
|
sid = session["id"]
|
|
run = service.repository.reserve_run(
|
|
sid, "run", kind="exploration", goal="探索", task_config={}
|
|
)
|
|
first = service.repository.ask(run["id"], "第一个问题", "建议", "q1")
|
|
reply = service.message(sid, "可以", "answer")
|
|
consumed = task_control.consume_instructions(task_control.control_dir(run["id"]))
|
|
assert consumed["instruction"] == "可以"
|
|
second = service.repository.ask(run["id"], "第二个问题", "建议", "q2")
|
|
assert service.message(sid, "可以", "answer") == reply
|
|
assert first["id"] == reply["question_id"]
|
|
assert task_control.consume_instructions(task_control.control_dir(run["id"])) is None
|
|
assert service.repository.get(sid)["pending_question"] == second["id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retrying_continue_after_run_ended_never_starts_another_run(service, monkeypatch):
|
|
session = service.create(CreateExploration(goal="探索", request_id="create"))
|
|
run = service.repository.reserve_run(
|
|
session["id"], "start", kind="exploration", goal="探索", task_config={}
|
|
)
|
|
monkeypatch.setattr(
|
|
"apps.admin_console.services.exploration_service.read_state",
|
|
lambda _: {"status": "task_paused"},
|
|
)
|
|
first = await service.resume(session["id"], "continue")
|
|
service.repository.finish_run(run["id"], {"status": "completed"})
|
|
assert await service.resume(session["id"], "continue") == first
|
|
assert len(service.repository.runs(session["id"])) == 1
|
|
|
|
|
|
def test_validation_guidance_is_delivered_and_counted_once(service):
|
|
session = service.create(CreateExploration(goal="探索", request_id="create"))
|
|
service.repository.save_case(session["id"], CaseDraft(name="设置", goal="复跑设置"), 0, "save")
|
|
run = service.repository.reserve_run(
|
|
session["id"], "validate", kind="validation", goal="复跑", task_config={}, case_version=1
|
|
)
|
|
service.message(session["id"], "只检查页面，不更改设置", "guide")
|
|
service.message(session["id"], "只检查页面，不更改设置", "guide")
|
|
assert service.repository.run(run["id"])["user_guidance_count"] == 1
|
|
assert (
|
|
task_control.consume_instructions(task_control.control_dir(run["id"]))["instruction"]
|
|
== "只检查页面，不更改设置"
|
|
)
|
|
|
|
|
|
def test_seal_rejects_unreviewed_guidance_then_refuses_silent_editorial_delivery(service):
|
|
from artemis.exploration.store import ExplorationConflict
|
|
|
|
session = service.create(CreateExploration(goal="探索", request_id="create"))
|
|
run = service.repository.reserve_run(
|
|
session["id"], "run", kind="exploration", goal="探索", task_config={}
|
|
)
|
|
service.message(session["id"], "先回到主页，再结束", "home")
|
|
assert not service.repository.seal_execution(run["id"], 0)
|
|
assert service.repository.seal_execution(run["id"], 1)
|
|
with pytest.raises(ExplorationConflict, match="指导尚未发送"):
|
|
service.message(session["id"], "再检查一次", "late")
|
|
assert service.repository.run(run["id"])["user_guidance_count"] == 1
|
|
assert service.repository.messages(session["id"])[-1]["text"] == "先回到主页，再结束"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_continue_cannot_skip_an_unanswered_question(service):
|
|
from artemis.exploration.store import ExplorationConflict
|
|
|
|
session = service.create(CreateExploration(goal="探索", request_id="create"))
|
|
run = service.repository.reserve_run(
|
|
session["id"], "run", kind="exploration", goal="探索", task_config={}
|
|
)
|
|
service.repository.ask(run["id"], "下一步？", "返回主页", "ask")
|
|
with pytest.raises(ExplorationConflict, match="先回答"):
|
|
await service.resume(session["id"], "resume")
|
|
|
|
|
|
def test_validation_uses_saved_case_title_without_replacing_its_full_goal(service):
|
|
from apps.admin_console.services.exploration_listing import exploration_metadata
|
|
|
|
session = service.create(CreateExploration(goal="探索设置", request_id="create"))
|
|
service.repository.save_case(
|
|
session["id"], CaseDraft(name="深色主题检查", goal="完整执行提示词"), 0, "save"
|
|
)
|
|
run = service.repository.reserve_run(
|
|
session["id"],
|
|
"validate",
|
|
kind="validation",
|
|
goal="完整执行提示词",
|
|
task_config={},
|
|
case_version=1,
|
|
)
|
|
with service.repository.database.connection() as connection:
|
|
metadata = exploration_metadata(connection)[run["id"]]
|
|
assert metadata["display_name"] == "深色主题检查"
|
|
assert metadata["case_version"] == 1
|
|
assert service.repository.run(run["id"])["goal"] == "完整执行提示词"
|
|
|
|
|
|
def test_execution_context_omits_duplicate_assistant_narration_but_retains_guidance(service):
|
|
session = service.create(CreateExploration(goal="探索设置", request_id="create"))
|
|
service.repository.add_message(session["id"], "正在读取页面", "narration", role="assistant")
|
|
service.repository.add_message(
|
|
session["id"], "下一步？", "question", role="assistant", metadata={"kind": "question"}
|
|
)
|
|
service.repository.add_message(session["id"], "返回主页", "home")
|
|
texts = [
|
|
m["text"] for m in service.repository.recent_messages(session["id"], guidance_only=True)
|
|
]
|
|
assert texts == ["探索设置", "下一步？", "返回主页"]
|
|
assert len(service.repository.recent_messages(session["id"])) == 4
|
|
|
|
|
|
def test_independent_saved_defaults_do_not_change_existing_sessions(service):
|
|
session = service.create(CreateExploration(goal="探索设置", request_id="create"))
|
|
service.save_settings(ExplorationConfig(model="gpt-6-astra", reasoning_effort="max"))
|
|
later = service.create(CreateExploration(goal="另一次探索", request_id="later"))
|
|
assert service.repository.get(session["id"])["config"]["model"] == "gpt-6.1-terra"
|
|
assert later["config"]["model"] == "gpt-6-astra"
|
|
|
|
|
|
def test_creation_rejects_secrets_and_unknown_task_options(service):
|
|
with pytest.raises(ValidationError):
|
|
service.create(
|
|
CreateExploration(goal="探索", request_id="create", task_config={"api_key": "secret"})
|
|
)
|
|
assert service.repository.list() == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fast_worker_state_cannot_be_replaced_with_queued(service, monkeypatch):
|
|
async def enqueue(request):
|
|
service.repository.set_run_state(request.session_id, "running")
|
|
return {"status": "started"}
|
|
|
|
monkeypatch.setattr("apps.admin_console.routers.tasks.run_task", enqueue)
|
|
session = service.create(CreateExploration(goal="探索设置", request_id="create"))
|
|
await service.start(session["id"], "start")
|
|
assert service.repository.get(session["id"])["status"] == "running"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rerun_includes_user_edited_preconditions(service, monkeypatch):
|
|
enqueue = AsyncMock(return_value={"status": "queued"})
|
|
monkeypatch.setattr("apps.admin_console.routers.tasks.run_task", enqueue)
|
|
session = service.create(CreateExploration(goal="探索", request_id="create"))
|
|
service.repository.save_case(
|
|
session["id"],
|
|
CaseDraft(name="用例", goal="观察设置", preconditions="已连接测试网络"),
|
|
0,
|
|
"save",
|
|
)
|
|
await service.validate(session["id"], ValidateCaseRequest(request_id="validate", version=1))
|
|
assert "已连接测试网络" in enqueue.call_args.args[0].goal
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("field", ["goal", "preconditions"])
|
|
async def test_unresolved_case_never_reserves_run_or_creates_batch(service, monkeypatch, field):
|
|
from apps.admin_console.services import automation_service
|
|
|
|
enqueue = AsyncMock()
|
|
monkeypatch.setattr("apps.admin_console.routers.tasks.run_task", enqueue)
|
|
monkeypatch.setattr(automation_service, "store", service.repository.database)
|
|
session = service.create(CreateExploration(goal="探索", request_id="create"))
|
|
draft = {"name": "旧版用例", "goal": "检查页面", field: "报告保存到 ${报告位置}"}
|
|
service.repository.save_case(session["id"], CaseDraft(**draft), 0, "save")
|
|
with pytest.raises(ValueError, match=r"未填写参数.*\$\{报告位置\}"):
|
|
await service.validate(session["id"], ValidateCaseRequest(request_id="validate", version=1))
|
|
with pytest.raises(ValueError, match="未填写参数"):
|
|
service.create_case_batch(session["id"], CaseBatchRequest(request_id="batch", version=1))
|
|
enqueue.assert_not_called()
|
|
assert service.repository.runs(session["id"]) == []
|
|
assert service.repository.database.records("batch") == []
|
|
assert service.repository.case(session["id"])["version"] == 1
|
|
|
|
# A saved correction can use the same admission request ID; rejection created no run.
|
|
service.repository.save_case(
|
|
session["id"],
|
|
CaseDraft(name="可执行", goal="检查页面，报告保存到本次任务 trace"),
|
|
1,
|
|
"edit",
|
|
)
|
|
enqueue.return_value = {"status": "queued"}
|
|
await service.validate(session["id"], ValidateCaseRequest(request_id="validate", version=2))
|
|
enqueue.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.parametrize("profile", ["flash", "pro"])
|
|
def test_case_batch_creates_only_an_idempotent_draft_with_bound_resources(
|
|
service, monkeypatch, profile
|
|
):
|
|
from apps.admin_console.services import automation_service
|
|
|
|
monkeypatch.setattr(automation_service, "store", service.repository.database)
|
|
session = service.create(
|
|
CreateExploration(
|
|
goal="探索设置",
|
|
request_id="create",
|
|
task_config={
|
|
"device_serial": "test-device",
|
|
"execution_target": "android",
|
|
"model": "exploration-model",
|
|
"reasoning_effort": "max",
|
|
"timeout_seconds": 900,
|
|
},
|
|
)
|
|
)
|
|
service.repository.save_case(
|
|
session["id"],
|
|
CaseDraft(name="设置", goal="观察设置", preconditions="已解锁"),
|
|
0,
|
|
"save",
|
|
)
|
|
request = CaseBatchRequest(request_id="batch", version=1, profile=profile)
|
|
batch = service.create_case_batch(session["id"], request)
|
|
assert service.create_case_batch(session["id"], request)["id"] == batch["id"]
|
|
assert batch["status"] == "draft"
|
|
jobs = service.repository.database.jobs(batch["id"])
|
|
assert len(jobs) == 1
|
|
assert jobs[0]["status"] == "pending"
|
|
case = jobs[0]["case"]
|
|
assert case["case_id"] == f"{session['id']}-v1"
|
|
assert case["category"] == "探索用例"
|
|
assert "已解锁" in case["goal"]
|
|
assert case["task_config"]["device_serial"] == "test-device"
|
|
assert case["task_config"]["execution_target"] == "android"
|
|
assert case["task_config"]["profile"] == profile
|
|
assert case["task_config"]["model"] is None
|
|
assert case["task_config"]["reasoning_effort"] is None
|
|
assert case["task_config"]["timeout_seconds"] is None
|
|
assert service.repository.runs(session["id"]) == []
|
|
|
|
|
|
def test_stale_validation_guidance_never_targets_a_new_active_run(service):
|
|
from artemis.exploration.store import ExplorationConflict
|
|
|
|
session = service.create(CreateExploration(goal="探索", request_id="create"))
|
|
run = service.repository.reserve_run(
|
|
session["id"], "run", kind="exploration", goal="探索", task_config={}
|
|
)
|
|
with pytest.raises(ExplorationConflict, match="所选执行已结束"):
|
|
service.message(session["id"], "旧复跑的指导", "stale", expected_run_id="previous-run")
|
|
assert task_control.list_instructions(run["id"]) == []
|