546 lines
23 KiB
Python
546 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from openpyxl import Workbook
|
|
|
|
from apps.web import pipeline_backend as backend
|
|
from gfa_core.ortholog_materialize import (
|
|
OrthologRunRecord,
|
|
build_ortholog_run_identity,
|
|
materialize_ortholog_final_outputs,
|
|
ortholog_run_identity_matches,
|
|
write_ortholog_run_sidecar,
|
|
)
|
|
|
|
|
|
def _job(tmp_path: Path) -> backend.JobInfo:
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
(workspace / "final").mkdir(parents=True)
|
|
(workspace / "logs").mkdir(parents=True)
|
|
input_dir.mkdir()
|
|
return backend.JobInfo(
|
|
job_id="task5b",
|
|
input_dir=str(input_dir),
|
|
workspace_dir=str(workspace),
|
|
logs_dir=str(workspace / "logs"),
|
|
created_at="",
|
|
)
|
|
|
|
|
|
def test_ppi_owned_ortholog_resolves_storage_pep_and_preflight_reuses_matching_mapping(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
job = _job(tmp_path)
|
|
input_dir = Path(job.input_dir)
|
|
final_dir = Path(job.workspace_dir) / "final"
|
|
storage_dir = input_dir / "string_taxon_4555"
|
|
storage_dir.mkdir()
|
|
workbook = Workbook()
|
|
sheet = workbook.active
|
|
sheet.title = "config"
|
|
sheet.append(["Species_Code", "Species_Name", "TaxID"])
|
|
sheet.append(["Sv", "Setaria viridis", "4556"])
|
|
workbook.save(input_dir / "config.xlsx")
|
|
reference_pep = storage_dir / "string_taxon_4555.pep.fasta"
|
|
reference_pep.write_text(">Seita.1\nMA\n", encoding="utf-8")
|
|
(storage_dir / "string_taxon_4555.cds.fasta").write_text(
|
|
">Seita.1\nATGGCT\n", encoding="utf-8"
|
|
)
|
|
query_pep = final_dir / "Sv.HKL.final_members.pep.fa"
|
|
query_pep.write_text(">SvGene1\nMA\n", encoding="utf-8")
|
|
scope_file = final_dir / "Sv.HKL.scope.json"
|
|
scope_file.write_text(json.dumps({
|
|
"family": "HKL", "scope_id": "HKL.target-Sv.ref-At.target_plus_reference",
|
|
"scope_mode": "target_plus_reference", "target_species": ["Sv"],
|
|
"reference_species": ["At"], "selected_species": ["Sv", "At"],
|
|
}), encoding="utf-8")
|
|
launched: list[list[str]] = []
|
|
|
|
def fake_run(cmd, *_args, **_kwargs):
|
|
launched.append(list(cmd))
|
|
run_prefix = cmd[cmd.index("--prefix") + 1]
|
|
pairs = Path(job.workspace_dir) / "ortholog" / f"{run_prefix}_pairs.txt"
|
|
pairs.write_text("Seita.1\tSvGene1\t91\n", encoding="utf-8")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run)
|
|
|
|
message, manifest_path = backend.run_stage(
|
|
"ortholog", job, "Sv.HKL", scope_file=str(scope_file),
|
|
ppi_reference_taxon_id="4555",
|
|
ppi_reference_storage_key="string_taxon_4555",
|
|
ppi_reference_selection_mode="tree_nearest",
|
|
ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi",
|
|
)
|
|
|
|
assert message.startswith("✅")
|
|
assert launched[0][launched[0].index("--subject") + 1] == str(reference_pep)
|
|
assert (final_dir / "Sv.HKL.ppi.orthologs.txt").exists()
|
|
assert not (final_dir / "Sv.HKL.orthologs.txt").exists()
|
|
manifest = json.loads(Path(manifest_path).read_text(encoding="utf-8"))
|
|
assert ".ppi." in Path(manifest_path).name
|
|
assert manifest["parameters"]["dependency_owner"] == "ppi"
|
|
assert manifest["parameters"]["reference_species"] == "string_taxon_4555"
|
|
assert manifest["parameters"]["reference_taxon_id"] == "4555"
|
|
assert manifest["scope_id"] == "HKL.target-Sv.target_only"
|
|
assert manifest["scope_mode"] == "target_only"
|
|
assert manifest["stage_context"]["reference_species"] == []
|
|
assert manifest["stage_context"]["selected_species"] == ["Sv"]
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job, prefix="Sv.HKL", scope_file=str(scope_file),
|
|
target_species=["Sv"], family="HKL",
|
|
ppi_reference_taxon_id="4555", ppi_reference_storage_key="string_taxon_4555",
|
|
ppi_reference_selection_mode="tree_nearest", ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi", requested_stages=["ppi"],
|
|
)
|
|
assert resolution["status"] == "ready"
|
|
assert resolution["mapping_reference_species"] == "string_taxon_4555"
|
|
reference_pep.write_text(">Seita.1\nMAGA\n", encoding="utf-8")
|
|
stale_reference = backend.resolve_ppi_species_resolution(
|
|
job=job, prefix="Sv.HKL", target_species=["Sv"], family="HKL",
|
|
ppi_reference_taxon_id="4555", ppi_reference_storage_key="string_taxon_4555",
|
|
ppi_reference_selection_mode="tree_nearest", ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi", requested_stages=["ppi"],
|
|
)
|
|
assert stale_reference["status"] == "needs_ortholog"
|
|
|
|
reference_pep.write_text(">Seita.1\nMA\n", encoding="utf-8")
|
|
query_pep.write_text(">SvGene1\nMAGA\n", encoding="utf-8")
|
|
stale_target = backend.resolve_ppi_species_resolution(
|
|
job=job, prefix="Sv.HKL", target_species=["Sv"], family="HKL",
|
|
ppi_reference_taxon_id="4555", ppi_reference_storage_key="string_taxon_4555",
|
|
ppi_reference_selection_mode="tree_nearest", ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi", requested_stages=["ppi"],
|
|
)
|
|
assert stale_target["status"] == "needs_ortholog"
|
|
|
|
|
|
def test_ppi_owner_effective_scope_ignores_project_and_technical_references() -> None:
|
|
assert backend._effective_ppi_owner_scope_id(
|
|
family="SPL",
|
|
selected_targets=["Ft"],
|
|
scope_data={
|
|
"scope_id": "SPL.target-Ft.ref-At.target_plus_reference",
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Ft"],
|
|
"reference_species": ["At"],
|
|
"comparative_species": ["Os"],
|
|
},
|
|
reference_storage_key="string_taxon_3562",
|
|
) == "SPL.target-Ft.target_only"
|
|
|
|
|
|
def test_ppi_owned_ortholog_uses_whole_reference_pep_when_family_pep_also_exists(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
job = _job(tmp_path)
|
|
input_dir = Path(job.input_dir)
|
|
final_dir = Path(job.workspace_dir) / "final"
|
|
reference_dir = input_dir / "At"
|
|
reference_dir.mkdir()
|
|
whole_reference_pep = reference_dir / "At.pep.fasta"
|
|
whole_reference_pep.write_text(
|
|
">AT5G16970.1 gene:AT5G16970\nMAGA\n", encoding="utf-8"
|
|
)
|
|
(reference_dir / "At.cds.fasta").write_text(
|
|
">AT5G16970.1 gene:AT5G16970\nATGGCTGGCGCT\n", encoding="utf-8"
|
|
)
|
|
family_reference_pep = input_dir / "At.SPL.pep.fasta"
|
|
family_reference_pep.write_text(">AT5G16970.1\nMA\n", encoding="utf-8")
|
|
query_pep = final_dir / "Ft.SPL.final_members.pep.fa"
|
|
query_pep.write_text(">FtSPL1\nMA\n", encoding="utf-8")
|
|
scope_file = final_dir / "Ft.SPL.scope.json"
|
|
scope_file.write_text(json.dumps({
|
|
"family": "SPL",
|
|
"scope_id": "SPL.target-Ft.target_only",
|
|
"scope_mode": "target_only",
|
|
"target_species": ["Ft"],
|
|
"reference_species": [],
|
|
"comparative_species": [],
|
|
"selected_species": ["Ft"],
|
|
}), encoding="utf-8")
|
|
launched: list[list[str]] = []
|
|
|
|
def fake_run(cmd, *_args, **_kwargs):
|
|
launched.append(list(cmd))
|
|
run_prefix = cmd[cmd.index("--prefix") + 1]
|
|
pairs = Path(job.workspace_dir) / "ortholog" / f"{run_prefix}_pairs.txt"
|
|
pairs.write_text("AT5G16970.1\tFtSPL1\t91\n", encoding="utf-8")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run)
|
|
|
|
message, manifest_path = backend.run_stage(
|
|
"ortholog",
|
|
job,
|
|
"Ft.SPL",
|
|
scope_file=str(scope_file),
|
|
ppi_reference_taxon_id="3702",
|
|
ppi_reference_storage_key="At",
|
|
ppi_reference_selection_mode="model_fallback",
|
|
ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi",
|
|
)
|
|
|
|
assert message.startswith("✅")
|
|
assert launched[0][launched[0].index("--subject") + 1] == str(whole_reference_pep)
|
|
manifest = json.loads(Path(manifest_path).read_text(encoding="utf-8"))
|
|
assert manifest["scope_mode"] == "target_only"
|
|
assert manifest["stage_context"]["reference_species"] == []
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
scope_file=str(scope_file),
|
|
target_species=["Ft"],
|
|
family="SPL",
|
|
ppi_reference_taxon_id="3702",
|
|
ppi_reference_storage_key="At",
|
|
ppi_reference_selection_mode="model_fallback",
|
|
ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi",
|
|
requested_stages=["ppi"],
|
|
)
|
|
assert resolution["status"] == "ready", resolution
|
|
|
|
|
|
def test_independent_ortholog_keeps_family_reference_pep_priority(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
job = _job(tmp_path)
|
|
input_dir = Path(job.input_dir)
|
|
final_dir = Path(job.workspace_dir) / "final"
|
|
reference_dir = input_dir / "At"
|
|
reference_dir.mkdir()
|
|
whole_reference_pep = reference_dir / "At.pep.fasta"
|
|
whole_reference_pep.write_text(">AT5G16970.1\nMAGA\n", encoding="utf-8")
|
|
family_reference_pep = input_dir / "At.SPL.pep.fasta"
|
|
family_reference_pep.write_text(">AT5G16970.1\nMA\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.final_members.pep.fa").write_text(
|
|
">FtSPL1\nMA\n", encoding="utf-8"
|
|
)
|
|
scope_file = final_dir / "Ft.SPL.scope.json"
|
|
scope_file.write_text(json.dumps({
|
|
"family": "SPL",
|
|
"scope_id": "SPL.target-Ft.ref-At.target_plus_reference",
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Ft"],
|
|
"reference_species": ["At"],
|
|
"comparative_species": [],
|
|
"selected_species": ["Ft", "At"],
|
|
}), encoding="utf-8")
|
|
launched: list[list[str]] = []
|
|
|
|
def fake_run(cmd, *_args, **_kwargs):
|
|
launched.append(list(cmd))
|
|
run_prefix = cmd[cmd.index("--prefix") + 1]
|
|
pairs = Path(job.workspace_dir) / "ortholog" / f"{run_prefix}_pairs.txt"
|
|
pairs.write_text("AT5G16970.1\tFtSPL1\t91\n", encoding="utf-8")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run)
|
|
|
|
message, _manifest_path = backend.run_stage(
|
|
"ortholog",
|
|
job,
|
|
"Ft.SPL",
|
|
scope_file=str(scope_file),
|
|
)
|
|
|
|
assert message.startswith("✅")
|
|
assert launched[0][launched[0].index("--subject") + 1] == str(family_reference_pep)
|
|
assert launched[0][launched[0].index("--subject") + 1] != str(whole_reference_pep)
|
|
|
|
|
|
@pytest.mark.parametrize("scope_id_input", ["present", "empty", "missing"])
|
|
def test_ppi_owned_ortholog_preserves_target_only_biological_scope_identity(
|
|
tmp_path: Path, monkeypatch, scope_id_input: str
|
|
) -> None:
|
|
job = _job(tmp_path)
|
|
input_dir = Path(job.input_dir)
|
|
final_dir = Path(job.workspace_dir) / "final"
|
|
storage_dir = input_dir / "string_taxon_4555"
|
|
storage_dir.mkdir()
|
|
workbook = Workbook()
|
|
sheet = workbook.active
|
|
sheet.title = "config"
|
|
sheet.append(["Species_Code", "Species_Name", "TaxID"])
|
|
sheet.append(["Sv", "Setaria viridis", "4556"])
|
|
workbook.save(input_dir / "config.xlsx")
|
|
reference_pep = storage_dir / "string_taxon_4555.pep.fasta"
|
|
reference_pep.write_text(">Seita.1\nMA\n", encoding="utf-8")
|
|
(storage_dir / "string_taxon_4555.cds.fasta").write_text(
|
|
">Seita.1\nATGGCT\n", encoding="utf-8"
|
|
)
|
|
query_pep = final_dir / "Sv.HKL.final_members.pep.fa"
|
|
query_pep.write_text(">SvGene1\nMA\n", encoding="utf-8")
|
|
scope_id = "HKL.target-Sv.target_only"
|
|
scope_file = final_dir / "Sv.HKL.scope.json"
|
|
scope_payload = {
|
|
"family": "HKL",
|
|
"scope_mode": "target_only", "target_species": ["Sv"],
|
|
"reference_species": [], "comparative_species": [],
|
|
"selected_species": ["Sv"],
|
|
}
|
|
if scope_id_input != "missing":
|
|
scope_payload["scope_id"] = scope_id if scope_id_input == "present" else ""
|
|
scope_file.write_text(json.dumps(scope_payload), encoding="utf-8")
|
|
|
|
def fake_run(cmd, *_args, **_kwargs):
|
|
run_prefix = cmd[cmd.index("--prefix") + 1]
|
|
pairs = Path(job.workspace_dir) / "ortholog" / f"{run_prefix}_pairs.txt"
|
|
pairs.write_text("Seita.1\tSvGene1\t91\n", encoding="utf-8")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run)
|
|
|
|
message, manifest_path = backend.run_stage(
|
|
"ortholog", job, "Sv.HKL", scope_file=str(scope_file),
|
|
ppi_reference_taxon_id="4555",
|
|
ppi_reference_storage_key="string_taxon_4555",
|
|
ppi_reference_selection_mode="tree_nearest",
|
|
ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi",
|
|
)
|
|
|
|
assert message.startswith("✅")
|
|
manifest = json.loads(Path(manifest_path).read_text(encoding="utf-8"))
|
|
assert manifest["scope_id"] == scope_id
|
|
assert manifest["scope_mode"] == "target_only"
|
|
assert manifest.get("reference_species", []) == []
|
|
assert manifest["selected_species"] == ["Sv"]
|
|
assert manifest["stage_context"]["scope_id"] == scope_id
|
|
assert manifest["stage_context"]["scope_mode"] == "target_only"
|
|
assert manifest["stage_context"]["reference_species"] == []
|
|
|
|
mapping = final_dir / "Sv.HKL.ppi.orthologs.txt"
|
|
recorded = json.loads(Path(f"{mapping}.identity.json").read_text(encoding="utf-8"))
|
|
expected = build_ortholog_run_identity(
|
|
query_species="Sv", reference_taxon_id="4555",
|
|
reference_storage_key="string_taxon_4555", query_fasta=query_pep,
|
|
reference_pep=reference_pep, scope_id=scope_id, member_set_id="",
|
|
mapping_path=mapping,
|
|
)
|
|
assert recorded["member_identity"] == expected["member_identity"]
|
|
assert recorded["mapping_identity"] == expected["mapping_identity"]
|
|
assert manifest["parameters"]["member_identity"] == expected["member_identity"]
|
|
assert manifest["parameters"]["mapping_identity"] == expected["mapping_identity"]
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job, prefix="Sv.HKL", scope_file=str(scope_file),
|
|
target_species=["Sv"], family="HKL",
|
|
ppi_reference_taxon_id="4555", ppi_reference_storage_key="string_taxon_4555",
|
|
ppi_reference_selection_mode="tree_nearest", ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi", requested_stages=["ppi"],
|
|
)
|
|
assert resolution["status"] == "ready", resolution
|
|
assert resolution["ppi_provenance"]["ortholog_mapping"] == {
|
|
"reference_species": "string_taxon_4555",
|
|
"path": str(mapping),
|
|
}
|
|
|
|
scope_data = json.loads(scope_file.read_text(encoding="utf-8"))
|
|
identity_args = {
|
|
"mapping_file": mapping, "final_dir": final_dir, "in_dir": input_dir,
|
|
"ws": Path(job.workspace_dir), "prefix": "Sv.HKL", "family": "HKL",
|
|
"selected_targets": ["Sv"], "scope_data": scope_data,
|
|
"reference_taxon_id": "4555", "reference_storage_key": "string_taxon_4555",
|
|
"reference_pep": reference_pep,
|
|
}
|
|
for changed in (
|
|
{"reference_taxon_id": "3702"},
|
|
{"reference_storage_key": "string_taxon_3702"},
|
|
):
|
|
matches, reason = backend._ppi_mapping_identity_matches_current_inputs(
|
|
**{**identity_args, **changed}
|
|
)
|
|
assert matches is False
|
|
assert reason == "ppi_owner_mapping_sidecar_identity_mismatch"
|
|
|
|
reference_pep.write_text(">Seita.1\nMAGA\n", encoding="utf-8")
|
|
matches, reason = backend._ppi_mapping_identity_matches_current_inputs(**identity_args)
|
|
assert matches is False
|
|
assert reason == "ppi_owner_mapping_sidecar_identity_mismatch"
|
|
|
|
|
|
def test_ortholog_identity_sidecar_records_complete_identity_and_requires_exact_match(tmp_path: Path) -> None:
|
|
query = tmp_path / "Sv.pep.fa"
|
|
reference = tmp_path / "string_taxon_4555.pep.fasta"
|
|
pairs = tmp_path / "Sv_to_string_taxon_4555_pairs.txt"
|
|
query.write_text(">Sv1\nMA\n", encoding="utf-8")
|
|
reference.write_text(">Seita1\nMA\n", encoding="utf-8")
|
|
pairs.write_text("Seita1\tSv1\t90\n", encoding="utf-8")
|
|
identity = build_ortholog_run_identity(
|
|
query_species="Sv", reference_taxon_id="4555",
|
|
reference_storage_key="string_taxon_4555", query_fasta=query,
|
|
reference_pep=reference, scope_id="scope-v1", member_set_id="members-v1",
|
|
mapping_path=pairs,
|
|
)
|
|
sidecar = write_ortholog_run_sidecar(pairs, identity)
|
|
data = json.loads(sidecar.read_text(encoding="utf-8"))
|
|
|
|
for key in (
|
|
"reference_taxon_id", "reference_storage_key", "reference_pep_sha256",
|
|
"target_identity", "member_identity", "mapping_identity", "mapping_path",
|
|
):
|
|
assert data[key]
|
|
assert ortholog_run_identity_matches(pairs, identity) is True
|
|
missing_sidecar_pairs = tmp_path / "missing_pairs.txt"
|
|
missing_sidecar_pairs.write_text(pairs.read_text(encoding="utf-8"), encoding="utf-8")
|
|
assert ortholog_run_identity_matches(missing_sidecar_pairs, identity) is False
|
|
conflict = dict(identity, reference_taxon_id="3702")
|
|
assert ortholog_run_identity_matches(pairs, conflict) is False
|
|
|
|
|
|
def test_materialized_manifest_persists_owner_and_mapping_identity(tmp_path: Path) -> None:
|
|
final_dir = tmp_path / "final"
|
|
query = tmp_path / "Sv.pep.fa"
|
|
reference = tmp_path / "string_taxon_4555.pep.fasta"
|
|
pairs = tmp_path / "pairs.txt"
|
|
query.write_text(">Sv1\nMA\n", encoding="utf-8")
|
|
reference.write_text(">Seita1\nMA\n", encoding="utf-8")
|
|
pairs.write_text("Seita1\tSv1\t90\n", encoding="utf-8")
|
|
identity = build_ortholog_run_identity(
|
|
query_species="Sv", reference_taxon_id="4555",
|
|
reference_storage_key="string_taxon_4555", query_fasta=query,
|
|
reference_pep=reference, scope_id="scope-v1", member_set_id="members-v1",
|
|
mapping_path=pairs,
|
|
)
|
|
count, manifest_path = materialize_ortholog_final_outputs(
|
|
final_dir=final_dir, prefix="Sv.HKL",
|
|
run_records=[OrthologRunRecord("Sv", "string_taxon_4555", query, reference, pairs)],
|
|
job_id="task5b", workspace=tmp_path, reference_species="string_taxon_4555",
|
|
reference_taxon_id="4555", reference_storage_key="string_taxon_4555",
|
|
ortholog_identity=identity, member_set_id="members-v1",
|
|
scope_fields={"scope_id": "scope-v1", "target_species": ["Sv"], "reference_species": ["string_taxon_4555"]},
|
|
)
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
|
|
assert count == 1
|
|
for key in (
|
|
"reference_taxon_id", "reference_storage_key", "reference_pep_sha256",
|
|
"target_identity", "member_identity", "mapping_identity", "mapping_path",
|
|
):
|
|
assert manifest["parameters"][key]
|
|
|
|
|
|
def test_ppi_and_kaks_owned_mappings_coexist_without_overwrite(tmp_path: Path) -> None:
|
|
final_dir = tmp_path / "final"
|
|
query = tmp_path / "Ft.pep.fa"
|
|
query.write_text(">Ft1\nMA\n", encoding="utf-8")
|
|
|
|
def materialize(owner: str, reference_species: str, reference_id: str) -> tuple[Path, Path]:
|
|
reference = tmp_path / f"{reference_species}.pep.fasta"
|
|
pairs = tmp_path / f"{owner}.pairs.txt"
|
|
reference.write_text(f">{reference_id}\nMA\n", encoding="utf-8")
|
|
pairs.write_text(f"{reference_id}\tFt1\t90\n", encoding="utf-8")
|
|
_, manifest_path = materialize_ortholog_final_outputs(
|
|
final_dir=final_dir,
|
|
prefix="Ft.SPL",
|
|
dependency_owner=owner,
|
|
run_records=[OrthologRunRecord("Ft", reference_species, query, reference, pairs)],
|
|
job_id="task5b",
|
|
workspace=tmp_path,
|
|
reference_species=reference_species,
|
|
reference_storage_key=reference_species,
|
|
scope_fields={
|
|
"scope_id": f"SPL.target-Ft.ref-{reference_species}.target_plus_reference",
|
|
"target_species": ["Ft"],
|
|
"reference_species": [reference_species],
|
|
},
|
|
)
|
|
return final_dir / f"Ft.SPL.{owner}.orthologs.txt", manifest_path
|
|
|
|
ppi_mapping, ppi_manifest = materialize("ppi", "string_taxon_3562", "Soe1")
|
|
ppi_mapping_before = ppi_mapping.read_bytes()
|
|
ppi_manifest_before = ppi_manifest.read_bytes()
|
|
|
|
kaks_mapping, kaks_manifest = materialize("kaks", "At", "AT1G01010")
|
|
|
|
assert ppi_mapping.exists()
|
|
assert kaks_mapping.exists()
|
|
assert ppi_mapping != kaks_mapping
|
|
assert ppi_mapping.read_bytes() == ppi_mapping_before
|
|
assert ppi_manifest.read_bytes() == ppi_manifest_before
|
|
assert json.loads(ppi_manifest.read_text(encoding="utf-8"))["parameters"]["dependency_owner"] == "ppi"
|
|
assert json.loads(kaks_manifest.read_text(encoding="utf-8"))["parameters"]["dependency_owner"] == "kaks"
|
|
|
|
|
|
def test_owner_specific_mapping_candidates_never_fall_back_to_another_owner(tmp_path: Path) -> None:
|
|
final_dir = tmp_path / "final"
|
|
final_dir.mkdir()
|
|
|
|
ppi_candidates = backend.ortholog_mapping_candidates(
|
|
final_dir,
|
|
"Ft.SPL",
|
|
dependency_owner="ppi",
|
|
scope_id="scope-v1",
|
|
)
|
|
|
|
assert ppi_candidates == [
|
|
final_dir / "Ft.SPL.ppi.scope-v1.orthologs.txt",
|
|
final_dir / "Ft.SPL.ppi.scope-v1.ortholog_pairs.txt",
|
|
final_dir / "Ft.SPL.ppi.orthologs.txt",
|
|
final_dir / "Ft.SPL.ppi.ortholog_pairs.txt",
|
|
]
|
|
assert all(".kaks." not in path.name for path in ppi_candidates)
|
|
assert all(path.name != "Ft.SPL.orthologs.txt" for path in ppi_candidates)
|
|
|
|
|
|
def test_kaks_selected_ppi_owner_consumes_only_ppi_mapping(tmp_path: Path, monkeypatch) -> None:
|
|
job = _job(tmp_path)
|
|
input_dir = Path(job.input_dir)
|
|
final_dir = Path(job.workspace_dir) / "final"
|
|
reference = "string_taxon_3562"
|
|
scope_file = final_dir / "Ft.SPL.scope.json"
|
|
scope_file.write_text(json.dumps({
|
|
"family": "SPL",
|
|
"scope_id": f"SPL.target-Ft.ref-{reference}.target_plus_reference",
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Ft"],
|
|
"reference_species": [reference],
|
|
"selected_species": ["Ft", reference],
|
|
}), encoding="utf-8")
|
|
(input_dir / "Ft.cds.fasta").write_text(">Ft1\nATGGCT\n", encoding="utf-8")
|
|
(input_dir / f"{reference}.cds.fasta").write_text(">PpiGene\nATGGCT\n", encoding="utf-8")
|
|
(input_dir / f"{reference}.pep.fasta").write_text(">PpiGene\nMA\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.final_members.pep.fa").write_text(">Ft1\nMA\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.ppi.orthologs.txt").write_text(
|
|
f"Ft1\tPpiGene\t91\tFt\t{reference}\n", encoding="utf-8"
|
|
)
|
|
(final_dir / "Ft.SPL.kaks.orthologs.txt").write_text(
|
|
f"Ft1\tWrongKaksGene\t88\tFt\t{reference}\n", encoding="utf-8"
|
|
)
|
|
observed_pairs: list[str] = []
|
|
|
|
def fake_run(cmd, *_args, **_kwargs):
|
|
pair_path = Path(cmd[cmd.index("--pairs") + 1])
|
|
observed_pairs.append(pair_path.read_text(encoding="utf-8"))
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run)
|
|
|
|
message, manifest_path = backend.run_stage(
|
|
"kaks",
|
|
job,
|
|
"Ft.SPL",
|
|
scope_file=str(scope_file),
|
|
ortholog_mapping_owner="ppi",
|
|
)
|
|
|
|
assert message.startswith("✅")
|
|
assert observed_pairs and "PpiGene" in observed_pairs[0]
|
|
assert "WrongKaksGene" not in observed_pairs[0]
|
|
manifest = json.loads(Path(manifest_path).read_text(encoding="utf-8"))
|
|
assert manifest["parameters"]["ortholog_mapping"].endswith("Ft.SPL.ppi.orthologs.txt")
|