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