239 lines
9.1 KiB
Python
Executable File
239 lines
9.1 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
|
|
from gfa_core.gene_family_stage_context import write_legacy_manifest_copy
|
|
from gfa_core.mas.tools.stage_status import get_stage_status
|
|
|
|
|
|
def test_legacy_manifest_alias_redirects_stage_status_to_canonical(tmp_path):
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_alias" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
|
|
scope_id = "SPL.target-Lc.ref-At.target_plus_reference"
|
|
member_set_id = f"{scope_id}:subfamily_prediction:all_members"
|
|
canonical = final_dir / "Target.SPL.target-Lc.ref-At.subfamily_prediction.phylogeny.manifest.json"
|
|
tree = final_dir / "Target.SPL.target-Lc.ref-At.subfamily_prediction.tree.nwk"
|
|
tree.write_text("(Lc:1,At:1);\n", encoding="utf-8")
|
|
canonical.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": "phylogeny",
|
|
"status": "success",
|
|
"prefix": "Lc.SPL",
|
|
"output_prefix": "Target.SPL.target-Lc.ref-At.subfamily_prediction",
|
|
"scope_id": scope_id,
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Lc"],
|
|
"reference_species": ["At"],
|
|
"analysis_route": "subfamily_prediction",
|
|
"member_set_id": member_set_id,
|
|
"artifacts": {
|
|
"tree": {"path": f"final/{tree.name}", "type": "newick"}
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
legacy = write_legacy_manifest_copy(canonical, "Lc.SPL", scope_id, "phylogeny")
|
|
assert legacy is not None
|
|
alias_manifest = json.loads(legacy.read_text(encoding="utf-8"))
|
|
assert alias_manifest["compatibility_alias"] is True
|
|
assert alias_manifest["alias_type"] == "legacy_scoped_manifest"
|
|
assert alias_manifest["canonical_manifest"] == f"final/{canonical.name}"
|
|
|
|
report = get_stage_status(
|
|
"job_alias",
|
|
"Lc.SPL",
|
|
"phylogeny",
|
|
workspace_root=workspace_root,
|
|
scope_id=scope_id,
|
|
analysis_route="subfamily_prediction",
|
|
member_set_id=member_set_id,
|
|
strict_identity=True,
|
|
)
|
|
|
|
assert report.completed is True
|
|
assert report.manifest_path == canonical
|
|
assert report.manifest is not None
|
|
assert not report.manifest.get("compatibility_alias")
|
|
assert report.artifacts_checked == 1
|
|
|
|
|
|
def test_pipeline_manifest_lookup_prefers_canonical_over_legacy_alias(tmp_path):
|
|
from apps.web.pipeline_backend import first_existing_stage_manifest
|
|
|
|
final_dir = tmp_path / "final"
|
|
final_dir.mkdir()
|
|
scope_data = {
|
|
"family": "SPL",
|
|
"scope_id": "SPL.target-Lc.ref-At.target_plus_reference",
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Lc"],
|
|
"reference_species": ["At"],
|
|
"comparative_species": [],
|
|
}
|
|
canonical = final_dir / "Target.SPL.target-Lc.ref-At.subfamily_prediction.phylogeny.manifest.json"
|
|
canonical.write_text(
|
|
json.dumps({"stage": "phylogeny", "status": "success", "prefix": "Lc.SPL"}),
|
|
encoding="utf-8",
|
|
)
|
|
legacy = write_legacy_manifest_copy(canonical, "Lc.SPL", scope_data["scope_id"], "phylogeny")
|
|
assert legacy is not None
|
|
|
|
assert first_existing_stage_manifest(final_dir, "phylogeny", "Lc.SPL", scope_data) == canonical
|
|
|
|
|
|
def test_ortholog_stage_status_rejects_mapping_target_species_mismatch(tmp_path):
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_ortholog_stale" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
|
|
scope_id = "SPL.target-Ft.ref-soe.target_plus_reference"
|
|
mapping = final_dir / "Ft.SPL.orthologs.txt"
|
|
mapping.write_text("FtGene1\tAT1G01010\t90\tFt\tAt\n", encoding="utf-8")
|
|
pairs = final_dir / "Ft.SPL.ortholog_pairs.txt"
|
|
pairs.write_text("FtGene1\tAT1G01010\t90\tFt\tAt\n", encoding="utf-8")
|
|
manifest = final_dir / "Ft.SPL.SPL.target-Ft.ref-soe.target_plus_reference.ortholog.manifest.json"
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": "ortholog",
|
|
"status": "success",
|
|
"prefix": "Ft.SPL",
|
|
"scope_id": scope_id,
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Ft"],
|
|
"reference_species": ["soe"],
|
|
"analysis_route": "multi_species",
|
|
"member_set_id": f"{scope_id}:multi_species:all_members",
|
|
"artifacts": {
|
|
"ortholog_mapping": {"path": f"final/{mapping.name}", "type": "tsv"},
|
|
"ortholog_pairs": {"path": f"final/{pairs.name}", "type": "tsv"},
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
report = get_stage_status(
|
|
"job_ortholog_stale",
|
|
"Ft.SPL",
|
|
"ortholog",
|
|
workspace_root=workspace_root,
|
|
scope_id=scope_id,
|
|
analysis_route="multi_species",
|
|
member_set_id=f"{scope_id}:multi_species:all_members",
|
|
strict_identity=True,
|
|
)
|
|
|
|
assert report.completed is False
|
|
assert report.status == "invalid_outputs"
|
|
assert any("Ortholog mapping reference species mismatch" in item for item in report.errors)
|
|
|
|
|
|
def test_stage_status_api_does_not_recover_invalid_ortholog_mapping_as_completed(tmp_path, monkeypatch):
|
|
from apps.web.vue3 import main as webmain
|
|
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_ortholog_api_stale" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
|
|
scope_id = "SPL.target-Ft.ref-soe.target_plus_reference"
|
|
member_set_id = f"{scope_id}:multi_species:all_members"
|
|
mapping = final_dir / "Ft.SPL.orthologs.txt"
|
|
mapping.write_text("FtGene1\tAT1G01010\t90\tFt\tAt\n", encoding="utf-8")
|
|
pairs = final_dir / "Ft.SPL.ortholog_pairs.txt"
|
|
pairs.write_text("FtGene1\tAT1G01010\t90\tFt\tAt\n", encoding="utf-8")
|
|
manifest = final_dir / "Ft.SPL.SPL.target-Ft.ref-soe.target_plus_reference.ortholog.manifest.json"
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": "ortholog",
|
|
"status": "success",
|
|
"prefix": "Ft.SPL",
|
|
"scope_id": scope_id,
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Ft"],
|
|
"reference_species": ["soe"],
|
|
"analysis_route": "multi_species",
|
|
"member_set_id": member_set_id,
|
|
"artifacts": {
|
|
"ortholog_mapping": {"path": f"final/{mapping.name}", "type": "tsv"},
|
|
"ortholog_pairs": {"path": f"final/{pairs.name}", "type": "tsv"},
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(webmain, "WORKSPACE_ROOT", workspace_root)
|
|
|
|
payload = asyncio.run(
|
|
webmain.get_stage_status(
|
|
job_id="job_ortholog_api_stale",
|
|
stage="ortholog",
|
|
prefix="Ft.SPL",
|
|
scope_id=scope_id,
|
|
analysis_route="multi_species",
|
|
member_set_id=member_set_id,
|
|
)
|
|
)
|
|
|
|
assert payload["completed"] is False
|
|
assert payload["status"] == "invalid_outputs"
|
|
assert any("Ortholog mapping reference species mismatch" in item for item in payload["errors"])
|
|
|
|
|
|
def test_unmarked_legacy_manifest_redirects_by_output_prefix(tmp_path):
|
|
workspace_root = tmp_path / "workspace"
|
|
final_dir = workspace_root / "job_unmarked" / "final"
|
|
final_dir.mkdir(parents=True)
|
|
|
|
scope_id = "SPL.target-Lc.ref-At.target_plus_reference"
|
|
canonical = final_dir / "Target.SPL.target-Lc.ref-At.subfamily_prediction.domain.manifest.json"
|
|
html = final_dir / "Target.SPL.target-Lc.ref-At.subfamily_prediction.domain.html"
|
|
html.write_text("<html></html>", encoding="utf-8")
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"stage": "domain",
|
|
"status": "success",
|
|
"prefix": "Lc.SPL",
|
|
"output_prefix": "Target.SPL.target-Lc.ref-At.subfamily_prediction",
|
|
"scope_id": scope_id,
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Lc"],
|
|
"reference_species": ["At"],
|
|
"analysis_route": "multi_species",
|
|
"member_set_id": f"{scope_id}:multi_species:all_members",
|
|
"artifacts": {"html": {"path": f"final/{html.name}", "type": "html"}},
|
|
}
|
|
canonical.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
legacy = final_dir / f"Lc.SPL.{scope_id}.domain.manifest.json"
|
|
legacy.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
report = get_stage_status(
|
|
"job_unmarked",
|
|
"Lc.SPL",
|
|
"domain",
|
|
workspace_root=workspace_root,
|
|
scope_id=scope_id,
|
|
analysis_route="multi_species",
|
|
member_set_id=f"{scope_id}:multi_species:all_members",
|
|
strict_identity=True,
|
|
)
|
|
|
|
assert report.completed is True
|
|
assert report.manifest_path == canonical
|