234 lines
8.5 KiB
Python
Executable File
234 lines
8.5 KiB
Python
Executable File
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gfa_core.mas.effective_members import (
|
|
MemberResolutionError,
|
|
resolve_effective_members,
|
|
)
|
|
|
|
|
|
def _write(path: Path, text: str) -> Path:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(text, encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def _pep(ids: list[str]) -> str:
|
|
return "".join(f">{gene_id}\nM{idx}A\n" for idx, gene_id in enumerate(ids, start=1))
|
|
|
|
|
|
def _confirmed_manifest(final_dir: Path, prefix: str, ids: list[str]) -> None:
|
|
manifest = {
|
|
"stage": "subfamily_confirm",
|
|
"status": "success",
|
|
"prefix": prefix,
|
|
"parameters": {
|
|
"final_members_source_stage": "subfamily_confirm",
|
|
"final_members_promoted": True,
|
|
},
|
|
"artifacts": {
|
|
"final_members_id": {"full_path": str(final_dir / f"{prefix}.final_members.id")},
|
|
"final_members_pep": {"full_path": str(final_dir / f"{prefix}.final_members.pep.fa")},
|
|
},
|
|
"metrics": {"final_members_count": len(ids)},
|
|
}
|
|
_write(final_dir / f"{prefix}.subfamily_confirm.manifest.json", json.dumps(manifest))
|
|
|
|
|
|
def test_post_confirm_required_uses_confirmed_final_not_legacy_identify(tmp_path: Path):
|
|
ws = tmp_path / "workspace" / "job1"
|
|
final_dir = ws / "final"
|
|
input_dir = tmp_path / "input" / "job1"
|
|
prefix = "Target.Os.CrR"
|
|
|
|
legacy_ids = [f"legacy{i}" for i in range(500)]
|
|
confirmed_ids = [f"confirmed{i}" for i in range(20)]
|
|
_write(final_dir / f"{prefix}.identified_members.id", "\n".join(legacy_ids) + "\n")
|
|
_write(final_dir / f"{prefix}.identified_members.pep.fa", _pep(legacy_ids))
|
|
_write(final_dir / f"{prefix}.final_members.id", "\n".join(confirmed_ids) + "\n")
|
|
_write(final_dir / f"{prefix}.final_members.pep.fa", _pep(confirmed_ids))
|
|
_confirmed_manifest(final_dir, prefix, confirmed_ids)
|
|
|
|
result = resolve_effective_members(
|
|
workspace=ws,
|
|
final_dir=final_dir,
|
|
input_dir=input_dir,
|
|
prefix=prefix,
|
|
phase="post_confirm_required",
|
|
require="pair",
|
|
)
|
|
|
|
assert result.path_id == final_dir / f"{prefix}.final_members.id"
|
|
assert result.path_pep == final_dir / f"{prefix}.final_members.pep.fa"
|
|
assert result.source_stage == "subfamily_confirm"
|
|
assert result.source_kind == "confirmed_final_members"
|
|
assert result.member_count == 20
|
|
|
|
|
|
def test_post_confirm_required_fails_without_confirmed_provenance(tmp_path: Path):
|
|
ws = tmp_path / "workspace" / "job1"
|
|
final_dir = ws / "final"
|
|
input_dir = tmp_path / "input" / "job1"
|
|
prefix = "Target.Os.CrR"
|
|
|
|
ids = ["a", "b"]
|
|
_write(final_dir / f"{prefix}.identified_members.id", "\n".join(ids) + "\n")
|
|
_write(final_dir / f"{prefix}.identified_members.pep.fa", _pep(ids))
|
|
|
|
with pytest.raises(MemberResolutionError, match="confirmed final members"):
|
|
resolve_effective_members(
|
|
workspace=ws,
|
|
final_dir=final_dir,
|
|
input_dir=input_dir,
|
|
prefix=prefix,
|
|
phase="post_confirm_required",
|
|
require="pair",
|
|
)
|
|
|
|
|
|
def test_pre_confirm_prefers_curated_before_identified(tmp_path: Path):
|
|
ws = tmp_path / "workspace" / "job1"
|
|
final_dir = ws / "final"
|
|
input_dir = tmp_path / "input" / "job1"
|
|
prefix = "Target.Os.CrR"
|
|
|
|
identified_ids = ["id1", "id2", "id3"]
|
|
curated_ids = ["id1", "id3"]
|
|
_write(final_dir / f"{prefix}.identified_members.id", "\n".join(identified_ids) + "\n")
|
|
_write(final_dir / f"{prefix}.identified_members.pep.fa", _pep(identified_ids))
|
|
_write(final_dir / f"{prefix}.curated_members.id", "\n".join(curated_ids) + "\n")
|
|
_write(final_dir / f"{prefix}.curated_members.pep.fa", _pep(curated_ids))
|
|
|
|
result = resolve_effective_members(
|
|
workspace=ws,
|
|
final_dir=final_dir,
|
|
input_dir=input_dir,
|
|
prefix=prefix,
|
|
phase="pre_confirm",
|
|
require="pair",
|
|
)
|
|
|
|
assert result.path_id == final_dir / f"{prefix}.curated_members.id"
|
|
assert result.path_pep == final_dir / f"{prefix}.curated_members.pep.fa"
|
|
assert result.source_kind == "curated_members"
|
|
assert result.member_count == 2
|
|
|
|
|
|
def test_subfamily_evidence_required_rejects_unreviewed_identified_members(tmp_path: Path):
|
|
ws = tmp_path / "workspace" / "job1"
|
|
final_dir = ws / "final"
|
|
input_dir = tmp_path / "input" / "job1"
|
|
prefix = "Target.Os.CrR"
|
|
|
|
identified_ids = ["id1", "id2", "id3"]
|
|
_write(final_dir / f"{prefix}.identified_members.id", "\n".join(identified_ids) + "\n")
|
|
_write(final_dir / f"{prefix}.identified_members.pep.fa", _pep(identified_ids))
|
|
|
|
with pytest.raises(MemberResolutionError, match="member review"):
|
|
resolve_effective_members(
|
|
workspace=ws,
|
|
final_dir=final_dir,
|
|
input_dir=input_dir,
|
|
prefix=prefix,
|
|
phase="subfamily_evidence_required", # type: ignore[arg-type]
|
|
require="pair",
|
|
)
|
|
|
|
|
|
def test_subfamily_evidence_required_uses_only_curated_members(tmp_path: Path):
|
|
ws = tmp_path / "workspace" / "job1"
|
|
final_dir = ws / "final"
|
|
input_dir = tmp_path / "input" / "job1"
|
|
prefix = "Target.Os.CrR"
|
|
|
|
curated_ids = ["id1", "id3"]
|
|
final_ids = ["id1", "id2", "id3", "id4"]
|
|
_write(final_dir / f"{prefix}.curated_members.id", "\n".join(curated_ids) + "\n")
|
|
_write(final_dir / f"{prefix}.curated_members.pep.fa", _pep(curated_ids))
|
|
_write(final_dir / f"{prefix}.final_members.id", "\n".join(final_ids) + "\n")
|
|
_write(final_dir / f"{prefix}.final_members.pep.fa", _pep(final_ids))
|
|
|
|
result = resolve_effective_members(
|
|
workspace=ws,
|
|
final_dir=final_dir,
|
|
input_dir=input_dir,
|
|
prefix=prefix,
|
|
phase="subfamily_evidence_required", # type: ignore[arg-type]
|
|
require="pair",
|
|
)
|
|
|
|
assert result.path_id == final_dir / f"{prefix}.curated_members.id"
|
|
assert result.path_pep == final_dir / f"{prefix}.curated_members.pep.fa"
|
|
assert result.source_kind == "curated_members"
|
|
assert result.member_count == 2
|
|
|
|
|
|
def test_legacy_compatible_reads_old_final_members_with_warning(tmp_path: Path):
|
|
ws = tmp_path / "workspace" / "job1"
|
|
final_dir = ws / "final"
|
|
input_dir = tmp_path / "input" / "job1"
|
|
prefix = "Target.Os.CrR"
|
|
|
|
ids = ["old1", "old2"]
|
|
_write(final_dir / f"{prefix}.final_members.id", "\n".join(ids) + "\n")
|
|
_write(final_dir / f"{prefix}.final_members.pep.fa", _pep(ids))
|
|
|
|
result = resolve_effective_members(
|
|
workspace=ws,
|
|
final_dir=final_dir,
|
|
input_dir=input_dir,
|
|
prefix=prefix,
|
|
phase="legacy_compatible",
|
|
require="pair",
|
|
)
|
|
|
|
assert result.source_kind == "legacy_identify_final_members"
|
|
assert result.member_count == 2
|
|
assert any("legacy" in warning for warning in result.warnings)
|
|
|
|
|
|
def test_pipeline_backend_member_helpers_use_effective_resolver(tmp_path: Path):
|
|
from apps.web.pipeline_backend import resolve_single_target_gene_id_file, resolve_species_family_pep
|
|
|
|
ws = tmp_path / "workspace" / "job1"
|
|
final_dir = ws / "final"
|
|
input_dir = tmp_path / "input" / "job1"
|
|
prefix = "Target.Os.CrR"
|
|
species = "Os"
|
|
family = "CrR"
|
|
|
|
identified_ids = [f"identified{i}" for i in range(500)]
|
|
confirmed_ids = [f"confirmed{i}" for i in range(20)]
|
|
_write(final_dir / f"{prefix}.identified_members.id", "\n".join(identified_ids) + "\n")
|
|
_write(final_dir / f"{prefix}.identified_members.pep.fa", _pep(identified_ids))
|
|
_write(final_dir / f"{prefix}.final_members.id", "\n".join(confirmed_ids) + "\n")
|
|
_write(final_dir / f"{prefix}.final_members.pep.fa", _pep(confirmed_ids))
|
|
_confirmed_manifest(final_dir, prefix, confirmed_ids)
|
|
|
|
assert resolve_single_target_gene_id_file(ws, input_dir, final_dir, species, prefix) == (
|
|
final_dir / f"{prefix}.final_members.id"
|
|
)
|
|
assert resolve_species_family_pep(ws, final_dir, input_dir, species, family) == (
|
|
final_dir / f"{prefix}.final_members.pep.fa"
|
|
)
|
|
|
|
|
|
def test_pipeline_backend_post_confirm_does_not_fallback_to_curated(tmp_path: Path):
|
|
from apps.web.pipeline_backend import resolve_single_target_gene_id_file
|
|
|
|
ws = tmp_path / "workspace" / "job1"
|
|
final_dir = ws / "final"
|
|
input_dir = tmp_path / "input" / "job1"
|
|
prefix = "Target.Os.CrR"
|
|
species = "Os"
|
|
family = "CrR"
|
|
|
|
curated_ids = ["c1", "c2"]
|
|
_write(final_dir / f"{prefix}.curated_members.id", "\n".join(curated_ids) + "\n")
|
|
_write(final_dir / f"{prefix}.curated_members.pep.fa", _pep(curated_ids))
|
|
_confirmed_manifest(final_dir, prefix, curated_ids)
|
|
|
|
assert resolve_single_target_gene_id_file(ws, input_dir, final_dir, species, prefix) is None
|