1121 lines
40 KiB
Python
Executable File
1121 lines
40 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
|
|
|
|
def _write_ft_technical_reference_manifest(final_dir, stage, *, status="success"):
|
|
scope_id = "SPL.target-Ft.ref-soe.target_plus_reference"
|
|
artifact = final_dir / f"Ft.SPL.{stage}.result.tsv"
|
|
artifact.write_text(f"{stage}\tresult\n", encoding="utf-8")
|
|
manifest = final_dir / f"Ft.SPL.{stage}.manifest.json"
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": stage,
|
|
"status": status,
|
|
"prefix": "Ft.SPL",
|
|
"output_prefix": "Ft.SPL",
|
|
"scope_id": scope_id,
|
|
"scope_mode": "target_plus_reference",
|
|
"analysis_route": "multi_species",
|
|
"member_set_id": f"{scope_id}:multi_species:all_members",
|
|
"target_species": ["Ft"],
|
|
"reference_species": ["soe"],
|
|
"artifacts": {
|
|
"result": {"path": f"final/{artifact.name}", "type": "tsv"},
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return manifest
|
|
|
|
|
|
def _write_regular_gene_family_manifest(
|
|
final_dir,
|
|
stage,
|
|
*,
|
|
scope_id="SPL.target-Ft.ref-At.target_plus_reference",
|
|
route="multi_species",
|
|
references=None,
|
|
member_set_id="",
|
|
filename="",
|
|
with_artifact=True,
|
|
):
|
|
references = ["At"] if references is None else list(references)
|
|
member_set_id = member_set_id or f"{scope_id}:{route}:all_members"
|
|
artifact = final_dir / f"Ft.SPL.{stage}.result.tsv"
|
|
if with_artifact:
|
|
artifact.write_text(f"{stage}\tresult\n", encoding="utf-8")
|
|
manifest = final_dir / (filename or f"Ft.SPL.scoped.{stage}.manifest.json")
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": stage,
|
|
"status": "success",
|
|
"prefix": "Ft.SPL",
|
|
"scope_id": scope_id,
|
|
"scope_mode": "target_plus_reference" if references else "target_only",
|
|
"analysis_route": route,
|
|
"member_set_id": member_set_id,
|
|
"family": "SPL",
|
|
"target_species": ["Ft"],
|
|
"reference_species": references,
|
|
"artifacts": (
|
|
{"result": {"path": f"final/{artifact.name}", "type": "tsv"}}
|
|
if with_artifact
|
|
else {"result": {"path": "final/missing.tsv", "type": "tsv"}}
|
|
),
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return manifest
|
|
|
|
|
|
def _confirmed_contextual_ppi_fixture(web_main, workspace_root, job_id, intent_params):
|
|
from types import SimpleNamespace
|
|
|
|
final_dir = workspace_root / job_id / "final"
|
|
final_dir.mkdir(parents=True)
|
|
(final_dir / "Ft.SPL.curated_members.pep.fa").write_text(">Ft1\nMA\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.curated_members.id").write_text("Ft1\n", encoding="utf-8")
|
|
confirmed_file = final_dir / "Ft.SPL.confirmed_subfamily.tsv"
|
|
confirmed_file.write_text("Ft1\tGroupA\n", encoding="utf-8")
|
|
_write_ft_technical_reference_manifest(final_dir, "ppi")
|
|
request = SimpleNamespace(
|
|
job_id=job_id,
|
|
prefix="Ft.SPL",
|
|
message="执行Ft的完整基因家族分析",
|
|
intent={"params": dict(intent_params)},
|
|
)
|
|
contextual_request = {
|
|
"family": "SPL",
|
|
"prefix": "Ft.SPL",
|
|
"stages": list(web_main.FULL_GENE_FAMILY_WORKFLOW_STAGES),
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": [],
|
|
"comparativeSpecies": [],
|
|
"scopeMode": "target_only",
|
|
"contextGate": {
|
|
"valid": True,
|
|
"evidence": {
|
|
"family_members": {
|
|
"Ft": {"pep": str(final_dir / "Ft.SPL.curated_members.pep.fa")},
|
|
},
|
|
},
|
|
},
|
|
"context_route": "post_subfamily_gene_family_analysis",
|
|
"requiresConfirmedSubfamily": True,
|
|
"skipSubfamilyLabels": False,
|
|
"subfamilyState": {
|
|
"all_confirmed": True,
|
|
"by_species": {
|
|
"Ft": {
|
|
"status": "confirmed_valid",
|
|
"confirmed_file": str(confirmed_file),
|
|
},
|
|
},
|
|
},
|
|
}
|
|
return request, contextual_request
|
|
|
|
|
|
def test_attach_stage_status_does_not_complete_nonterminal_task_from_old_manifest(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_old" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
tree = final_dir / "Lc.SPL.tree.nwk"
|
|
tree.write_text("(a,b);", encoding="utf-8")
|
|
manifest = final_dir / "Lc.SPL.phylogeny.manifest.json"
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": "phylogeny",
|
|
"status": "success",
|
|
"prefix": "Lc.SPL",
|
|
"artifacts": {"tree": {"path": f"final/{tree.name}", "type": "newick"}},
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
task = {
|
|
"id": "task-running",
|
|
"job_id": "job_old",
|
|
"stage": "phylogeny",
|
|
"prefix": "Lc.SPL",
|
|
"status": "queued",
|
|
"message": "任务已提交，等待 Celery worker 接收",
|
|
"created_at": "2026-05-29T14:38:32",
|
|
}
|
|
|
|
web_main._attach_single_stage_status(task)
|
|
|
|
assert task["status"] == "queued"
|
|
assert "stage_status" not in task
|
|
|
|
|
|
def test_attach_single_stage_status_does_not_recover_failed_wgs_qc_from_stale_success_manifest(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
from gfa_core.mas.tools import stage_identity
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
job_id = "job_stale_wgs_qc"
|
|
prefix = "ThreeSample"
|
|
final_dir = workspace_root / job_id / "final"
|
|
final_dir.mkdir(parents=True)
|
|
sample_ids = ["102", "108", "112"]
|
|
artifacts = {}
|
|
|
|
def artifact_record(path, role, sample_id=""):
|
|
contents = path.read_bytes()
|
|
record = {
|
|
"role": role,
|
|
"path": f"final/{path.name}",
|
|
"sha256": hashlib.sha256(contents).hexdigest(),
|
|
"size": len(contents),
|
|
"size_bytes": len(contents),
|
|
}
|
|
if sample_id:
|
|
record["sample_id"] = sample_id
|
|
return record
|
|
|
|
config = final_dir / "wgs_samples.yaml"
|
|
config.write_text("samples: {}\n", encoding="utf-8")
|
|
artifacts["wgs_sample_config"] = artifact_record(config, "wgs_sample_config")
|
|
for sample_id in sample_ids:
|
|
for role in ("clean_reads_r1", "clean_reads_r2", "fastp_html", "fastp_json"):
|
|
artifact = final_dir / f"{prefix}.wgs_qc.{sample_id}.{role}"
|
|
artifact.write_text(f"{sample_id} {role}\n", encoding="utf-8")
|
|
artifacts[f"{role}.{sample_id}"] = artifact_record(artifact, role, sample_id)
|
|
stale_success_manifest = {
|
|
"manifest_version": "1.0",
|
|
"stage": "wgs_qc",
|
|
"status": "success",
|
|
"prefix": prefix,
|
|
"job_id": job_id,
|
|
"wgs_sample_set_id": "wgs-samples:stale-three-sample-set",
|
|
"wgs_analysis_id": "wgs-analysis:stale-success-run",
|
|
"metrics": {"input_sample_ids": sample_ids, "output_sample_ids": sample_ids},
|
|
"artifacts": artifacts,
|
|
}
|
|
(final_dir / f"{prefix}.wgs_qc.manifest.json").write_text(
|
|
json.dumps(stale_success_manifest), encoding="utf-8"
|
|
)
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
original_get_status = stage_identity.get_stage_status_for_identity
|
|
|
|
def get_status_in_test_workspace(job_id, identity, **kwargs):
|
|
return original_get_status(
|
|
job_id,
|
|
identity,
|
|
workspace_root=workspace_root,
|
|
**kwargs,
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
stage_identity,
|
|
"get_stage_status_for_identity",
|
|
get_status_in_test_workspace,
|
|
)
|
|
task = {
|
|
"id": "failed-wgs-qc-task",
|
|
"job_id": job_id,
|
|
"stage": "wgs_qc",
|
|
"prefix": prefix,
|
|
"status": "failed",
|
|
"result": {
|
|
"status": "failed",
|
|
"success": False,
|
|
"result": {
|
|
"status": "error",
|
|
"message": "wgs_qc requires exactly one wgs_samples.yaml configuration",
|
|
"stage": "wgs_qc",
|
|
"manifest": stale_success_manifest,
|
|
},
|
|
},
|
|
}
|
|
|
|
web_main._attach_single_stage_status(task)
|
|
|
|
assert task["status"] == "failed"
|
|
assert task["result"]["success"] is False
|
|
assert not task["result"].get("recovered_from_manifest", False)
|
|
|
|
|
|
def test_identity_completed_stages_for_plan_respects_target_only_scope(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_plan" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
scope_id = "SPL.target-Lc.target_only"
|
|
tree = final_dir / "Lc.SPL.tree.nwk"
|
|
tree.write_text("(Lc1,Lc2);", encoding="utf-8")
|
|
manifest = final_dir / "Lc.SPL.phylogeny.manifest.json"
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": "phylogeny",
|
|
"status": "success",
|
|
"prefix": "Lc.SPL",
|
|
"output_prefix": "Lc.SPL",
|
|
"scope_id": scope_id,
|
|
"scope_mode": "target_only",
|
|
"analysis_route": "single_species",
|
|
"member_set_id": f"{scope_id}:single_species:all_members",
|
|
"target_species": ["Lc"],
|
|
"reference_species": [],
|
|
"artifacts": {"tree": {"path": f"final/{tree.name}", "type": "newick"}},
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
plan_params = {
|
|
"prefix": "Lc.SPL",
|
|
"geneFamily": "SPL",
|
|
"targetSpecies": ["Lc"],
|
|
"scopeMode": "target_only",
|
|
"analysisRoute": "single_species",
|
|
"memberSetId": f"{scope_id}:single_species:all_members",
|
|
}
|
|
completed = web_main._identity_completed_stages_for_plan(
|
|
"job_plan",
|
|
["phylogeny", "motif"],
|
|
plan_params,
|
|
[
|
|
{"id": "phylogeny", "params": dict(plan_params)},
|
|
{"id": "motif", "params": dict(plan_params)},
|
|
],
|
|
request_prefix="Lc.SPL",
|
|
)
|
|
|
|
assert completed == ["phylogeny"]
|
|
|
|
|
|
def test_identity_completed_stages_prunes_technical_reference_results_from_regular_target_only_plan(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_ft_technical" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
stage_ids = ["ortholog", "kaks", "ppi"]
|
|
for stage_id in stage_ids:
|
|
_write_ft_technical_reference_manifest(final_dir, stage_id)
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
plan_params = {
|
|
"prefix": "Ft.SPL",
|
|
"geneFamily": "SPL",
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": [],
|
|
"scopeMode": "target_only",
|
|
"analysisRoute": "single_species",
|
|
}
|
|
completed = web_main._identity_completed_stages_for_plan(
|
|
"job_ft_technical",
|
|
stage_ids,
|
|
plan_params,
|
|
[{"id": stage_id, "params": dict(plan_params)} for stage_id in stage_ids],
|
|
request_prefix="Ft.SPL",
|
|
)
|
|
|
|
assert completed == stage_ids
|
|
|
|
|
|
def test_identity_completed_stages_does_not_substitute_different_explicit_reference(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_ft_explicit_at" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
_write_ft_technical_reference_manifest(final_dir, "ppi")
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
plan_params = {
|
|
"prefix": "Ft.SPL",
|
|
"geneFamily": "SPL",
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": ["At"],
|
|
"scopeMode": "target_only",
|
|
"analysisRoute": "single_species",
|
|
}
|
|
|
|
completed = web_main._identity_completed_stages_for_plan(
|
|
"job_ft_explicit_at",
|
|
["ppi"],
|
|
plan_params,
|
|
[{"id": "ppi", "params": dict(plan_params)}],
|
|
request_prefix="Ft.SPL",
|
|
)
|
|
|
|
assert completed == []
|
|
|
|
|
|
def test_identity_completed_stages_does_not_inherit_manifest_for_explicit_scope_file(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
for scope_key in ("scopeFile", "scope_file"):
|
|
job_id = f"job_ft_{scope_key}"
|
|
final_dir = workspace_root / job_id / "final"
|
|
final_dir.mkdir(parents=True)
|
|
_write_ft_technical_reference_manifest(final_dir, "ppi")
|
|
plan_params = {
|
|
"prefix": "Ft.SPL",
|
|
"geneFamily": "SPL",
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": [],
|
|
"scopeMode": "target_only",
|
|
"analysisRoute": "single_species",
|
|
scope_key: "explicit-at.scope.json",
|
|
}
|
|
|
|
completed = web_main._identity_completed_stages_for_plan(
|
|
job_id,
|
|
["ppi"],
|
|
plan_params,
|
|
[{"id": "ppi", "params": dict(plan_params)}],
|
|
request_prefix="Ft.SPL",
|
|
)
|
|
|
|
assert completed == [], scope_key
|
|
|
|
|
|
def test_identity_completed_stages_does_not_prune_failed_or_partial_technical_results(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_ft_non_success" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
_write_ft_technical_reference_manifest(final_dir, "ortholog", status="failed")
|
|
_write_ft_technical_reference_manifest(final_dir, "ppi", status="partial_success")
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
plan_params = {
|
|
"prefix": "Ft.SPL",
|
|
"geneFamily": "SPL",
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": [],
|
|
"scopeMode": "target_only",
|
|
"analysisRoute": "single_species",
|
|
}
|
|
|
|
completed = web_main._identity_completed_stages_for_plan(
|
|
"job_ft_non_success",
|
|
["ortholog", "ppi"],
|
|
plan_params,
|
|
[
|
|
{"id": "ortholog", "params": dict(plan_params)},
|
|
{"id": "ppi", "params": dict(plan_params)},
|
|
],
|
|
request_prefix="Ft.SPL",
|
|
)
|
|
|
|
assert completed == []
|
|
|
|
|
|
def test_identity_completed_stages_force_rerun_keeps_technical_stages_pending(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_ft_force_rerun" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
stage_ids = ["ortholog", "kaks", "ppi"]
|
|
for stage_id in stage_ids:
|
|
_write_ft_technical_reference_manifest(final_dir, stage_id)
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
plan_params = {
|
|
"prefix": "Ft.SPL",
|
|
"geneFamily": "SPL",
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": [],
|
|
"scopeMode": "target_only",
|
|
"analysisRoute": "single_species",
|
|
}
|
|
|
|
completed = web_main._identity_completed_stages_for_plan(
|
|
"job_ft_force_rerun",
|
|
stage_ids,
|
|
plan_params,
|
|
[{"id": stage_id, "params": dict(plan_params)} for stage_id in stage_ids],
|
|
request_prefix="Ft.SPL",
|
|
force_rerun_stage_ids=set(stage_ids),
|
|
)
|
|
|
|
assert completed == []
|
|
|
|
|
|
def test_post_confirm_plan_recovers_regular_manifest_identity(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_regular_recovery" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
confirm = final_dir / "Ft.SPL.subfamily_confirm.manifest.json"
|
|
confirm.write_text('{"stage":"subfamily_confirm","status":"success","prefix":"Ft.SPL"}', encoding="utf-8")
|
|
manifest = _write_regular_gene_family_manifest(final_dir, "phylogeny")
|
|
os.utime(confirm, (1, 1))
|
|
os.utime(manifest, (2, 2))
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
scope_id = "SPL.target-Ft.ref-At.target_plus_reference"
|
|
plan_params = {
|
|
"prefix": "Ft.SPL",
|
|
"geneFamily": "SPL",
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": ["At"],
|
|
"workflowPhase": "post_subfamily_confirm",
|
|
"requiresConfirmedSubfamily": True,
|
|
}
|
|
stage_params = {
|
|
**plan_params,
|
|
"scopeMode": "target_plus_reference",
|
|
"analysisRoute": "multi_species",
|
|
"memberSetId": f"{scope_id}:confirmed_subfamily:all_members",
|
|
}
|
|
|
|
completed = web_main._identity_completed_stages_for_plan(
|
|
"job_regular_recovery",
|
|
["phylogeny"],
|
|
plan_params,
|
|
[{"id": "phylogeny", "params": stage_params}],
|
|
request_prefix="Ft.SPL",
|
|
)
|
|
|
|
assert completed == ["phylogeny"]
|
|
|
|
|
|
def test_post_confirm_plan_tries_next_manifest_after_incomplete_alias(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_regular_alias" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
confirm = final_dir / "Ft.SPL.subfamily_confirm.manifest.json"
|
|
confirm.write_text('{"stage":"subfamily_confirm","status":"success","prefix":"Ft.SPL"}', encoding="utf-8")
|
|
canonical = _write_regular_gene_family_manifest(
|
|
final_dir,
|
|
"synteny",
|
|
route="single_species",
|
|
filename="Ft.SPL.scoped.synteny.manifest.json",
|
|
)
|
|
alias = _write_regular_gene_family_manifest(
|
|
final_dir,
|
|
"synteny",
|
|
route="single_species",
|
|
filename="Ft.SPL.synteny.manifest.json",
|
|
with_artifact=False,
|
|
)
|
|
alias_payload = json.loads(alias.read_text(encoding="utf-8"))
|
|
for key in ("scope_id", "scope_mode", "analysis_route", "member_set_id"):
|
|
alias_payload.pop(key, None)
|
|
alias.write_text(json.dumps(alias_payload), encoding="utf-8")
|
|
os.utime(confirm, (1, 1))
|
|
os.utime(canonical, (2, 2))
|
|
os.utime(alias, (3, 3))
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
target_scope = "SPL.target-Ft.target_only"
|
|
plan_params = {
|
|
"prefix": "Ft.SPL",
|
|
"geneFamily": "SPL",
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": ["At"],
|
|
"workflowPhase": "post_subfamily_confirm",
|
|
"requiresConfirmedSubfamily": True,
|
|
}
|
|
stage_params = {
|
|
**plan_params,
|
|
"referenceSpecies": [],
|
|
"scopeMode": "target_only",
|
|
"analysisRoute": "single_species",
|
|
"memberSetId": f"{target_scope}:confirmed_subfamily:all_members",
|
|
}
|
|
|
|
completed = web_main._identity_completed_stages_for_plan(
|
|
"job_regular_alias",
|
|
["synteny"],
|
|
plan_params,
|
|
[{"id": "synteny", "params": stage_params}],
|
|
request_prefix="Ft.SPL",
|
|
)
|
|
|
|
assert completed == ["synteny"]
|
|
|
|
|
|
def test_post_confirm_plan_rejects_result_older_than_confirmation(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_regular_stale" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
manifest = _write_regular_gene_family_manifest(final_dir, "phylogeny")
|
|
confirm = final_dir / "Ft.SPL.subfamily_confirm.manifest.json"
|
|
confirm.write_text('{"stage":"subfamily_confirm","status":"success","prefix":"Ft.SPL"}', encoding="utf-8")
|
|
os.utime(manifest, (1, 1))
|
|
os.utime(confirm, (2, 2))
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
scope_id = "SPL.target-Ft.ref-At.target_plus_reference"
|
|
plan_params = {
|
|
"prefix": "Ft.SPL",
|
|
"geneFamily": "SPL",
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": ["At"],
|
|
"workflowPhase": "post_subfamily_confirm",
|
|
"requiresConfirmedSubfamily": True,
|
|
}
|
|
stage_params = {
|
|
**plan_params,
|
|
"scopeMode": "target_plus_reference",
|
|
"analysisRoute": "multi_species",
|
|
"memberSetId": f"{scope_id}:confirmed_subfamily:all_members",
|
|
}
|
|
|
|
completed = web_main._identity_completed_stages_for_plan(
|
|
"job_regular_stale",
|
|
["phylogeny"],
|
|
plan_params,
|
|
[{"id": "phylogeny", "params": stage_params}],
|
|
request_prefix="Ft.SPL",
|
|
)
|
|
|
|
assert completed == []
|
|
|
|
|
|
def test_post_confirm_kaks_inherits_completed_shared_ppi_reference(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_contextual_kaks" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
_write_ft_technical_reference_manifest(final_dir, "kaks")
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
plan_params = {
|
|
"prefix": "Ft.SPL",
|
|
"geneFamily": "SPL",
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": ["At"],
|
|
"workflowPhase": "post_subfamily_confirm",
|
|
"requiresConfirmedSubfamily": True,
|
|
}
|
|
stage_params = {
|
|
**plan_params,
|
|
"scopeMode": "target_plus_reference",
|
|
"analysisRoute": "multi_species",
|
|
"memberSetId": "SPL.target-Ft.ref-At.target_plus_reference:confirmed_subfamily:all_members",
|
|
}
|
|
|
|
completed = web_main._identity_completed_stages_for_plan(
|
|
"job_contextual_kaks",
|
|
["kaks"],
|
|
plan_params,
|
|
[{"id": "kaks", "params": stage_params}],
|
|
request_prefix="Ft.SPL",
|
|
)
|
|
|
|
assert completed == ["kaks"]
|
|
|
|
|
|
|
|
|
|
def test_contextual_plan_returns_only_pending_stages(tmp_path, monkeypatch):
|
|
from types import SimpleNamespace
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_context" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
scope_id = "SPL.target-Lc.target_only"
|
|
tree = final_dir / "Lc.SPL.tree.nwk"
|
|
tree.write_text("(Lc1,Lc2);", encoding="utf-8")
|
|
(final_dir / "Lc.SPL.phylogeny.manifest.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": "phylogeny",
|
|
"status": "success",
|
|
"prefix": "Lc.SPL",
|
|
"output_prefix": "Lc.SPL",
|
|
"scope_id": scope_id,
|
|
"scope_mode": "target_only",
|
|
"analysis_route": "single_species",
|
|
"member_set_id": f"{scope_id}:single_species:all_members",
|
|
"target_species": ["Lc"],
|
|
"reference_species": [],
|
|
"artifacts": {"tree": {"path": f"final/{tree.name}", "type": "newick"}},
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
request = SimpleNamespace(job_id="job_context", prefix="Lc.SPL", message="帮我执行顺式作用元件预测")
|
|
contextual_request = {
|
|
"family": "SPL",
|
|
"prefix": "Lc.SPL",
|
|
"stages": ["phylogeny", "promoter"],
|
|
"targetSpecies": ["Lc"],
|
|
"referenceSpecies": [],
|
|
"comparativeSpecies": [],
|
|
"scopeMode": "target_only",
|
|
"contextGate": {},
|
|
"context_route": "gene_family_default_scope",
|
|
"requiresConfirmedSubfamily": False,
|
|
"skipSubfamilyLabels": True,
|
|
"subfamilyState": {},
|
|
}
|
|
|
|
response = web_main._build_contextual_gene_family_plan_response(request, "session-test", contextual_request)
|
|
|
|
assert response["completed_stages"] == ["phylogeny"]
|
|
assert response["plan"]["all_stages"] == ["phylogeny", "promoter"]
|
|
assert response["plan"]["stages"] == ["promoter"]
|
|
assert [item["id"] for item in response["plan"]["stages_detail"]] == ["promoter"]
|
|
|
|
|
|
def test_contextual_gene_family_plan_uses_subfamily_evidence_before_confirmation(tmp_path, monkeypatch):
|
|
from types import SimpleNamespace
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_context_subfamily" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
(final_dir / "Ft.SPL.final_members.pep.fa").write_text(">Ft1\nMA\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.final_members.id").write_text("Ft1\n", encoding="utf-8")
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
request = SimpleNamespace(job_id="job_context_subfamily", prefix="Ft.SPL", message="执行Ft的基因家族分析")
|
|
contextual_request = {
|
|
"family": "SPL",
|
|
"prefix": "Ft.SPL",
|
|
"stages": [
|
|
"phylogeny",
|
|
"motif",
|
|
"domain",
|
|
"structure",
|
|
"combo",
|
|
"promoter",
|
|
],
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": ["At"],
|
|
"comparativeSpecies": [],
|
|
"scopeMode": "target_plus_reference",
|
|
"contextGate": {
|
|
"valid": False,
|
|
"issues": [{"code": "subfamily_classification_missing"}],
|
|
"evidence": {"family_members": {"Ft": {"pep": str(final_dir / "Ft.SPL.final_members.pep.fa")}}},
|
|
},
|
|
"context_route": "gene_family_default_scope",
|
|
"requiresConfirmedSubfamily": True,
|
|
"skipSubfamilyLabels": False,
|
|
"subfamilyState": {
|
|
"all_confirmed": False,
|
|
"by_species": {
|
|
"Ft": {
|
|
"status": "not_classified",
|
|
"confirmed_file": "",
|
|
"uploaded_subfamily_file": "",
|
|
}
|
|
},
|
|
},
|
|
}
|
|
|
|
response = web_main._build_contextual_gene_family_plan_response(request, "session-test", contextual_request)
|
|
details = {item["id"]: item for item in response["plan"]["all_stages_detail"]}
|
|
|
|
assert response["plan"]["all_stages"][:7] == [
|
|
"phylogeny",
|
|
"motif",
|
|
"domain",
|
|
"structure",
|
|
"combo",
|
|
"reference_classification",
|
|
"subfamily_confirm",
|
|
]
|
|
for stage_id in ["phylogeny", "motif", "domain", "structure", "combo"]:
|
|
assert details[stage_id]["params"]["analysisRoute"] == "subfamily_evidence"
|
|
assert details[stage_id]["params"]["requiresConfirmedSubfamily"] is False
|
|
assert details["reference_classification"]["dependencies"] == ["combo"]
|
|
assert details["reference_classification"]["params"]["analysisRoute"] == "subfamily_prediction"
|
|
assert details["subfamily_confirm"]["params"]["analysisRoute"] == "confirmed_subfamily"
|
|
|
|
|
|
def test_contextual_request_does_not_treat_reference_subfamily_as_target_confirmation(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
input_root = tmp_path / "input"
|
|
final_dir = workspace_root / "job_ref_label" / "final"
|
|
input_dir = input_root / "job_ref_label"
|
|
final_dir.mkdir(parents=True)
|
|
input_dir.mkdir(parents=True)
|
|
(final_dir / "Ft.SPL.curated_members.pep.fa").write_text(">Ft1\nMA\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.curated_members.id").write_text("Ft1\n", encoding="utf-8")
|
|
(input_dir / "At.SPL.subfamily").write_text("AT1\tGroupA\n", encoding="utf-8")
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
monkeypatch.setattr(web_main, "INPUT_ROOT", input_root)
|
|
monkeypatch.setattr(
|
|
web_main,
|
|
"_load_identify_species_selection",
|
|
lambda job_id: {
|
|
"groups": {
|
|
"target": [{"species": "Ft"}],
|
|
"reference": [{"species": "At"}],
|
|
"comparative": [],
|
|
}
|
|
},
|
|
)
|
|
monkeypatch.setattr(web_main, "_gene_family_from_config", lambda path: "SPL")
|
|
monkeypatch.setattr(web_main, "_extract_species_codes_from_text_for_job", lambda job_id, text: ["Ft"])
|
|
|
|
contextual = web_main._resolve_contextual_gene_family_request("job_ref_label", "执行Ft的基因家族分析", "Ft.SPL")
|
|
|
|
assert contextual["targetSpecies"] == ["Ft"]
|
|
assert contextual["referenceSpecies"] == ["At"]
|
|
assert contextual["scopeMode"] == "target_plus_reference"
|
|
assert contextual["confirmed_subfamily_available"] is False
|
|
assert contextual["confirmed_subfamily_files"] == {"Ft": ""}
|
|
assert contextual["subfamilyState"]["overall"] == "not_classified"
|
|
|
|
|
|
def test_contextual_plan_for_predicted_unconfirmed_only_requires_confirmation(tmp_path, monkeypatch):
|
|
from types import SimpleNamespace
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_predicted" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
(final_dir / "Ft.SPL.final_members.pep.fa").write_text(">Ft1\nMA\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.final_members.id").write_text("Ft1\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.reference_classification.subfamily.tsv").write_text("Ft1\tGroupA\n", encoding="utf-8")
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
request = SimpleNamespace(job_id="job_predicted", prefix="Ft.SPL", message="执行Ft的基因家族分析")
|
|
contextual_request = {
|
|
"family": "SPL",
|
|
"prefix": "Ft.SPL",
|
|
"stages": ["phylogeny", "motif", "domain", "structure", "combo", "promoter"],
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": ["At"],
|
|
"comparativeSpecies": [],
|
|
"scopeMode": "target_plus_reference",
|
|
"contextGate": {"valid": False, "issues": [{"code": "subfamily_prediction_unconfirmed"}]},
|
|
"context_route": "gene_family_default_scope",
|
|
"requiresConfirmedSubfamily": True,
|
|
"skipSubfamilyLabels": False,
|
|
"subfamilyState": {
|
|
"all_confirmed": False,
|
|
"by_species": {
|
|
"Ft": {
|
|
"status": "predicted_unconfirmed",
|
|
"prediction_file": str(final_dir / "Ft.SPL.reference_classification.subfamily.tsv"),
|
|
}
|
|
},
|
|
},
|
|
}
|
|
|
|
response = web_main._build_contextual_gene_family_plan_response(request, "session-test", contextual_request)
|
|
|
|
assert response["plan"]["all_stages"] == ["subfamily_confirm"]
|
|
assert response["plan"]["stages"] == ["subfamily_confirm"]
|
|
assert response["plan"]["workflowPhase"] == "subfamily_prerequisite"
|
|
assert response["plan"]["postConfirmReplan"] is True
|
|
detail = response["plan"]["all_stages_detail"][0]
|
|
assert detail["id"] == "subfamily_confirm"
|
|
assert detail["params"]["analysisRoute"] == "confirmed_subfamily"
|
|
|
|
|
|
def test_contextual_plan_after_confirmation_keeps_formal_confirmed_gate(tmp_path, monkeypatch):
|
|
from types import SimpleNamespace
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_confirmed" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
(final_dir / "Ft.SPL.curated_members.pep.fa").write_text(">Ft1\nMA\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.curated_members.id").write_text("Ft1\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.confirmed_subfamily.tsv").write_text("Ft1\tGroupA\n", encoding="utf-8")
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
request = SimpleNamespace(job_id="job_confirmed", prefix="Ft.SPL", message="执行Ft的基因家族分析")
|
|
contextual_request = {
|
|
"family": "SPL",
|
|
"prefix": "Ft.SPL",
|
|
"stages": ["phylogeny", "motif", "domain", "structure", "combo", "promoter"],
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": [],
|
|
"comparativeSpecies": [],
|
|
"scopeMode": "target_only",
|
|
"contextGate": {"valid": True, "evidence": {"family_members": {"Ft": {"pep": str(final_dir / "Ft.SPL.curated_members.pep.fa")}}}},
|
|
"context_route": "post_subfamily_gene_family_analysis",
|
|
"requiresConfirmedSubfamily": True,
|
|
"skipSubfamilyLabels": False,
|
|
"subfamilyState": {
|
|
"all_confirmed": True,
|
|
"by_species": {
|
|
"Ft": {
|
|
"status": "confirmed_valid",
|
|
"confirmed_file": str(final_dir / "Ft.SPL.confirmed_subfamily.tsv"),
|
|
}
|
|
},
|
|
},
|
|
}
|
|
|
|
response = web_main._build_contextual_gene_family_plan_response(request, "session-test", contextual_request)
|
|
details = {item["id"]: item for item in response["plan"]["all_stages_detail"]}
|
|
|
|
assert "reference_classification" not in response["plan"]["all_stages"]
|
|
assert response["plan"]["workflowPhase"] == "post_subfamily_confirm"
|
|
assert details["phylogeny"]["params"]["analysisRoute"] == "single_species"
|
|
assert details["phylogeny"]["params"]["requiresConfirmedSubfamily"] is True
|
|
|
|
|
|
def test_contextual_full_plan_prunes_completed_technical_reference_ppi(tmp_path, monkeypatch):
|
|
from types import SimpleNamespace
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_confirmed_ppi" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
(final_dir / "Ft.SPL.curated_members.pep.fa").write_text(">Ft1\nMA\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.curated_members.id").write_text("Ft1\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.confirmed_subfamily.tsv").write_text("Ft1\tGroupA\n", encoding="utf-8")
|
|
_write_ft_technical_reference_manifest(final_dir, "ppi")
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
request = SimpleNamespace(
|
|
job_id="job_confirmed_ppi",
|
|
prefix="Ft.SPL",
|
|
message="执行Ft的完整基因家族分析",
|
|
)
|
|
contextual_request = {
|
|
"family": "SPL",
|
|
"prefix": "Ft.SPL",
|
|
"stages": list(web_main.FULL_GENE_FAMILY_WORKFLOW_STAGES),
|
|
"targetSpecies": ["Ft"],
|
|
"referenceSpecies": [],
|
|
"comparativeSpecies": [],
|
|
"scopeMode": "target_only",
|
|
"contextGate": {
|
|
"valid": True,
|
|
"evidence": {
|
|
"family_members": {
|
|
"Ft": {"pep": str(final_dir / "Ft.SPL.curated_members.pep.fa")},
|
|
},
|
|
},
|
|
},
|
|
"context_route": "post_subfamily_gene_family_analysis",
|
|
"requiresConfirmedSubfamily": True,
|
|
"skipSubfamilyLabels": False,
|
|
"subfamilyState": {
|
|
"all_confirmed": True,
|
|
"by_species": {
|
|
"Ft": {
|
|
"status": "confirmed_valid",
|
|
"confirmed_file": str(final_dir / "Ft.SPL.confirmed_subfamily.tsv"),
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
response = web_main._build_contextual_gene_family_plan_response(
|
|
request,
|
|
"session-confirmed-ppi",
|
|
contextual_request,
|
|
)
|
|
expected_pending = [
|
|
stage_id for stage_id in web_main.FULL_GENE_FAMILY_WORKFLOW_STAGES
|
|
if stage_id != "ppi"
|
|
]
|
|
|
|
assert "ppi" in response["completed_stages"]
|
|
assert response["plan"]["stages"] == expected_pending
|
|
assert [item["id"] for item in response["plan"]["stages_detail"]] == expected_pending
|
|
|
|
|
|
def test_contextual_full_plan_keeps_force_rerun_ppi_from_request_intent(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
request, contextual_request = _confirmed_contextual_ppi_fixture(
|
|
web_main,
|
|
workspace_root,
|
|
"job_confirmed_ppi_force",
|
|
{"forceRerunStages": ["ppi"]},
|
|
)
|
|
|
|
response = web_main._build_contextual_gene_family_plan_response(
|
|
request,
|
|
"session-confirmed-ppi-force",
|
|
contextual_request,
|
|
)
|
|
|
|
assert "ppi" not in response["completed_stages"]
|
|
assert "ppi" in response["plan"]["stages"]
|
|
assert "ppi" in [item["id"] for item in response["plan"]["stages_detail"]]
|
|
|
|
|
|
def test_contextual_full_plan_keeps_ppi_for_explicit_request_identity(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as web_main
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
at_scope_id = "SPL.target-Ft.ref-At.target_plus_reference"
|
|
at_scope_file = tmp_path / "explicit-at.scope.json"
|
|
at_scope_file.write_text(
|
|
json.dumps({
|
|
"family": "SPL",
|
|
"scope_id": at_scope_id,
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Ft"],
|
|
"reference_species": ["At"],
|
|
"comparative_species": [],
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
cases = {
|
|
"scope_id": {
|
|
"scopeId": at_scope_id,
|
|
"memberSetId": f"{at_scope_id}:multi_species:all_members",
|
|
},
|
|
"scope_file": {"scopeFile": str(at_scope_file)},
|
|
}
|
|
for case_name, intent_params in cases.items():
|
|
job_id = f"job_confirmed_ppi_{case_name}"
|
|
request, contextual_request = _confirmed_contextual_ppi_fixture(
|
|
web_main,
|
|
workspace_root,
|
|
job_id,
|
|
intent_params,
|
|
)
|
|
|
|
response = web_main._build_contextual_gene_family_plan_response(
|
|
request,
|
|
f"session-{case_name}",
|
|
contextual_request,
|
|
)
|
|
|
|
assert "ppi" not in response["completed_stages"], case_name
|
|
assert "ppi" in response["plan"]["stages"], case_name
|
|
assert "ppi" in [item["id"] for item in response["plan"]["stages_detail"]], case_name
|
|
|
|
|
|
def test_attach_single_stage_status_completes_successful_wgs_index(
|
|
tmp_path, monkeypatch
|
|
):
|
|
from apps.web.vue3 import main as web_main
|
|
from gfa_core.mas.tools import stage_identity
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
job_id = "job_wgs_index"
|
|
prefix = "Os"
|
|
final_dir = workspace_root / job_id / "final"
|
|
final_dir.mkdir(parents=True)
|
|
|
|
def artifact_record(path, role, component=""):
|
|
contents = path.read_bytes()
|
|
record = {
|
|
"role": role,
|
|
"path": f"final/{path.name}",
|
|
"sha256": hashlib.sha256(contents).hexdigest(),
|
|
"size": len(contents),
|
|
"size_bytes": len(contents),
|
|
}
|
|
if component:
|
|
record["component"] = component
|
|
return record
|
|
|
|
reference = final_dir / f"{prefix}.wgs_index.reference.fasta"
|
|
reference.write_text(">chr1\nACGT\n", encoding="utf-8")
|
|
fai = final_dir / f"{reference.name}.fai"
|
|
fai.write_text("chr1\t4\t6\t4\t5\n", encoding="utf-8")
|
|
dictionary = final_dir / f"{prefix}.wgs_index.reference.dict"
|
|
dictionary.write_text("@HD\tVN:1.6\n@SQ\tSN:chr1\tLN:4\n", encoding="utf-8")
|
|
artifacts = {
|
|
"reference_fasta": artifact_record(reference, "reference_fasta"),
|
|
"reference_fai": artifact_record(fai, "reference_fai"),
|
|
"reference_dict": artifact_record(dictionary, "reference_dict"),
|
|
}
|
|
for component in (".0123", ".amb", ".ann", ".bwt.2bit.64", ".pac"):
|
|
path = final_dir / f"{reference.name}{component}"
|
|
path.write_text(component, encoding="utf-8")
|
|
artifacts[f"bwa_mem2_index_component.{component}"] = artifact_record(
|
|
path, "bwa_mem2_index_component", component
|
|
)
|
|
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"stage": "wgs_index",
|
|
"status": "success",
|
|
"prefix": prefix,
|
|
"job_id": job_id,
|
|
"wgs_reference_id": "wgs-reference:" + "b" * 64,
|
|
"wgs_analysis_id": "wgs-analysis:" + "c" * 64,
|
|
"artifacts": artifacts,
|
|
}
|
|
manifest_path = final_dir / f"{prefix}.wgs_index.manifest.json"
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
|
|
original_get_status = stage_identity.get_stage_status_for_identity
|
|
|
|
def get_status_in_test_workspace(job_id, identity, **kwargs):
|
|
kwargs.pop("workspace_root", None)
|
|
return original_get_status(
|
|
job_id,
|
|
identity,
|
|
workspace_root=workspace_root,
|
|
**kwargs,
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
stage_identity,
|
|
"get_stage_status_for_identity",
|
|
get_status_in_test_workspace,
|
|
)
|
|
task = {
|
|
"id": "successful-wgs-index-task",
|
|
"job_id": job_id,
|
|
"stage": "wgs_index",
|
|
"prefix": prefix,
|
|
"status": "success",
|
|
"result": {
|
|
"status": "completed",
|
|
"success": True,
|
|
"current_stage": "wgs_index",
|
|
"result": {
|
|
"status": "success",
|
|
"stage": "wgs_index",
|
|
"manifest": manifest,
|
|
"manifest_path": str(manifest_path),
|
|
},
|
|
},
|
|
}
|
|
|
|
web_main._attach_single_stage_status(task)
|
|
|
|
assert task["status"] == "success"
|
|
assert task["result"]["success"] is True
|
|
assert task["stage_status"]["completed"] is True
|
|
assert task["stage_status"]["errors"] == []
|
|
assert task["stage_status"]["artifacts_checked"] == 8
|