Files
PlantAI-dev/tests/unit/test_autofill_entrypoints.py
potaserandClaude Opus 5 e641c8b9ab feat: 轻量拓展分析 + WGS2GWAS 阶段支持 + 结果区域重构 + 物种角色解析增强 + 会话 Job 隔离
【轻量拓展分析】
- 新增 gfa_core/extension_stage_lightweight/ 核心模块,支持用户上传代码/数据/原始命令
- 前端卡片式交互流程:候选核对 → 验证 → 结果审查 → 授权发布 → 运行
- Celery 不受限任务 + 轮询机制 + sessionStorage 操作恢复
- 启动脚本新增 BBAV5_LIGHTWEIGHT_* 环境变量

【WGS2GWAS 支持】
- 新增 13 个 WGS 阶段定义(QC → MAF/Missing Filter)
- email_haplotype_region.py 新增 LTR-RT 分析功能

【结果区域重构】
- 单层折叠 → 三组双层折叠:RNA-seq / 基因家族 / WGS2GWAS
- 结果按执行时间排序,保留最新尝试

【物种角色解析增强】
- 新增 _resolve_subfamily_request_species_roles()
- 支持 reference 角色的识别
- 直接 stage 请求保留 intent 中的物种角色

【会话 Job 隔离】
- rebind_job() 跨 job 时创建新会话,返回新 conversation_id

【前端交互增强】
- RAG 文献问答模式
- NL 触发 config/samples 自动填写、亚家族确认表、基因家族成员确认表
- 执行状态从后端 resource_tasks 恢复

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-05 09:40:45 +08:00

145 lines
5.6 KiB
Python
Executable File

from pathlib import Path
import sys
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from apps.cli.run_config_autofill import build_rows, merge_rows
from apps.cli.autofill_common import scan_files
from gfa_core.mas.agents.router_runtime import RouterRuntime
def test_direct_config_open_commands_map_to_config_dialog_action():
for query in (
"打开 config",
"打开 config.xlsx",
"重新打开配置表",
"帮我打开当前 job 的配置",
):
assert RouterRuntime._config_dialog_ui_action(query) == "open_config_dialog"
def test_config_discussion_does_not_map_to_config_dialog_action():
for query in (
"读取 config.xlsx 中有哪些物种",
"检查 config 是否正确",
"我已经打开 config,下一步做什么",
"config 为什么打不开",
"config 已确认,执行数据准备",
):
assert RouterRuntime._config_dialog_ui_action(query) is None
def test_at_spl_pep_upload_is_detected_as_seed_reference(tmp_path):
input_dir = tmp_path / "input"
input_dir.mkdir()
(input_dir / "At.SPL.pep.fa").write_text(">AT1G01010\nMAAA\n", encoding="utf-8")
rows = build_rows(
input_dir=input_dir,
source_dir=input_dir,
description="",
species_code="",
gene_family="SPL",
)
row = next(item for item in rows if item["Original_File_Name"] == "At.SPL.pep.fa")
assert row["Species_Code"] == "At"
assert row["Gene_Family"] == "SPL"
assert row["File_Type"] == "pep"
assert row["Ref"] == "TRUE"
assert row["Species_Role"] == "seed_reference"
assert row["File_Scope"] == "known_family"
assert row["confidence"] == "high"
def test_scan_files_uses_job_input_but_ignores_generated_or_archive_dirs(tmp_path):
input_dir = tmp_path / "input"
input_dir.mkdir()
(input_dir / "At.SPL.pep.fa").write_text(">AT\nMA\n", encoding="utf-8")
(input_dir / "final").mkdir()
(input_dir / "final" / "At.SPL.pep.fa").write_text(">generated\nMA\n", encoding="utf-8")
(input_dir / "uploaded_configs").mkdir()
(input_dir / "uploaded_configs" / "At.SPL.pep.fa").write_text(">archive\nMA\n", encoding="utf-8")
scanned = [path.relative_to(input_dir).as_posix() for path in scan_files(input_dir)]
assert scanned == ["At.SPL.pep.fa"]
def test_regenerated_draft_merges_new_upload_without_dropping_existing_rows(tmp_path):
input_dir = tmp_path / "input"
input_dir.mkdir()
(input_dir / "At.SPL.pep.fa").write_text(">AT1G01010\nMAAA\n", encoding="utf-8")
existing_rows = [
{
"Original_File_Name": "Arabidopsis_thaliana.TAIR10.pep.all.fa",
"Species_Code": "At",
"Gene_Family": "",
"File_Type": "pep",
"Ref": "TRUE",
"Species_Role": "seed_reference",
"File_Scope": "known_family",
"confidence": "high",
"notes": "",
}
]
detected_rows = build_rows(input_dir, input_dir, "", "", "SPL")
merged = merge_rows(existing_rows, detected_rows)
names = [row["Original_File_Name"] for row in merged]
assert "Arabidopsis_thaliana.TAIR10.pep.all.fa" in names
assert "At.SPL.pep.fa" in names
assert len(names) == len(set(names))
def test_frontend_autofill_entrypoints_keep_open_and_regenerate_separate():
html = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
queue_start = html.index("const queueAutofillFromPayload")
queue_end = html.index("const handleUploadSuccess", queue_start)
queue_block = html[queue_start:queue_end]
assert queue_block.count("sourceDir: ''") >= 2
assert "await openAutofillDialog('config'" in queue_block
assert "await openAutofillDialog('samples'" in queue_block
assert "const openExistingAutofillDraft" in html
assert "options.keepCurrentRows && !(autofillState.rows && autofillState.rows.length)" in html
assert "autoGenerate: wantsAutofillRegenerate" in html
def test_backend_open_endpoints_prefer_draft_then_confirmed_file():
main_py = (ROOT / "apps/web/vue3/main.py").read_text(encoding="utf-8")
assert '@app.post("/api/autofill/config/open")' in main_py
assert 'workspace_dir / "final" / "config.autofill.draft.xlsx"' in main_py
assert 'input_dir / "config.xlsx"' in main_py
assert '@app.post("/api/autofill/samples/open")' in main_py
assert 'workspace_dir / "final" / "samples.autofill.draft.yaml"' in main_py
assert 'input_dir / "samples.yaml"' in main_py
def test_chat_config_dialog_action_is_transmitted_and_opens_current_job_config():
main_py = (ROOT / "apps/web/vue3/main.py").read_text(encoding="utf-8")
html = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
assert '"ui_action": outcome.get("ui_action")' in main_py
assert "data.intent.ui_action === 'open_config_dialog'" in html
assert "await openAutofillDialog('config'" in html
assert "await openExistingAutofillDraft('config')" in html
def test_samples_confirm_no_longer_blocks_on_pending_de_roles_and_closes_dialog():
html = (ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8")
confirm_start = html.index("const confirmAutofillDraft")
confirm_end = html.index("const loadFromPath", confirm_start)
confirm_block = html[confirm_start:confirm_end]
assert "链特异性为必填" in confirm_block
assert "分组角色为必填" not in confirm_block
assert "请在样本表中标记唯一对照组" not in confirm_block
assert "请在样本表中至少标记一个比较组" not in confirm_block
assert "autofillDialogVisible.value = false" in confirm_block