【轻量拓展分析】 - 新增 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>
41 lines
1.6 KiB
Python
41 lines
1.6 KiB
Python
from pathlib import Path
|
|
|
|
|
|
INDEX_HTML = Path(__file__).resolve().parents[2] / "apps" / "web" / "vue3" / "index.html"
|
|
MAIN_PY = Path(__file__).resolve().parents[2] / "apps" / "web" / "vue3" / "main.py"
|
|
|
|
|
|
def test_execution_indicator_recovers_current_job_tasks_from_backend_runtime():
|
|
source = INDEX_HTML.read_text(encoding="utf-8")
|
|
|
|
assert "resource_tasks: []" in source
|
|
assert "runtimeOverview.resource_tasks = Array.isArray(data.resource_tasks) ? data.resource_tasks : [];" in source
|
|
assert "const currentJobRuntimeTasks = computed" in source
|
|
assert "String(task.job_id || '') === currentJobId" in source
|
|
assert "const executionIndicatorTask = computed" in source
|
|
|
|
|
|
def test_execution_indicator_visibility_and_stop_use_recovered_backend_task():
|
|
source = INDEX_HTML.read_text(encoding="utf-8")
|
|
|
|
assert 'v-if="showExecutionIndicator"' in source
|
|
assert "const showExecutionIndicator = computed" in source
|
|
assert "executionIndicatorTask.value" in source
|
|
assert "currentJobRuntimeTasks.value.forEach" in source
|
|
assert "task.task_id" in source
|
|
|
|
|
|
def test_execution_indicator_clears_only_after_backend_task_is_terminal():
|
|
source = INDEX_HTML.read_text(encoding="utf-8")
|
|
|
|
assert "const ACTIVE_RUNTIME_TASK_STATUSES" in source
|
|
assert "'queued', 'pending', 'running', 'started', 'waiting_resource', 'retry'" in source
|
|
assert "ACTIVE_RUNTIME_TASK_STATUSES.has" in source
|
|
|
|
|
|
def test_runtime_overview_does_not_truncate_active_resource_tasks():
|
|
source = MAIN_PY.read_text(encoding="utf-8")
|
|
|
|
assert 'overview["resource_tasks"] = resource.get("tasks", [])' in source
|
|
assert 'resource.get("tasks", [])[:20]' not in source
|