356 lines
14 KiB
Python
356 lines
14 KiB
Python
"""Workspace plugins use isolated persistence and never execute models or devices."""
|
|
|
|
import json
|
|
from io import BytesIO
|
|
import sys
|
|
from uuid import uuid4
|
|
import zipfile
|
|
|
|
from docx import Document
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from openpyxl import load_workbook
|
|
import pytest
|
|
|
|
from apps.admin_console.database.repositories.session_repository import SessionRepository
|
|
from apps.admin_console.database.repositories.step_repository import StepRepository
|
|
from apps.admin_console.routers import task_extensions as api
|
|
from apps.admin_console.services import batch_archive, report_export, task_extensions
|
|
from apps.admin_console.services.automation_store import AutomationStore
|
|
from artemis.data_engine.storage import StorageManager
|
|
from artemis.plugins.installer import install_archive
|
|
from artemis.plugins.models import PluginManifest, PluginSelection
|
|
from artemis.plugins.store import PluginStore
|
|
|
|
|
|
def manifest_payload():
|
|
return {
|
|
"schema_version": 3,
|
|
"id": "review",
|
|
"name": "人工评审",
|
|
"version": "1.0.0",
|
|
"platforms": [sys.platform],
|
|
"targets": [],
|
|
"capabilities": {"kind": "other"},
|
|
"workspace": {
|
|
"panels": [
|
|
{"id": "prompt", "title": "任务需求", "side": "left", "kind": "initial_prompt"},
|
|
{"id": "review", "title": "人工评审", "side": "right", "kind": "form"},
|
|
],
|
|
"fields": [
|
|
{
|
|
"id": "accuracy",
|
|
"label": "AI执行准确性",
|
|
"kind": "choice",
|
|
"options": ["Pass", "Failed"],
|
|
},
|
|
{"id": "accuracy_note", "label": "执行备注", "kind": "text", "max_length": 100},
|
|
{
|
|
"id": "correctness",
|
|
"label": "测试结果正确性",
|
|
"kind": "choice",
|
|
"options": ["Pass", "Failed"],
|
|
},
|
|
{"id": "correctness_note", "label": "结果备注", "kind": "text"},
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def install(plugins, original_case=False):
|
|
package = BytesIO()
|
|
manifest = manifest_payload()
|
|
if original_case:
|
|
manifest["workspace"]["panels"][0]["kind"] = "original_case"
|
|
with zipfile.ZipFile(package, "w") as archive:
|
|
archive.writestr("plugin.json", json.dumps(manifest))
|
|
install_archive(plugins, package.getvalue())
|
|
plugins.set_enabled("review", True)
|
|
|
|
|
|
@pytest.fixture
|
|
def context(tmp_path, monkeypatch):
|
|
path = tmp_path / "data.db"
|
|
StorageManager(db_path=path, base_trace_dir=tmp_path / "traces")
|
|
store = AutomationStore(path)
|
|
plugins = PluginStore(tmp_path / "plugins")
|
|
monkeypatch.setattr(task_extensions, "plugin_store", lambda: plugins)
|
|
for module in (api, report_export):
|
|
monkeypatch.setattr(module, "store", store)
|
|
monkeypatch.setattr(report_export, "SessionRepository", lambda *args: SessionRepository(path))
|
|
monkeypatch.setattr(report_export, "StepRepository", lambda *args: StepRepository(path))
|
|
monkeypatch.setattr(report_export, "TRACES_PATH", tmp_path / "traces")
|
|
monkeypatch.setattr(report_export, "EXPORT_ROOT", tmp_path / "exports")
|
|
store.save_job(
|
|
{
|
|
"session_id": "task-one",
|
|
"status": "completed",
|
|
"case": {
|
|
"name": "测试任务",
|
|
"goal": "# 最初的需求\n\n|项目|要求|\n|---|---|\n|页面|可见|",
|
|
},
|
|
}
|
|
)
|
|
app = FastAPI()
|
|
app.include_router(api.router)
|
|
with TestClient(app) as client:
|
|
yield store, plugins, client
|
|
|
|
|
|
def save(client, values, revision=0, request_id=None):
|
|
return client.put(
|
|
"/api/task-extensions/task-one/review",
|
|
json={
|
|
"values": values,
|
|
"expected_revision": revision,
|
|
"request_id": request_id or str(uuid4()),
|
|
},
|
|
)
|
|
|
|
|
|
def test_original_case_import_requires_enabled_plugin_and_keeps_execution_prompt(context):
|
|
from apps.admin_console.services import task_import
|
|
|
|
store, plugins, client = context
|
|
legacy = load_workbook(BytesIO(task_import.template()))
|
|
assert legacy["任务"].max_column == 7
|
|
legacy["任务"].append(["OLD", "", "", "旧用例", "", "原样执行", ""])
|
|
legacy_bytes = BytesIO()
|
|
legacy.save(legacy_bytes)
|
|
assert "original_case" not in task_import.parse_workbook(legacy_bytes.getvalue())["tasks"][0]
|
|
|
|
install(plugins, original_case=True)
|
|
workbook = load_workbook(BytesIO(task_import.template()))
|
|
assert [cell.value for cell in workbook["任务"][1]] == [
|
|
*task_import.HEADERS,
|
|
*task_import.CASE_HEADERS,
|
|
]
|
|
workbook["任务"].append(
|
|
[
|
|
"CASE-001",
|
|
"检测项目",
|
|
"测试子项",
|
|
"测试标题",
|
|
"提取的约束",
|
|
"只执行此提示词",
|
|
"{}",
|
|
"原始预置条件",
|
|
"1. 第一步\n2. 第二步",
|
|
"原始预期结果",
|
|
]
|
|
)
|
|
stream = BytesIO()
|
|
workbook.save(stream)
|
|
case = task_import.parse_workbook(stream.getvalue())["tasks"][0]
|
|
assert task_import.case_goal(case) == "只执行此提示词"
|
|
assert case["preconditions"] == "提取的约束"
|
|
assert case["original_case"] == {
|
|
"preconditions": "原始预置条件",
|
|
"steps": "1. 第一步\n2. 第二步",
|
|
"expected": "原始预期结果",
|
|
}
|
|
assert task_import.parse_workbook(legacy_bytes.getvalue())["valid"]
|
|
store.save_job({"session_id": "task-one", "status": "completed", "case": case})
|
|
result = client.get("/api/task-extensions/task-one").json()
|
|
assert "CASE-001" in result["case_title"] and "测试标题" in result["case_title"]
|
|
assert result["original_case"] == case["original_case"]
|
|
assert result["initial_prompt"] == "只执行此提示词"
|
|
assert report_export.session_report("task-one")["original_case"] == case["original_case"]
|
|
|
|
plugins.set_enabled("review", False)
|
|
assert load_workbook(BytesIO(task_import.template()))["任务"].max_column == 7
|
|
with pytest.raises(ValueError, match="列名或顺序"):
|
|
task_import.parse_workbook(stream.getvalue())
|
|
|
|
|
|
def test_review_navigation_uses_batch_order_and_excludes_unfinished_jobs(context):
|
|
store, _, client = context
|
|
store.create_batch({"id": "review-batch", "status": "completed"}, [], {})
|
|
for position, status in [(3, "completed"), (1, "failed"), (2, "pending")]:
|
|
store.save_job(
|
|
{
|
|
"session_id": f"batch-task-{position}",
|
|
"batch_id": "review-batch",
|
|
"position": position,
|
|
"status": status,
|
|
"case": {"name": f"用例{position}", "goal": "测试"},
|
|
}
|
|
)
|
|
first = client.get("/api/task-extensions/batch-task-1").json()["navigation"]
|
|
assert first == {
|
|
"scope": "batch",
|
|
"position": 1,
|
|
"total": 2,
|
|
"previous": None,
|
|
"next": "batch-task-3",
|
|
}
|
|
last = client.get("/api/task-extensions/batch-task-3").json()["navigation"]
|
|
assert last["previous"] == "batch-task-1" and last["next"] is None
|
|
|
|
|
|
def test_workspace_requires_v3_and_valid_bounded_declarations():
|
|
for change in (
|
|
{"schema_version": 2},
|
|
{
|
|
"workspace": {
|
|
"panels": [
|
|
{
|
|
"id": "prompt",
|
|
"title": "需求",
|
|
"side": "left",
|
|
"kind": "initial_prompt",
|
|
"width": {"minimum": 500, "preferred": 300},
|
|
}
|
|
]
|
|
}
|
|
},
|
|
{
|
|
"workspace": {
|
|
"panels": [{"id": "prompt", "title": "需求", "side": "left", "kind": "javascript"}]
|
|
}
|
|
},
|
|
):
|
|
with pytest.raises(ValueError):
|
|
PluginManifest.model_validate({**manifest_payload(), **change})
|
|
|
|
|
|
def test_plugin_is_optional_and_not_selected_for_execution(context):
|
|
store, plugins, client = context
|
|
assert client.get("/api/task-extensions/task-one").json()["plugins"] == []
|
|
assert save(client, {}).status_code == 409
|
|
install(plugins)
|
|
assert plugins.selections() == []
|
|
with pytest.raises(ValueError, match="工作区扩展"):
|
|
plugins.pin([PluginSelection(id="review")], "android")
|
|
data = client.get("/api/task-extensions/task-one").json()
|
|
assert data["initial_prompt"].startswith("# 最初的需求")
|
|
assert set(data["plugins"][0]["values"].values()) == {""}
|
|
assert save(client, {}).status_code == 200
|
|
assert store.get("case_review", "task-one") is None
|
|
|
|
|
|
def test_prompt_is_initial_session_request_not_case_or_step_text(context):
|
|
store, _, client = context
|
|
with store.connection() as conn:
|
|
conn.execute(
|
|
"INSERT INTO sessions(session_id,initial_goal,status,start_time) VALUES (?,?,?,?)",
|
|
("task-one", "原始任务需求", "completed", 1),
|
|
)
|
|
conn.commit()
|
|
assert client.get("/api/task-extensions/task-one").json()["initial_prompt"] == "原始任务需求"
|
|
|
|
|
|
def test_partial_clear_cas_and_idempotent_retry_preserve_history(context):
|
|
store, plugins, client = context
|
|
install(plugins)
|
|
request_id = str(uuid4())
|
|
assert (
|
|
save(
|
|
client, {"accuracy": "Pass", "accuracy_note": "人工核对"}, request_id=request_id
|
|
).status_code
|
|
== 200
|
|
)
|
|
assert (
|
|
save(
|
|
client, {"accuracy": "Pass", "accuracy_note": "人工核对"}, request_id=request_id
|
|
).status_code
|
|
== 200
|
|
)
|
|
assert save(client, {"accuracy": "Failed"}, request_id=request_id).status_code == 409
|
|
assert save(client, {"correctness": "Failed"}, revision=0).status_code == 409
|
|
assert save(client, {"correctness": "Failed"}, revision=1).status_code == 200
|
|
result = save(client, {"accuracy": "", "accuracy_note": ""}, revision=2).json()["plugins"][0]
|
|
assert result["values"]["accuracy"] == ""
|
|
assert result["values"]["correctness"] == "Failed"
|
|
assert len(result["history"]) == 3
|
|
assert result["history"][0]["values"]["accuracy"] == "Pass"
|
|
assert store.get("case_review", "task-one") is None
|
|
|
|
|
|
def test_invalid_running_missing_and_disabled_writes_are_rejected(context):
|
|
store, plugins, client = context
|
|
install(plugins)
|
|
for values in ({"unknown": "Pass"}, {"accuracy": "Maybe"}, {"accuracy_note": "x" * 101}):
|
|
assert save(client, values).status_code == 409
|
|
store.save_job({**store.job("task-one"), "status": "running"})
|
|
assert not client.get("/api/task-extensions/task-one").json()["editable"]
|
|
assert save(client, {"accuracy": "Pass"}).status_code == 409
|
|
assert client.get("/api/task-extensions/missing").status_code == 404
|
|
assert store.get("task_extensions", "task-one") is None
|
|
|
|
|
|
def test_disable_uninstall_reinstall_keep_history_and_gate_exports(context):
|
|
store, plugins, client = context
|
|
install(plugins)
|
|
save(client, {"accuracy": "Pass"})
|
|
plugins.set_enabled("review", False)
|
|
assert task_extensions.report_sections("task-one", store) == []
|
|
assert save(client, {"accuracy": "Failed"}, revision=1).status_code == 409
|
|
plugins.uninstall("review")
|
|
assert client.get("/api/task-extensions/task-one").json()["plugins"] == []
|
|
install(plugins)
|
|
data = client.get("/api/task-extensions/task-one").json()["plugins"][0]
|
|
assert data["values"]["accuracy"] == "Pass"
|
|
assert data["revision"] == 1
|
|
store.save_job({"session_id": "rerun", "status": "completed", "case": {"goal": "同样需求"}})
|
|
assert client.get("/api/task-extensions/rerun").json()["plugins"][0]["revision"] == 0
|
|
store.delete_history("task-one")
|
|
assert store.get("task_extensions", "task-one") is None
|
|
|
|
|
|
def test_word_excel_blank_fields_and_cache_refresh(context):
|
|
store, plugins, client = context
|
|
install(plugins)
|
|
original_result = report_export.session_report("task-one")["test_result"]
|
|
paths = report_export.export_session("task-one", ["docx", "xlsx"])
|
|
wb = load_workbook(paths[1])
|
|
assert wb["人工评审"].cell(2, 3).value is None
|
|
save(client, {"accuracy": "Pass", "accuracy_note": "=1+1"})
|
|
paths = report_export.export_session("task-one", ["docx", "xlsx"])
|
|
wb = load_workbook(paths[1])
|
|
assert wb["人工评审"].cell(2, 3).value == "Pass"
|
|
assert wb["人工评审"].cell(3, 3).data_type != "f"
|
|
assert "Pass" in " ".join(
|
|
cell.text for table in Document(paths[0]).tables for row in table.rows for cell in row.cells
|
|
)
|
|
assert report_export.session_report("task-one")["test_result"] == original_result
|
|
plugins.set_enabled("review", False)
|
|
paths = report_export.export_session("task-one", ["docx", "xlsx"])
|
|
assert "人工评审" not in load_workbook(paths[1]).sheetnames
|
|
assert not any("人工评审" in p.text for p in Document(paths[0]).paragraphs)
|
|
|
|
|
|
def test_batch_snapshot_signature_changes_on_review_and_plugin_state(context):
|
|
store, plugins, client = context
|
|
install(plugins)
|
|
store.create_batch(
|
|
{
|
|
"id": "batch",
|
|
"name": "批次",
|
|
"status": "completed",
|
|
"formats": ["docx", "xlsx"],
|
|
"report_ready": True,
|
|
},
|
|
[],
|
|
{},
|
|
)
|
|
job = store.job("task-one")
|
|
store.delete_history("task-one")
|
|
store.save_job({**job, "batch_id": "batch"})
|
|
path = report_export.export_batch("batch")
|
|
assert batch_archive.cached("batch", store.get("batch", "batch"))
|
|
with zipfile.ZipFile(path) as archive:
|
|
assert "人工评审" in load_workbook(BytesIO(archive.read("批次结果.xlsx"))).sheetnames
|
|
assert "人工评审" in " ".join(
|
|
p.text for p in Document(BytesIO(archive.read("批次汇总.docx"))).paragraphs
|
|
)
|
|
save(client, {"correctness": "Failed"})
|
|
# Even a late worker publishing report_ready cannot resurrect an old archive.
|
|
assert not batch_archive.cached("batch", {"report_ready": True})
|
|
report_export.export_batch("batch")
|
|
plugins.set_enabled("review", False)
|
|
assert not batch_archive.cached("batch", {"report_ready": True})
|
|
store.begin_delete_batch("batch")
|
|
store.delete_batch("batch")
|
|
assert store.get("task_extensions", "task-one") is None
|