91 lines
3.4 KiB
Python
91 lines
3.4 KiB
Python
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
FRONTEND_PATH = PROJECT_ROOT / "apps/web/vue3/index_ml.html"
|
|
I18N_PATH = PROJECT_ROOT / "apps/web/vue3/static/i18n_en.js"
|
|
|
|
|
|
def _run_empty_plan_branch(tmp_path: Path, plan_data: dict) -> dict:
|
|
frontend = FRONTEND_PATH.read_text(encoding="utf-8")
|
|
start = frontend.index("const plan = planData.plan;")
|
|
end = frontend.index("// 显示智能规划结果", start)
|
|
branch = frontend[start:end]
|
|
script = tmp_path / "empty-plan-branch.cjs"
|
|
script.write_text(
|
|
"const planData=" + json.dumps(plan_data, ensure_ascii=False) + ";"
|
|
"const events={messages:[],filters:[],refreshes:[]};"
|
|
"const t=value=>value;"
|
|
"const getStageLabel=id=>({genome_prepare:'0️⃣ Genome Prepare',ppi:'PPI分析'}[id]||id);"
|
|
"const setResultStageFilter=id=>events.filters.push(id);"
|
|
"const refreshResults=async(jobId,id)=>events.refreshes.push([jobId,id]);"
|
|
"const addAgentMessage=(role,message,meta)=>events.messages.push({role,message,meta});"
|
|
"const jobInfo={job_id:'job-test'};"
|
|
"const nlLoading={value:true};const nlInput={value:'request'};"
|
|
"async function run(){" + branch + "}"
|
|
"run().then(()=>process.stdout.write(JSON.stringify(events)));",
|
|
encoding="utf-8",
|
|
)
|
|
completed = subprocess.run(
|
|
["/usr/bin/node", str(script)],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
assert completed.returncode == 0, completed.stderr
|
|
return json.loads(completed.stdout)
|
|
|
|
|
|
def test_completed_full_workflow_does_not_masquerade_as_genome_prepare(tmp_path: Path) -> None:
|
|
requested = ["phylogeny", "motif", "domain", "structure", "combo", "promoter"]
|
|
result = _run_empty_plan_branch(
|
|
tmp_path,
|
|
{
|
|
"success": True,
|
|
"plan": {
|
|
"stages": [],
|
|
"all_stages": requested,
|
|
"description": "所有阶段已完成",
|
|
},
|
|
"completed_stages": [*requested, "genome_prepare"],
|
|
},
|
|
)
|
|
|
|
assert result["filters"] == []
|
|
assert result["refreshes"] == []
|
|
assert [item["message"] for item in result["messages"]] == [
|
|
"当前请求中的分析阶段均已有同一分析范围的有效结果，无需重复执行。"
|
|
]
|
|
|
|
|
|
def test_completed_single_stage_keeps_stage_specific_result_handoff(tmp_path: Path) -> None:
|
|
result = _run_empty_plan_branch(
|
|
tmp_path,
|
|
{
|
|
"success": True,
|
|
"plan": {
|
|
"stages": [],
|
|
"all_stages": ["ppi"],
|
|
"description": "所有阶段已完成",
|
|
},
|
|
"completed_stages": ["ppi"],
|
|
},
|
|
)
|
|
|
|
assert result["filters"] == ["ppi"]
|
|
assert result["refreshes"] == [["job-test", "ppi"]]
|
|
assert [item["message"] for item in result["messages"]] == [
|
|
"PPI分析 已有同一分析范围的有效结果，无需重复执行。"
|
|
]
|
|
|
|
|
|
def test_generic_completed_plan_label_has_english_translation() -> None:
|
|
translations = I18N_PATH.read_text(encoding="utf-8")
|
|
|
|
assert (
|
|
'"当前请求中的分析阶段均已有同一分析范围的有效结果，无需重复执行。": '
|
|
'"All analysis stages in the current request already have valid results for the same analysis scope. No need to re-execute."'
|
|
) in translations
|