1801 lines
67 KiB
Python
Executable File
1801 lines
67 KiB
Python
Executable File
import csv
|
|
import os
|
|
import json
|
|
import sys
|
|
import types
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
from apps.web import pipeline_backend as backend
|
|
from gfa_core.gene_family_scope import build_scope_id
|
|
from gfa_core.ortholog_materialize import (
|
|
OrthologRunRecord,
|
|
build_ortholog_run_identity,
|
|
materialize_ortholog_final_outputs,
|
|
)
|
|
|
|
|
|
def _materialize_trusted_ppi_mapping(
|
|
job, *, prefix, target, family, storage_key, taxon_id, query_id, subject_id,
|
|
):
|
|
workspace = Path(job.workspace_dir)
|
|
final_dir = workspace / "final"
|
|
query = final_dir / f"{prefix}.final_members.pep.fa"
|
|
reference = Path(job.input_dir) / storage_key / f"{storage_key}.pep.fasta"
|
|
query.write_text(f">{query_id}\nMA\n", encoding="utf-8")
|
|
pairs = workspace / "ortholog" / f"{prefix}.{target}_to_{storage_key}_pairs.txt"
|
|
pairs.parent.mkdir(parents=True, exist_ok=True)
|
|
pairs.write_text(f"{subject_id}\t{query_id}\t91\n", encoding="utf-8")
|
|
scope_id = build_scope_id(family, [target], [], [], "target_only")
|
|
identity = build_ortholog_run_identity(
|
|
query_species=target, reference_taxon_id=taxon_id,
|
|
reference_storage_key=storage_key, query_fasta=query,
|
|
reference_pep=reference, scope_id=scope_id, member_set_id="",
|
|
mapping_path=pairs,
|
|
)
|
|
materialize_ortholog_final_outputs(
|
|
final_dir=final_dir, prefix=prefix,
|
|
dependency_owner="ppi",
|
|
run_records=[OrthologRunRecord(target, storage_key, query, reference, pairs)],
|
|
job_id=job.job_id, workspace=workspace, reference_species=storage_key,
|
|
reference_taxon_id=taxon_id, reference_storage_key=storage_key,
|
|
ortholog_identity=identity,
|
|
scope_fields={
|
|
"scope_id": scope_id, "scope_mode": "target_only",
|
|
"target_species": [target], "reference_species": [],
|
|
"selected_species": [target], "family": family,
|
|
},
|
|
)
|
|
|
|
|
|
def test_run_cmd_logged_exports_project_root_to_pythonpath(tmp_path, monkeypatch):
|
|
captured = {}
|
|
|
|
class FakeProcess:
|
|
def wait(self):
|
|
return 0
|
|
|
|
def fake_popen(cmd, cwd=None, stdout=None, stderr=None, text=None, env=None):
|
|
captured["env"] = env or {}
|
|
return FakeProcess()
|
|
|
|
monkeypatch.setenv("PYTHONPATH", "/existing/path")
|
|
monkeypatch.setattr(backend.subprocess, "Popen", fake_popen)
|
|
|
|
exit_code = backend.run_cmd_logged(
|
|
["python", "skills/ppi/run.py"],
|
|
tmp_path / "ppi.log",
|
|
cwd=backend.PROJECT_ROOT,
|
|
)
|
|
|
|
pythonpath = captured["env"].get("PYTHONPATH", "").split(os.pathsep)
|
|
assert exit_code == 0
|
|
assert str(backend.PROJECT_ROOT) in pythonpath
|
|
assert "/existing/path" in pythonpath
|
|
|
|
|
|
def _write_ppi_scope(path, target_species="Os", reference_species="At"):
|
|
data = {
|
|
"family": "CRR",
|
|
"prefix": "Os.CRR",
|
|
"scope_id": "CRR.target-Os.ref-At.target_plus_reference",
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": [target_species],
|
|
"reference_species": [reference_species],
|
|
"selected_species": [target_species, reference_species],
|
|
}
|
|
path.write_text(json.dumps(data), encoding="utf-8")
|
|
|
|
|
|
def _write_fake_ppi_outputs(ppi_dir, prefix="Os.CRR"):
|
|
ppi_dir.mkdir(parents=True, exist_ok=True)
|
|
(ppi_dir / f"{prefix}_network.tsv").write_text("protein1\tprotein2\tscore\nA\tB\t0.9\n", encoding="utf-8")
|
|
(ppi_dir / f"{prefix}_report.md").write_text("# PPI\n", encoding="utf-8")
|
|
(ppi_dir / f"{prefix}_url.txt").write_text("View online: https://string-db.org/cgi/network?species=39947\n", encoding="utf-8")
|
|
(ppi_dir / f"{prefix}_id_resolution.tsv").write_text("raw_id\tnormalized_id\nOs01t0810800-01\tOs01g0810800\n", encoding="utf-8")
|
|
(ppi_dir / f"{prefix}.manifest.json").write_text(
|
|
json.dumps({"manifest_version": "1.0", "stage": "ppi", "status": "success", "prefix": prefix}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def test_reference_bundle_id_map_preserves_loc_and_numeric_geneid(tmp_path):
|
|
pep_path = tmp_path / "soe.pep.fasta"
|
|
cds_path = tmp_path / "soe.cds.fasta"
|
|
output_path = tmp_path / "soe.pep_cds_idmap.tsv"
|
|
pep_path.write_text(
|
|
">XP_021867507.1 [protein_id=XP_021867507.1]\nMAAA\n",
|
|
encoding="utf-8",
|
|
)
|
|
cds_path.write_text(
|
|
">XP_021867507.1 [gene=LOC110806180] "
|
|
"[db_xref=GeneID:110806180] [protein_id=XP_021867507.1]\nATGGCG\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = backend._build_reference_bundle_id_map(
|
|
species_code="soe",
|
|
pep_path=pep_path,
|
|
cds_path=cds_path,
|
|
output_path=output_path,
|
|
)
|
|
|
|
with open(result["id_map_path"], encoding="utf-8", newline="") as handle:
|
|
row = next(csv.DictReader(handle, delimiter="\t"))
|
|
assert row["gene_id"] == "LOC110806180"
|
|
assert row["ncbi_gene_id"] == "110806180"
|
|
|
|
|
|
def test_reference_bundle_id_map_reads_ensembl_gene_token_for_isoforms(tmp_path):
|
|
from gfa_core.id_resolution import ReferenceAliasIndex
|
|
|
|
pep_path = tmp_path / "At.pep.fasta"
|
|
cds_path = tmp_path / "At.cds.fasta"
|
|
output_path = tmp_path / "At.pep_cds_idmap.tsv"
|
|
pep_path.write_text(
|
|
">AT1G02065.1 pep chromosome:TAIR10:1:365209:367443:1 "
|
|
"gene:AT1G02065 transcript:AT1G02065.1 gene_symbol:SPL8\nMAAA\n"
|
|
">AT1G02065.2 pep chromosome:TAIR10:1:365165:367353:1 "
|
|
"gene:AT1G02065 transcript:AT1G02065.2 gene_symbol:SPL8\nMBBB\n",
|
|
encoding="utf-8",
|
|
)
|
|
cds_path.write_text(
|
|
">AT1G02065.1 cds chromosome:TAIR10:1:365209:367443:1 "
|
|
"gene:AT1G02065 transcript:AT1G02065.1 gene_symbol:SPL8\nATGGCG\n"
|
|
">AT1G02065.2 cds chromosome:TAIR10:1:365165:367353:1 "
|
|
"gene:AT1G02065 transcript:AT1G02065.2 gene_symbol:SPL8\nATGGCC\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
backend._build_reference_bundle_id_map(
|
|
species_code="At",
|
|
pep_path=pep_path,
|
|
cds_path=cds_path,
|
|
output_path=output_path,
|
|
)
|
|
|
|
with output_path.open(encoding="utf-8", newline="") as handle:
|
|
rows = list(csv.DictReader(handle, delimiter="\t"))
|
|
assert {row["protein_id"]: row["gene_id"] for row in rows} == {
|
|
"AT1G02065.1": "AT1G02065",
|
|
"AT1G02065.2": "AT1G02065",
|
|
}
|
|
|
|
candidates = ReferenceAliasIndex.from_tsv(output_path, "At").candidates("AT1G02065")
|
|
assert {candidate.value for candidate in candidates} >= {
|
|
"AT1G02065",
|
|
"AT1G02065.1",
|
|
"AT1G02065.2",
|
|
}
|
|
|
|
|
|
def test_reference_bundle_download_extracts_ncbi_zip_and_unblocks_ppi_kaks_preflight(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test_ref_bundle",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
(workspace / "final").mkdir(parents=True)
|
|
(workspace / "logs").mkdir(parents=True)
|
|
input_dir.mkdir(parents=True)
|
|
|
|
fake_zip = tmp_path / "ncbi_dataset.zip"
|
|
with zipfile.ZipFile(fake_zip, "w") as zf:
|
|
zf.writestr("ncbi_dataset/data/GCF_FAKE/protein.faa", ">SOE_PEP1\nMAAA\n")
|
|
zf.writestr(
|
|
"ncbi_dataset/data/GCF_FAKE/cds_from_genomic.fna",
|
|
">lcl|NC_000001.1_cds_XP_000001.1_1 [protein_id=XP_000001.1]\nATGGCG\n",
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
backend,
|
|
"_select_ncbi_reference_assembly",
|
|
lambda **kwargs: {
|
|
"accession": "GCF_FAKE",
|
|
"organism_name": "Spinacia oleracea",
|
|
"taxon_id": "3562",
|
|
"assembly_name": "fake",
|
|
"assembly_level": "Chromosome",
|
|
"annotation_name": "fake annotation",
|
|
},
|
|
)
|
|
|
|
def fake_download(url, output_path, timeout_sec=300):
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_bytes(fake_zip.read_bytes())
|
|
return output_path.stat().st_size
|
|
|
|
monkeypatch.setattr(backend, "_download_url_to_file", fake_download)
|
|
|
|
manifest = backend.download_reference_annotation_bundle(
|
|
job=job,
|
|
species_code="soe",
|
|
species_name="Spinacia oleracea",
|
|
taxon_id="3562",
|
|
requested_stages=["ppi", "kaks"],
|
|
)
|
|
|
|
assert manifest["status"] == "success"
|
|
assert (input_dir / "soe" / "soe.pep.fasta").exists()
|
|
assert (input_dir / "soe" / "soe.cds.fasta").exists()
|
|
assert (input_dir / "soe" / "soe.cds.fasta").read_text(encoding="utf-8").startswith(">XP_000001.1 ")
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
target_species=["Ft"],
|
|
reference_species=["Spinacia oleracea"],
|
|
family="SPL",
|
|
requested_stages=["ppi", "kaks"],
|
|
)
|
|
assert resolution["status"] == "needs_ortholog"
|
|
assert resolution["selected_string_species"] == "soe"
|
|
assert resolution["reference_species"] == "soe"
|
|
assert resolution["reference_bundle_status"]["kaks_ready"] is True
|
|
assert resolution["reference_bundle_status"]["missing_required"] == []
|
|
|
|
|
|
def test_ortholog_scope_uses_downloaded_reference_proteome_when_family_pep_missing(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test_ortholog_ref_bundle",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
final_dir = workspace / "final"
|
|
logs_dir = workspace / "logs"
|
|
(input_dir / "soe").mkdir(parents=True)
|
|
final_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
(final_dir / "Ft.SPL.identify.manifest.json").write_text(
|
|
json.dumps({"stage": "identify", "status": "success", "prefix": "Ft.SPL"}),
|
|
encoding="utf-8",
|
|
)
|
|
(final_dir / "Ft.SPL.final_members.pep.fa").write_text(">FtGene1\nMAAA\n", encoding="utf-8")
|
|
(input_dir / "soe" / "soe.pep.fasta").write_text(">SoeGene1\nMAAA\n", encoding="utf-8")
|
|
scope_file = final_dir / "Ft.SPL.SPL.target-Ft.ref-soe.target_plus_reference.scope.json"
|
|
scope_file.write_text(
|
|
json.dumps({
|
|
"family": "SPL",
|
|
"prefix": "Ft.SPL",
|
|
"scope_id": "SPL.target-Ft.ref-soe.target_plus_reference",
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Ft"],
|
|
"reference_species": ["Spinacia oleracea"],
|
|
"selected_species": ["Ft", "Spinacia oleracea"],
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
captured = {}
|
|
|
|
def fake_run_cmd_logged(cmd, log_path, cwd=None):
|
|
captured["cmd"] = cmd
|
|
output_dir = tmp_path / "workspace" / "ortholog"
|
|
run_prefix = "Ft.SPL.Ft_to_soe"
|
|
(output_dir / f"{run_prefix}_pairs.txt").write_text(
|
|
"# Ortholog pairs identified by BLASTP\n"
|
|
"# Format: Subject_ID\tQuery_ID\tIdentity(%)\n"
|
|
"SoeGene1\tFtGene1\t91.00\n",
|
|
encoding="utf-8",
|
|
)
|
|
(output_dir / f"{run_prefix}.blastp.txt").write_text("FtGene1\tSoeGene1\t91\n", encoding="utf-8")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run_cmd_logged)
|
|
|
|
message, manifest = backend.run_stage(
|
|
"ortholog",
|
|
job,
|
|
"Ft.SPL",
|
|
cpu=2,
|
|
scope_file=str(scope_file),
|
|
blast_evalue=1e-5,
|
|
)
|
|
|
|
assert message.startswith("✅ Ortholog analysis complete")
|
|
assert manifest is not None
|
|
assert str(input_dir / "soe" / "soe.pep.fasta") in captured["cmd"]
|
|
|
|
|
|
def test_ppi_auto_prefers_target_string_species_when_target_is_supported(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
final_dir = workspace / "final"
|
|
logs_dir = workspace / "logs"
|
|
final_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
input_dir.mkdir(parents=True)
|
|
(final_dir / "Os.CRR.final_members.id").write_text("Os01t0810800-01\n", encoding="utf-8")
|
|
stale_image = final_dir / "Os.CRR.ppi_network.string.png"
|
|
stale_image.write_bytes(b"old-at-image")
|
|
(final_dir / "Os.CRR.orthologs.txt").write_text(
|
|
"Os01t0810800-01\tAT1G51790\t90\tOs\tAt\n",
|
|
encoding="utf-8",
|
|
)
|
|
scope_file = final_dir / "Os.CRR.CRR.target-Os.ref-At.target_plus_reference.scope.json"
|
|
_write_ppi_scope(scope_file)
|
|
|
|
captured = {}
|
|
|
|
def fake_run_cmd_logged(cmd, log_path, cwd=None):
|
|
captured["cmd"] = cmd
|
|
_write_fake_ppi_outputs(workspace / "ppi")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda *args, **kwargs: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run_cmd_logged)
|
|
monkeypatch.setattr(backend, "download_string_network_image", lambda *args, **kwargs: None)
|
|
|
|
message, manifest = backend.run_stage(
|
|
"ppi",
|
|
job,
|
|
"Os.CRR",
|
|
ppi_species="auto",
|
|
ppi_score=400,
|
|
scope_file=str(scope_file),
|
|
analysis_route="target_plus_reference",
|
|
)
|
|
|
|
cmd = captured["cmd"]
|
|
assert "✅ PPI analysis completed" in message
|
|
assert manifest and manifest.exists()
|
|
assert cmd[cmd.index("--species") + 1] == "osa"
|
|
assert cmd[cmd.index("--gene-ids") + 1] == str(final_dir / "Os.CRR.final_members.id")
|
|
assert "--mapping" not in cmd
|
|
assert "--id-map" not in cmd
|
|
assert not stale_image.exists()
|
|
|
|
|
|
def test_ppi_direct_supported_target_forwards_exact_current_job_id_map(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
final_dir = workspace / "final"
|
|
(input_dir / "osa").mkdir(parents=True)
|
|
final_dir.mkdir(parents=True)
|
|
(workspace / "logs").mkdir(parents=True)
|
|
id_map_file = input_dir / "osa" / "osa.pep_cds_idmap.tsv"
|
|
id_map_file.write_text("species_code\tprotein_id\nosa\tOs01g0810800\n", encoding="utf-8")
|
|
(final_dir / "Os.CRR.final_members.id").write_text("Os01t0810800-01\n", encoding="utf-8")
|
|
scope_file = final_dir / "Os.CRR.CRR.target-Os.ref-At.target_plus_reference.scope.json"
|
|
_write_ppi_scope(scope_file)
|
|
captured = {}
|
|
|
|
def fake_run_cmd_logged(cmd, log_path, cwd=None):
|
|
captured["cmd"] = cmd
|
|
_write_fake_ppi_outputs(workspace / "ppi")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda *args, **kwargs: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run_cmd_logged)
|
|
monkeypatch.setattr(backend, "download_string_network_image", lambda *args, **kwargs: None)
|
|
|
|
message, manifest = backend.run_stage(
|
|
"ppi",
|
|
job,
|
|
"Os.CRR",
|
|
ppi_species="osa",
|
|
ppi_score=400,
|
|
scope_file=str(scope_file),
|
|
analysis_route="target_plus_reference",
|
|
)
|
|
|
|
cmd = captured["cmd"]
|
|
assert "✅ PPI analysis completed" in message
|
|
assert manifest and manifest.exists()
|
|
assert cmd[cmd.index("--id-map") + 1] == str(id_map_file)
|
|
assert "--mapping" not in cmd
|
|
|
|
|
|
def test_infer_string_species_from_ortholog_mapping_does_not_default_to_ath(tmp_path):
|
|
mapping = tmp_path / "Ft.unknown.orthologs.txt"
|
|
mapping.write_text("FtGene1\tRefGene1\t90\tFt\tUnknownSpecies\n", encoding="utf-8")
|
|
|
|
assert backend.infer_string_species_from_ortholog_mapping(mapping, "") == ""
|
|
|
|
|
|
def test_ppi_scope_with_mapping_but_unknown_reference_species_fails_instead_of_ath(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
final_dir = workspace / "final"
|
|
logs_dir = workspace / "logs"
|
|
final_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
input_dir.mkdir(parents=True)
|
|
(final_dir / "Ft.SPL.final_members.id").write_text("FtGene1\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.orthologs.txt").write_text(
|
|
"FtGene1\tRefGene1\t90\tFt\tUnknownSpecies\n",
|
|
encoding="utf-8",
|
|
)
|
|
scope = {
|
|
"family": "SPL",
|
|
"prefix": "Ft.SPL",
|
|
"scope_id": "SPL.target-Ft.ref-UnknownSpecies.target_plus_reference",
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Ft"],
|
|
"reference_species": ["UnknownSpecies"],
|
|
"selected_species": ["Ft", "UnknownSpecies"],
|
|
}
|
|
scope_file = final_dir / "Ft.SPL.scope.json"
|
|
scope_file.write_text(json.dumps(scope), encoding="utf-8")
|
|
|
|
called = {"run": False}
|
|
|
|
def fake_run_cmd_logged(*args, **kwargs):
|
|
called["run"] = True
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run_cmd_logged)
|
|
|
|
message, manifest = backend.run_stage(
|
|
"ppi",
|
|
job,
|
|
"Ft.SPL",
|
|
ppi_species="auto",
|
|
ppi_score=200,
|
|
scope_file=str(scope_file),
|
|
analysis_route="target_plus_reference",
|
|
)
|
|
|
|
assert called["run"] is False
|
|
assert "无法从 Ortholog 映射推断 STRING 参考物种" in message
|
|
assert manifest and manifest.exists()
|
|
data = json.loads(manifest.read_text(encoding="utf-8"))
|
|
assert data["diagnostics"]["issue_code"] == "ppi_invalid_ortholog_mapping"
|
|
assert data["parameters"]["ppi_species_resolution"]["decision"] == "failed_requires_ortholog"
|
|
|
|
|
|
def test_ppi_species_resolution_reports_reference_data_missing(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
(tmp_path / "input").mkdir()
|
|
(tmp_path / "workspace" / "final").mkdir(parents=True)
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Ft"],
|
|
reference_species=[],
|
|
family="SPL",
|
|
)
|
|
|
|
assert resolution["status"] == "confirmation_required"
|
|
assert resolution["issue_code"] == "ppi_reference_confirmation_required"
|
|
assert resolution["required_next_stage"] == ""
|
|
assert [row["selected_string_taxon_id"] for row in resolution["candidates"]] == ["3562", "3476", "3775"]
|
|
|
|
|
|
def test_os_direct_ppi_uses_complete_string_registry_without_ortholog(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test_os_direct_ppi",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
(tmp_path / "input").mkdir()
|
|
(tmp_path / "workspace" / "final").mkdir(parents=True)
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Os.HKL",
|
|
target_species=["Os"],
|
|
requested_stages=["ppi"],
|
|
)
|
|
|
|
assert resolution["status"] == "ready"
|
|
assert resolution["selection_mode"] == "exact_string"
|
|
assert resolution["selected_taxon_id"] == "39947"
|
|
assert resolution["required_next_stage"] == ""
|
|
assert resolution["ortholog_mapping_source"] == ""
|
|
|
|
|
|
def test_ppi_species_resolution_canonicalizes_fagopyrum_tataricum_alias(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
(tmp_path / "input").mkdir()
|
|
(tmp_path / "workspace" / "final").mkdir(parents=True)
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=[],
|
|
family="SPL",
|
|
)
|
|
|
|
assert resolution["target_species"] == ["Ft"]
|
|
assert resolution["target_string_supported"] is False
|
|
assert resolution["status"] == "confirmation_required"
|
|
|
|
|
|
def test_ppi_species_resolution_recommends_near_string_species_before_arabidopsis_for_fagopyrum(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
(tmp_path / "input").mkdir()
|
|
(tmp_path / "workspace" / "final").mkdir(parents=True)
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=[],
|
|
family="SPL",
|
|
)
|
|
|
|
candidates = resolution["diagnostics"]["reference_recommendations"]
|
|
assert resolution["status"] == "confirmation_required"
|
|
assert [row["selected_string_taxon_id"] for row in candidates] == ["3562", "3476", "3775"]
|
|
assert candidates[0]["selected_string_name"] == "Spinacia oleracea"
|
|
assert candidates[0]["lca_taxon_id"] == "3524"
|
|
assert candidates[0]["tree_distance"] == 10
|
|
|
|
|
|
def test_explicit_project_reference_is_not_overridden_by_downloaded_near_reference(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
(tmp_path / "workspace" / "final").mkdir(parents=True)
|
|
(input_dir / "At").mkdir(parents=True)
|
|
(input_dir / "soe").mkdir(parents=True)
|
|
(input_dir / "At" / "At.pep.fasta").write_text(">AT1\nMAAA\n", encoding="utf-8")
|
|
(input_dir / "At" / "At.cds.fasta").write_text(">AT1\nATGGCTGCTGCT\n", encoding="utf-8")
|
|
(input_dir / "soe" / "soe.pep.fasta").write_text(">SOE1\nMAAA\n", encoding="utf-8")
|
|
(input_dir / "soe" / "soe.cds.fasta").write_text(">SOE1\nATGGCTGCTGCT\n", encoding="utf-8")
|
|
(input_dir / "soe" / "soe.reference_bundle_download.manifest.json").write_text(
|
|
json.dumps({"status": "success", "species_code": "soe"}), encoding="utf-8"
|
|
)
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="ath",
|
|
target_species=["Ft"],
|
|
reference_species=["At"],
|
|
family="SPL",
|
|
requested_stages=["kaks"],
|
|
)
|
|
|
|
assert resolution["reference_species"] == "At"
|
|
assert resolution["selected_string_species"] == "ath"
|
|
assert resolution["reference_bundle_status"]["kaks_ready"] is True
|
|
|
|
|
|
def test_ppi_species_resolution_selected_near_string_species_still_needs_reference_data_without_pep(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
(tmp_path / "input").mkdir()
|
|
(tmp_path / "workspace" / "final").mkdir(parents=True)
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=["Spinacia oleracea"],
|
|
family="SPL",
|
|
ppi_reference_taxon_id="3562",
|
|
ppi_reference_storage_key="soe",
|
|
ppi_reference_selection_mode="tree_nearest",
|
|
ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi",
|
|
)
|
|
|
|
assert resolution["status"] == "needs_reference_data"
|
|
assert resolution["issue_code"] == "ppi_reference_data_missing"
|
|
assert resolution["selected_string_species"] == "soe"
|
|
assert resolution["selected_taxon_id"] == "3562"
|
|
assert resolution["reference_species"] == "soe"
|
|
assert resolution["reference_data_source"] == ""
|
|
|
|
|
|
def test_ppi_species_resolution_selected_near_string_species_with_complete_bundle_needs_ortholog(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
input_dir.mkdir()
|
|
final_dir.mkdir(parents=True)
|
|
(input_dir / "soe").mkdir()
|
|
(input_dir / "soe" / "soe.pep.fasta").write_text(">SoeGene1\nMA\n", encoding="utf-8")
|
|
(input_dir / "soe" / "soe.cds.fasta").write_text(">SoeGene1\nATG\n", encoding="utf-8")
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=["Spinacia oleracea"],
|
|
family="SPL",
|
|
ppi_reference_taxon_id="3562",
|
|
ppi_reference_storage_key="soe",
|
|
ppi_reference_selection_mode="tree_nearest",
|
|
ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi",
|
|
)
|
|
|
|
assert resolution["status"] == "needs_ortholog"
|
|
assert resolution["issue_code"] == "ppi_requires_ortholog_mapping"
|
|
assert resolution["selected_string_species"] == "soe"
|
|
assert resolution["selected_taxon_id"] == "3562"
|
|
assert resolution["reference_data_source"].endswith("soe/soe.pep.fasta")
|
|
assert resolution["reference_bundle_status"]["kaks_ready"] is True
|
|
|
|
|
|
def test_supported_target_ppi_is_ready_independently_when_kaks_lacks_reference_data(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
(tmp_path / "input").mkdir()
|
|
(tmp_path / "workspace" / "final").mkdir(parents=True)
|
|
(tmp_path / "input" / "soe").mkdir()
|
|
(tmp_path / "input" / "soe" / "soe.pep.fasta").write_text(">SoeGene1\nMA\n", encoding="utf-8")
|
|
|
|
ppi_resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Os.CRR",
|
|
ppi_species="auto",
|
|
target_species=["Os"],
|
|
reference_species=["At"],
|
|
family="CRR",
|
|
requested_stages=["ppi"],
|
|
)
|
|
kaks_resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Os.CRR",
|
|
ppi_species="auto",
|
|
target_species=["Os"],
|
|
reference_species=["Spinacia oleracea"],
|
|
family="CRR",
|
|
requested_stages=["kaks"],
|
|
)
|
|
|
|
assert ppi_resolution["status"] == "ready"
|
|
assert ppi_resolution["decision"] == "ready_target_string"
|
|
assert ppi_resolution["selected_string_species"] == "osa"
|
|
assert ppi_resolution["ortholog_mapping_source"] == ""
|
|
assert kaks_resolution["status"] == "needs_reference_data"
|
|
assert kaks_resolution["issue_code"] == "kaks_reference_cds_missing"
|
|
|
|
|
|
def test_ppi_kaks_dependency_preflight_requires_reference_cds_before_ortholog(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
input_dir.mkdir()
|
|
final_dir.mkdir(parents=True)
|
|
(input_dir / "soe").mkdir()
|
|
(input_dir / "soe" / "soe.pep.fasta").write_text(">SoeGene1\nMA\n", encoding="utf-8")
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=["Spinacia oleracea"],
|
|
family="SPL",
|
|
requested_stages=["ppi", "kaks"],
|
|
)
|
|
|
|
assert resolution["status"] == "needs_reference_data"
|
|
assert resolution["issue_code"] == "kaks_reference_cds_missing"
|
|
assert resolution["requested_stages"] == ["ppi", "kaks"]
|
|
assert resolution["reference_bundle_status"]["missing_required"] == ["cds"]
|
|
assert resolution["reference_bundle_status"]["kaks_ready"] is False
|
|
|
|
|
|
def test_ppi_kaks_dependency_preflight_with_pep_and_cds_needs_ortholog(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
input_dir.mkdir()
|
|
final_dir.mkdir(parents=True)
|
|
(input_dir / "soe").mkdir()
|
|
(input_dir / "soe" / "soe.pep.fasta").write_text(">SoeGene1\nMA\n", encoding="utf-8")
|
|
(input_dir / "soe" / "soe.cds.fasta").write_text(">SoeGene1\nATGGCT\n", encoding="utf-8")
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=["Spinacia oleracea"],
|
|
family="SPL",
|
|
requested_stages=["ppi", "kaks"],
|
|
)
|
|
|
|
assert resolution["status"] == "needs_ortholog"
|
|
assert resolution["reference_bundle_status"]["kaks_ready"] is True
|
|
assert resolution["diagnostics"]["ortholog_dependency_role"] == "producer"
|
|
|
|
|
|
def test_ppi_kaks_dependency_preflight_rejects_stale_at_mapping_when_spinacia_selected(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
input_dir.mkdir()
|
|
final_dir.mkdir(parents=True)
|
|
(input_dir / "soe").mkdir()
|
|
(input_dir / "soe" / "soe.pep.fasta").write_text(">SoeGene1\nMA\n", encoding="utf-8")
|
|
(input_dir / "soe" / "soe.cds.fasta").write_text(">SoeGene1\nATGGCT\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.orthologs.txt").write_text(
|
|
"FtGene1\tAT1G01010\t90\tFt\tAt\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=["Spinacia oleracea"],
|
|
family="SPL",
|
|
requested_stages=["ppi", "kaks"],
|
|
)
|
|
|
|
assert resolution["status"] == "needs_ortholog"
|
|
assert resolution["issue_code"] == "stale_ortholog_reference_conflict"
|
|
assert resolution["mapping_reference_species"] == "At"
|
|
|
|
|
|
def test_ppi_species_resolution_selected_near_string_species_with_mapping_is_ready(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
input_dir.mkdir()
|
|
final_dir.mkdir(parents=True)
|
|
(input_dir / "soe").mkdir()
|
|
(input_dir / "soe" / "soe.pep.fasta").write_text(">SoeGene1\nMA\n", encoding="utf-8")
|
|
(input_dir / "soe" / "soe.cds.fasta").write_text(">SoeGene1\nATGGCT\n", encoding="utf-8")
|
|
_materialize_trusted_ppi_mapping(
|
|
job, prefix="Ft.SPL", target="Ft", family="SPL",
|
|
storage_key="soe", taxon_id="3562", query_id="FtGene1", subject_id="SoeGene1",
|
|
)
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=["Spinacia oleracea"],
|
|
family="SPL",
|
|
ppi_reference_taxon_id="3562",
|
|
ppi_reference_storage_key="soe",
|
|
ppi_reference_selection_mode="tree_nearest",
|
|
ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi",
|
|
)
|
|
|
|
assert resolution["status"] == "ready"
|
|
assert resolution["selected_string_species"] == "soe"
|
|
assert resolution["selected_taxon_id"] == "3562"
|
|
assert resolution["ortholog_mapping_source"].endswith("Ft.SPL.ppi.orthologs.txt")
|
|
assert resolution["decision"] == "ready_reference_mapping"
|
|
|
|
|
|
def test_ppi_species_resolution_can_ignore_existing_mapping_for_pre_ortholog_check(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
input_dir.mkdir()
|
|
final_dir.mkdir(parents=True)
|
|
(input_dir / "soe").mkdir()
|
|
(input_dir / "soe" / "soe.pep.fasta").write_text(">SoeGene1\nMA\n", encoding="utf-8")
|
|
(input_dir / "soe" / "soe.cds.fasta").write_text(">SoeGene1\nATGGCT\n", encoding="utf-8")
|
|
_materialize_trusted_ppi_mapping(
|
|
job, prefix="Ft.SPL", target="Ft", family="SPL",
|
|
storage_key="soe", taxon_id="3562", query_id="FtGene1", subject_id="SoeGene1",
|
|
)
|
|
|
|
ready = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=["Spinacia oleracea"],
|
|
family="SPL",
|
|
ppi_reference_taxon_id="3562",
|
|
ppi_reference_storage_key="soe",
|
|
ppi_reference_selection_mode="tree_nearest",
|
|
ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi",
|
|
)
|
|
pre_ortholog = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=["Spinacia oleracea"],
|
|
family="SPL",
|
|
ignore_existing_ortholog=True,
|
|
ppi_reference_taxon_id="3562",
|
|
ppi_reference_storage_key="soe",
|
|
ppi_reference_selection_mode="tree_nearest",
|
|
ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi",
|
|
)
|
|
|
|
assert ready["status"] == "ready"
|
|
assert ready["ortholog_mapping_source"].endswith("Ft.SPL.ppi.orthologs.txt")
|
|
assert pre_ortholog["status"] == "needs_ortholog"
|
|
assert pre_ortholog["ortholog_mapping_source"] == ""
|
|
assert pre_ortholog["selected_string_species"] == "soe"
|
|
assert pre_ortholog["candidates"][0]["selected_string_taxon_id"] == "3562"
|
|
|
|
|
|
def test_ppi_kaks_preflight_recovers_prior_non_ath_reference_before_config_default(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
soe_dir = input_dir / "soe"
|
|
input_dir.mkdir()
|
|
soe_dir.mkdir()
|
|
final_dir.mkdir(parents=True)
|
|
(soe_dir / "soe.pep.fasta").write_text(">SoeGene1\nMA\n", encoding="utf-8")
|
|
(input_dir / "At.pep.fasta").write_text(">AT1G01010\nMA\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.orthologs.txt").write_text(
|
|
"FtGene1\tAT1G01010\t90\tFt\tAt\n",
|
|
encoding="utf-8",
|
|
)
|
|
soe_manifest = final_dir / "Ft.SPL.SPL.target-Ft.ref-soe.target_plus_reference.ortholog.manifest.json"
|
|
soe_manifest.write_text(json.dumps({
|
|
"stage": "ortholog",
|
|
"status": "success",
|
|
"reference_species": ["soe"],
|
|
"scope_id": "SPL.target-Ft.ref-soe.target_plus_reference",
|
|
}), encoding="utf-8")
|
|
at_manifest = final_dir / "Ft.SPL.SPL.target-Ft.ref-At.target_plus_reference.ppi.manifest.json"
|
|
at_manifest.write_text(json.dumps({
|
|
"stage": "ppi",
|
|
"status": "success",
|
|
"reference_species": ["At"],
|
|
"scope_id": "SPL.target-Ft.ref-At.target_plus_reference",
|
|
}), encoding="utf-8")
|
|
os.utime(soe_manifest, (1_000_000, 1_000_000))
|
|
os.utime(at_manifest, (2_000_000, 2_000_000))
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=["At"],
|
|
requested_stages=["ppi"],
|
|
family="SPL",
|
|
)
|
|
|
|
assert resolution["reference_species"] == ""
|
|
assert resolution["selected_string_species"] == ""
|
|
assert resolution["status"] == "confirmation_required"
|
|
assert resolution["issue_code"] == "ppi_reference_confirmation_required"
|
|
assert resolution["mapping_reference_species"] == ""
|
|
|
|
|
|
def test_kaks_preflight_recovers_non_ath_reference_and_prioritizes_missing_cds(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
soe_dir = input_dir / "soe"
|
|
input_dir.mkdir()
|
|
soe_dir.mkdir()
|
|
final_dir.mkdir(parents=True)
|
|
(soe_dir / "soe.pep.fasta").write_text(">SoeGene1\nMA\n", encoding="utf-8")
|
|
(input_dir / "At.pep.fasta").write_text(">AT1G01010\nMA\n", encoding="utf-8")
|
|
(input_dir / "At.cds.fasta").write_text(">AT1G01010\nATG\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.orthologs.txt").write_text(
|
|
"FtGene1\tAT1G01010\t90\tFt\tAt\n",
|
|
encoding="utf-8",
|
|
)
|
|
(final_dir / "Ft.SPL.SPL.target-Ft.ref-soe.target_plus_reference.ortholog.manifest.json").write_text(json.dumps({
|
|
"stage": "ortholog",
|
|
"status": "success",
|
|
"reference_species": ["soe"],
|
|
"scope_id": "SPL.target-Ft.ref-soe.target_plus_reference",
|
|
}), encoding="utf-8")
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Fagopyrum tataricum"],
|
|
reference_species=["At"],
|
|
requested_stages=["kaks"],
|
|
family="SPL",
|
|
)
|
|
|
|
assert resolution["reference_species"] == "soe"
|
|
assert resolution["selected_string_species"] == "soe"
|
|
assert resolution["status"] == "needs_reference_data"
|
|
assert resolution["issue_code"] == "kaks_reference_cds_missing"
|
|
assert resolution["reference_bundle_status"]["missing_required"] == ["cds"]
|
|
|
|
|
|
def test_ppi_species_resolution_requires_cds_when_only_reference_pep_exists(tmp_path):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
input_dir.mkdir()
|
|
final_dir.mkdir(parents=True)
|
|
(input_dir / "soe").mkdir()
|
|
(input_dir / "soe" / "soe.pep.fasta").write_text(">SoeGene1\nMA\n", encoding="utf-8")
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
target_species=["Ft"],
|
|
reference_species=["Spinacia oleracea"],
|
|
family="SPL",
|
|
ppi_reference_taxon_id="3562",
|
|
ppi_reference_storage_key="soe",
|
|
ppi_reference_selection_mode="tree_nearest",
|
|
ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi",
|
|
)
|
|
|
|
assert resolution["status"] == "needs_reference_data"
|
|
assert resolution["issue_code"] == "ppi_reference_data_missing"
|
|
assert resolution["reference_bundle_status"]["missing_required"] == ["cds"]
|
|
assert resolution["reference_data_source"].endswith("soe/soe.pep.fasta")
|
|
assert resolution["selected_string_species"] == "soe"
|
|
|
|
|
|
def test_ppi_reference_bundle_download_requests_pep_and_cds(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
(tmp_path / "input").mkdir()
|
|
(tmp_path / "input" / "soe").mkdir()
|
|
(tmp_path / "workspace").mkdir()
|
|
captured = {}
|
|
|
|
monkeypatch.setattr(
|
|
backend,
|
|
"_select_ncbi_reference_assembly",
|
|
lambda **kwargs: {"accession": "GCF_TEST", "organism_name": "Spinacia oleracea", "taxon_id": "3562"},
|
|
)
|
|
|
|
def fake_download_url(accession, file_types):
|
|
captured["file_types"] = list(file_types)
|
|
return "https://example.invalid/reference.zip"
|
|
|
|
def fake_download(_url, zip_path, timeout_sec=300):
|
|
zip_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with zipfile.ZipFile(zip_path, "w") as zf:
|
|
zf.writestr("README.txt", "fake")
|
|
return 4
|
|
|
|
monkeypatch.setattr(backend, "_ncbi_dataset_download_url", fake_download_url)
|
|
monkeypatch.setattr(backend, "_download_url_to_file", fake_download)
|
|
monkeypatch.setattr(
|
|
backend,
|
|
"_extract_ncbi_reference_bundle_zip",
|
|
lambda **kwargs: {
|
|
"pep": str(tmp_path / "input" / "soe" / "soe.pep.fasta"),
|
|
"cds": str(tmp_path / "input" / "soe" / "soe.cds.fasta"),
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
backend,
|
|
"_build_reference_bundle_id_map",
|
|
lambda **kwargs: {
|
|
"id_map_path": str(tmp_path / "input" / "soe" / "soe.pep_cds_idmap.tsv"),
|
|
"pep_records": 1,
|
|
"cds_records": 1,
|
|
"pep_cds_overlap": 1,
|
|
},
|
|
)
|
|
|
|
result = backend.download_reference_annotation_bundle(
|
|
job=job,
|
|
species_code="soe",
|
|
species_name="Spinacia oleracea",
|
|
taxon_id="3562",
|
|
requested_stages=["ppi"],
|
|
)
|
|
|
|
assert captured["file_types"] == ["pep", "cds"]
|
|
assert result["status"] == "success"
|
|
assert result["missing_required"] == []
|
|
assert result["requested_file_types"] == ["pep", "cds"]
|
|
|
|
|
|
def test_ppi_execution_scope_canonicalizes_full_species_names_before_mapping_filter(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
final_dir = workspace / "final"
|
|
logs_dir = workspace / "logs"
|
|
input_dir.mkdir(parents=True)
|
|
final_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
ath_dir = input_dir / "ath"
|
|
ath_dir.mkdir()
|
|
(ath_dir / "ath.pep_cds_idmap.tsv").write_text(
|
|
"species_code\tprotein_id\nath\tAT1G01010\n", encoding="utf-8"
|
|
)
|
|
(final_dir / "Ft.SPL.final_members.id").write_text("FtGene1\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.orthologs.txt").write_text(
|
|
"FtGene1\tAT1G01010\t90\tFt\tAt\n",
|
|
encoding="utf-8",
|
|
)
|
|
scope_file = final_dir / "Ft.SPL.full_name.scope.json"
|
|
scope_file.write_text(json.dumps({
|
|
"family": "SPL",
|
|
"prefix": "Ft.SPL",
|
|
"scope_id": "SPL.target-Fagopyrum_tataricum.ref-Arabidopsis_thaliana.target_plus_reference",
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Fagopyrum tataricum"],
|
|
"reference_species": ["Arabidopsis thaliana"],
|
|
}), encoding="utf-8")
|
|
|
|
captured = {}
|
|
|
|
def fake_run_cmd_logged(cmd, log_path, cwd=None):
|
|
captured["cmd"] = cmd
|
|
_write_fake_ppi_outputs(workspace / "ppi", prefix="Ft.SPL")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run_cmd_logged)
|
|
monkeypatch.setattr(backend, "download_string_network_image", lambda *args, **kwargs: None)
|
|
|
|
message, manifest = backend.run_stage(
|
|
"ppi",
|
|
job,
|
|
"Ft.SPL",
|
|
ppi_species="auto",
|
|
ppi_score=200,
|
|
scope_file=str(scope_file),
|
|
analysis_route="target_plus_reference",
|
|
)
|
|
|
|
gene_ids_file = backend.Path(captured["cmd"][captured["cmd"].index("--gene-ids") + 1])
|
|
assert "✅ PPI analysis completed" in message
|
|
assert manifest and manifest.exists()
|
|
assert captured["cmd"][captured["cmd"].index("--species") + 1] == "ath"
|
|
assert gene_ids_file.read_text(encoding="utf-8").splitlines() == ["FtGene1"]
|
|
|
|
resolution = backend.resolve_ppi_species_resolution(
|
|
job=job,
|
|
prefix="Ft.SPL",
|
|
ppi_species="auto",
|
|
scope_file=str(scope_file),
|
|
family="SPL",
|
|
)
|
|
assert resolution["status"] == "confirmation_required"
|
|
assert resolution["issue_code"] == "ppi_reference_confirmation_required"
|
|
assert resolution["required_next_stage"] == ""
|
|
|
|
|
|
def test_ppi_auto_prefers_curated_members_when_available(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
final_dir = workspace / "final"
|
|
logs_dir = workspace / "logs"
|
|
final_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
input_dir.mkdir(parents=True)
|
|
curated_file = final_dir / "Target.Os.CRR.curated_members.id"
|
|
final_file = final_dir / "Target.Os.CRR.final_members.id"
|
|
curated_file.write_text("Os01t0810800-01\nOs09t0359500-01\n", encoding="utf-8")
|
|
final_file.write_text(
|
|
"Os01t0810800-01\nOs09t0359500-01\nOs05t0393100-01\n",
|
|
encoding="utf-8",
|
|
)
|
|
scope_file = final_dir / "Os.CRR.CRR.target-Os.ref-At.target_plus_reference.scope.json"
|
|
_write_ppi_scope(scope_file)
|
|
|
|
captured = {}
|
|
|
|
def fake_run_cmd_logged(cmd, log_path, cwd=None):
|
|
captured["cmd"] = cmd
|
|
_write_fake_ppi_outputs(workspace / "ppi")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda *args, **kwargs: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run_cmd_logged)
|
|
monkeypatch.setattr(backend, "download_string_network_image", lambda *args, **kwargs: None)
|
|
|
|
message, manifest = backend.run_stage(
|
|
"ppi",
|
|
job,
|
|
"Os.CRR",
|
|
ppi_species="auto",
|
|
ppi_score=400,
|
|
scope_file=str(scope_file),
|
|
analysis_route="target_plus_reference",
|
|
)
|
|
|
|
cmd = captured["cmd"]
|
|
assert "✅ PPI analysis completed" in message
|
|
assert manifest and manifest.exists()
|
|
assert cmd[cmd.index("--gene-ids") + 1] == str(curated_file)
|
|
|
|
|
|
def test_ppi_projected_soe_uses_exact_mapping_and_same_species_id_map(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
final_dir = workspace / "final"
|
|
logs_dir = workspace / "logs"
|
|
final_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
input_dir.mkdir(parents=True)
|
|
soe_dir = input_dir / "soe"
|
|
soe_dir.mkdir()
|
|
id_map_file = soe_dir / "soe.pep_cds_idmap.tsv"
|
|
id_map_file.write_text("species_code\tprotein_id\nsoe\tSoeGene1\n", encoding="utf-8")
|
|
(final_dir / "Os.CRR.final_members.id").write_text("Os01t0810800-01\n", encoding="utf-8")
|
|
mapping_file = final_dir / "Os.CRR.orthologs.txt"
|
|
mapping_file.write_text("Os01t0810800-01\tSoeGene1\t90\tOs\tsoe\n", encoding="utf-8")
|
|
scope_file = final_dir / "Os.CRR.CRR.target-Os.ref-soe.target_plus_reference.scope.json"
|
|
_write_ppi_scope(scope_file, reference_species="soe")
|
|
|
|
captured = {}
|
|
|
|
def fake_run_cmd_logged(cmd, log_path, cwd=None):
|
|
captured["cmd"] = cmd
|
|
_write_fake_ppi_outputs(workspace / "ppi")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda *args, **kwargs: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run_cmd_logged)
|
|
monkeypatch.setattr(backend, "download_string_network_image", lambda *args, **kwargs: None)
|
|
|
|
message, manifest = backend.run_stage(
|
|
"ppi",
|
|
job,
|
|
"Os.CRR",
|
|
ppi_species="soe",
|
|
ppi_score=400,
|
|
scope_file=str(scope_file),
|
|
analysis_route="target_plus_reference",
|
|
)
|
|
|
|
cmd = captured["cmd"]
|
|
assert "✅ PPI analysis completed" in message
|
|
assert manifest and manifest.exists()
|
|
assert cmd[cmd.index("--species") + 1] == "soe"
|
|
assert cmd[cmd.index("--mapping") + 1] == str(mapping_file)
|
|
assert cmd[cmd.index("--id-map") + 1] == str(id_map_file)
|
|
|
|
|
|
def test_ppi_projected_soe_rejects_only_ath_id_map(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
logs_dir = tmp_path / "workspace" / "logs"
|
|
(input_dir / "ath").mkdir(parents=True)
|
|
final_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
ath_map = input_dir / "ath" / "ath.pep_cds_idmap.tsv"
|
|
ath_map.write_text("species_code\tprotein_id\nath\tAT1G51790\n", encoding="utf-8")
|
|
(final_dir / "Os.CRR.final_members.id").write_text("Os01t0810800-01\n", encoding="utf-8")
|
|
(final_dir / "Os.CRR.orthologs.txt").write_text(
|
|
"Os01t0810800-01\tSoeGene1\t90\tOs\tsoe\n", encoding="utf-8"
|
|
)
|
|
scope_file = final_dir / "Os.CRR.CRR.target-Os.ref-soe.target_plus_reference.scope.json"
|
|
_write_ppi_scope(scope_file, reference_species="soe")
|
|
launched = []
|
|
monkeypatch.setattr(backend, "run_cmd_logged", lambda *args, **kwargs: launched.append(args))
|
|
|
|
message, _manifest = backend.run_stage(
|
|
"ppi",
|
|
job,
|
|
"Os.CRR",
|
|
ppi_species="soe",
|
|
scope_file=str(scope_file),
|
|
analysis_route="target_plus_reference",
|
|
)
|
|
|
|
expected = input_dir / "soe" / "soe.pep_cds_idmap.tsv"
|
|
assert launched == []
|
|
assert str(expected) in message
|
|
assert "species-matched ID map" in message
|
|
assert str(ath_map) not in message
|
|
|
|
|
|
def test_ppi_projected_soe_rejects_missing_id_map(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
logs_dir = tmp_path / "workspace" / "logs"
|
|
input_dir.mkdir(parents=True)
|
|
final_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
(final_dir / "Os.CRR.final_members.id").write_text("Os01t0810800-01\n", encoding="utf-8")
|
|
(final_dir / "Os.CRR.orthologs.txt").write_text(
|
|
"Os01t0810800-01\tSoeGene1\t90\tOs\tsoe\n", encoding="utf-8"
|
|
)
|
|
scope_file = final_dir / "Os.CRR.CRR.target-Os.ref-soe.target_plus_reference.scope.json"
|
|
_write_ppi_scope(scope_file, reference_species="soe")
|
|
launched = []
|
|
monkeypatch.setattr(backend, "run_cmd_logged", lambda *args, **kwargs: launched.append(args))
|
|
|
|
message, _manifest = backend.run_stage(
|
|
"ppi",
|
|
job,
|
|
"Os.CRR",
|
|
ppi_species="soe",
|
|
scope_file=str(scope_file),
|
|
analysis_route="target_plus_reference",
|
|
)
|
|
|
|
expected = input_dir / "soe" / "soe.pep_cds_idmap.tsv"
|
|
assert launched == []
|
|
assert message == f"❌ PPI projected analysis requires species-matched ID map: {expected}"
|
|
|
|
|
|
def test_ppi_projected_builds_missing_id_map_from_species_matched_pep_and_cds(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
workspace = tmp_path / "workspace"
|
|
final_dir = workspace / "final"
|
|
logs_dir = workspace / "logs"
|
|
soe_dir = input_dir / "soe"
|
|
soe_dir.mkdir(parents=True)
|
|
final_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
(soe_dir / "soe.pep.fasta").write_text(">SoeGene1 [gene=SoeGene1]\nMAAA\n", encoding="utf-8")
|
|
(soe_dir / "soe.cds.fasta").write_text(">SoeGene1 [gene=SoeGene1]\nATGGCT\n", encoding="utf-8")
|
|
(final_dir / "Os.CRR.final_members.id").write_text("Os01t0810800-01\n", encoding="utf-8")
|
|
mapping_file = final_dir / "Os.CRR.ppi.orthologs.txt"
|
|
mapping_file.write_text("Os01t0810800-01\tSoeGene1\t90\tOs\tsoe\n", encoding="utf-8")
|
|
scope_file = final_dir / "Os.CRR.CRR.target-Os.target_only.scope.json"
|
|
scope_file.write_text(json.dumps({
|
|
"family": "CRR",
|
|
"prefix": "Os.CRR",
|
|
"scope_id": "CRR.target-Os.target_only",
|
|
"scope_mode": "target_only",
|
|
"target_species": ["Os"],
|
|
"reference_species": [],
|
|
"selected_species": ["Os"],
|
|
}), encoding="utf-8")
|
|
captured = {}
|
|
|
|
def fake_run_cmd_logged(cmd, log_path, cwd=None):
|
|
captured["cmd"] = cmd
|
|
_write_fake_ppi_outputs(workspace / "ppi")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda *args, **kwargs: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run_cmd_logged)
|
|
monkeypatch.setattr(backend, "download_string_network_image", lambda *args, **kwargs: None)
|
|
|
|
message, manifest = backend.run_stage(
|
|
"ppi",
|
|
job,
|
|
"Os.CRR",
|
|
ppi_species="auto",
|
|
ppi_reference_storage_key="soe",
|
|
ppi_reference_taxon_id="3562",
|
|
ppi_reference_user_confirmed=True,
|
|
dependency_owner="ppi",
|
|
ppi_score=400,
|
|
scope_file=str(scope_file),
|
|
analysis_route="single_species",
|
|
)
|
|
|
|
expected = soe_dir / "soe.pep_cds_idmap.tsv"
|
|
assert "✅ PPI analysis completed" in message
|
|
assert manifest and manifest.exists()
|
|
assert expected.is_file()
|
|
assert captured["cmd"][captured["cmd"].index("--id-map") + 1] == str(expected)
|
|
assert captured["cmd"][captured["cmd"].index("--normalization-species-label") + 1] == "soe"
|
|
|
|
|
|
def test_ppi_explicit_absolute_species_rejected_before_outside_id_map_lookup(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "job" / "input"),
|
|
workspace_dir=str(tmp_path / "job" / "workspace"),
|
|
logs_dir=str(tmp_path / "job" / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "job" / "input"
|
|
workspace = tmp_path / "job" / "workspace"
|
|
final_dir = workspace / "final"
|
|
outside_species = tmp_path / "outside" / "soe"
|
|
outside_id_map = outside_species.with_name("soe.pep_cds_idmap.tsv")
|
|
input_dir.mkdir(parents=True)
|
|
final_dir.mkdir(parents=True)
|
|
(workspace / "logs").mkdir(parents=True)
|
|
outside_id_map.parent.mkdir(parents=True)
|
|
outside_id_map.write_text("species_code\tprotein_id\nsoe\tSoeGene1\n", encoding="utf-8")
|
|
(final_dir / "Os.CRR.final_members.id").write_text("Os01t0810800-01\n", encoding="utf-8")
|
|
(final_dir / "Os.CRR.orthologs.txt").write_text(
|
|
"Os01t0810800-01\tSoeGene1\t90\tOs\tsoe\n", encoding="utf-8"
|
|
)
|
|
scope_file = final_dir / "Os.CRR.CRR.target-Os.ref-soe.target_plus_reference.scope.json"
|
|
_write_ppi_scope(scope_file, reference_species="soe")
|
|
launched = []
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda *args, **kwargs: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", lambda *args, **kwargs: launched.append(args))
|
|
|
|
requested_species = str(outside_species)
|
|
message, _manifest = backend.run_stage(
|
|
"ppi",
|
|
job,
|
|
"Os.CRR",
|
|
ppi_species=requested_species,
|
|
scope_file=str(scope_file),
|
|
analysis_route="target_plus_reference",
|
|
)
|
|
|
|
assert launched == []
|
|
assert "unsupported STRING species" in message
|
|
assert requested_species in message
|
|
assert str(outside_id_map) not in message
|
|
|
|
|
|
def test_ppi_missing_authoritative_skill_never_selects_legacy_cli(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
input_dir = tmp_path / "input"
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
logs_dir = tmp_path / "workspace" / "logs"
|
|
input_dir.mkdir(parents=True)
|
|
final_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
(final_dir / "Os.CRR.final_members.id").write_text(
|
|
"Os01g0810800\n", encoding="utf-8"
|
|
)
|
|
|
|
original_exists = backend.Path.exists
|
|
|
|
def fake_exists(path):
|
|
if str(path) == "skills/ppi/run.py":
|
|
return False
|
|
if str(path) == "apps/cli/run_ppi_v2.py":
|
|
return True
|
|
return original_exists(path)
|
|
|
|
launched = []
|
|
|
|
def fail_if_launched(*args, **kwargs):
|
|
launched.append((args, kwargs))
|
|
raise AssertionError("no PPI implementation may launch")
|
|
|
|
monkeypatch.setattr(backend.Path, "exists", fake_exists)
|
|
monkeypatch.setattr(backend, "can_start_new_task", lambda *args, **kwargs: (True, "ok"))
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fail_if_launched)
|
|
|
|
message, manifest = backend.run_stage(
|
|
"ppi", job, "Os.CRR", ppi_species="osa", ppi_score=400
|
|
)
|
|
|
|
assert "authoritative ppi skill" in message.lower()
|
|
assert "legacy fallback disabled" in message
|
|
assert "run_ppi_v2.py" not in message
|
|
assert launched == []
|
|
expected_manifest = final_dir / "Os.CRR.ppi.manifest.json"
|
|
assert manifest == expected_manifest
|
|
payload = json.loads(expected_manifest.read_text(encoding="utf-8"))
|
|
assert payload["stage"] == "ppi"
|
|
assert payload["prefix"] == "Os.CRR"
|
|
assert payload["status"] == "failed"
|
|
assert "authoritative ppi skill" in payload["error"].lower()
|
|
assert "legacy fallback disabled" in payload["error"]
|
|
|
|
|
|
def test_string_network_image_falls_back_to_standard_png(tmp_path, monkeypatch):
|
|
calls = []
|
|
|
|
class FakeResponse:
|
|
headers = {"content-type": "image/png"}
|
|
status_code = 200
|
|
|
|
def __init__(self, content, url):
|
|
self._content = content
|
|
self.url = url
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def iter_content(self, chunk_size=1):
|
|
yield self._content
|
|
|
|
def fake_post(url, data=None, timeout=None, stream=None, allow_redirects=None):
|
|
assert allow_redirects is False
|
|
calls.append(url)
|
|
if "highres_image" in url:
|
|
raise TimeoutError("highres timeout")
|
|
return FakeResponse(b"\x89PNG\r\n\x1a\nstandard-png", url)
|
|
|
|
monkeypatch.setitem(sys.modules, "requests", types.SimpleNamespace(post=fake_post))
|
|
|
|
output = tmp_path / "Os.CRR.ppi_network.string.png"
|
|
result = backend.download_string_network_image(
|
|
"https://string-db.org/cgi/network?identifiers=Os01g0810800&species=39947",
|
|
output,
|
|
)
|
|
|
|
assert result == output
|
|
assert output.read_bytes() == b"\x89PNG\r\n\x1a\nstandard-png"
|
|
assert calls == [
|
|
"https://string-db.org/api/highres_image/network",
|
|
"https://string-db.org/api/image/network",
|
|
]
|
|
|
|
|
|
def test_string_network_image_falls_back_to_svg(tmp_path, monkeypatch):
|
|
calls = []
|
|
|
|
class FakeResponse:
|
|
status_code = 200
|
|
|
|
def __init__(self, content_type, content, url):
|
|
self.headers = {"content-type": content_type}
|
|
self._content = content
|
|
self.url = url
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def iter_content(self, chunk_size=1):
|
|
yield self._content
|
|
|
|
def fake_post(url, data=None, timeout=None, stream=None, allow_redirects=None):
|
|
assert allow_redirects is False
|
|
calls.append(url)
|
|
if "svg" in url:
|
|
return FakeResponse("image/svg+xml", b"<svg></svg>", url)
|
|
raise TimeoutError("png timeout")
|
|
|
|
monkeypatch.setitem(sys.modules, "requests", types.SimpleNamespace(post=fake_post))
|
|
|
|
output = tmp_path / "Os.CRR.ppi_network.string.png"
|
|
result = backend.download_string_network_image(
|
|
"https://string-db.org/cgi/network?identifiers=Os01g0810800&species=39947",
|
|
output,
|
|
)
|
|
|
|
assert result == tmp_path / "Os.CRR.ppi_network.string.svg"
|
|
assert result.read_bytes() == b"<svg></svg>"
|
|
assert calls == [
|
|
"https://string-db.org/api/highres_image/network",
|
|
"https://string-db.org/api/image/network",
|
|
"https://string-db.org/api/svg/network",
|
|
]
|
|
|
|
|
|
def test_create_alphafold2_links_marks_single_protein_entries(tmp_path):
|
|
id_resolution = tmp_path / "Os.CRR.ppi_id_resolution.tsv"
|
|
id_resolution.write_text(
|
|
"\t".join(["raw_id", "normalized_id", "selected_query_id", "string_id", "mapping_status", "method", "reason"])
|
|
+ "\n"
|
|
+ "\t".join(["Os01t0810800-01", "Os01g0810800", "Os01g0810800", "", "candidate", "species_rule", ""])
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
network = tmp_path / "Os.CRR.ppi_network.tsv"
|
|
network.write_text("protein1\tprotein2\tscore\nA0A0N7KDY0\tQ60ER5_ORYSJ\t0.234\n", encoding="utf-8")
|
|
output = tmp_path / "Os.CRR.ppi_alphafold2_links.tsv"
|
|
|
|
result = backend.create_alphafold2_links(
|
|
id_resolution_file=id_resolution,
|
|
network_file=network,
|
|
output_file=output,
|
|
)
|
|
|
|
assert result == output
|
|
rows = output.read_text(encoding="utf-8").splitlines()
|
|
assert rows[0].split("\t") == [
|
|
"source_id",
|
|
"lookup_id",
|
|
"link_type",
|
|
"alphafold_db_url",
|
|
"structure_type",
|
|
"note",
|
|
]
|
|
assert "AlphaFold DB/Fold2 single protein" in rows[1]
|
|
assert "complex" not in output.read_text(encoding="utf-8").lower()
|
|
assert "https://alphafold.ebi.ac.uk/entry/A0A0N7KDY0" in output.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_sync_ppi_manifest_artifacts_points_to_final_files(tmp_path):
|
|
final_dir = tmp_path / "final"
|
|
final_dir.mkdir()
|
|
manifest = final_dir / "Os.CRR.ppi.manifest.json"
|
|
for name in [
|
|
"Os.CRR.ppi_network.tsv",
|
|
"Os.CRR.ppi_report.md",
|
|
"Os.CRR.ppi_id_resolution.tsv",
|
|
"Os.CRR.ppi_url.txt",
|
|
]:
|
|
(final_dir / name).write_text("x\n", encoding="utf-8")
|
|
manifest.write_text(
|
|
json.dumps({
|
|
"manifest_version": "1.0",
|
|
"stage": "ppi",
|
|
"status": "success",
|
|
"prefix": "Os.CRR",
|
|
"artifacts": {
|
|
"network": {"path": "workspace/job/ppi/Os.CRR_network.tsv", "type": "tsv"},
|
|
"report": {"path": "workspace/job/ppi/Os.CRR_report.md", "type": "markdown"},
|
|
"id_resolution": {"path": "workspace/job/ppi/Os.CRR_id_resolution.tsv", "type": "tsv"},
|
|
"string_url": {"path": "workspace/job/ppi/Os.CRR_url.txt", "type": "txt"},
|
|
},
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
backend.sync_ppi_manifest_final_artifacts(manifest, final_dir, "Os.CRR")
|
|
|
|
data = json.loads(manifest.read_text(encoding="utf-8"))
|
|
assert data["artifacts"]["network"]["path"] == "final/Os.CRR.ppi_network.tsv"
|
|
assert data["artifacts"]["report"]["path"] == "final/Os.CRR.ppi_report.md"
|
|
assert data["artifacts"]["id_resolution"]["path"] == "final/Os.CRR.ppi_id_resolution.tsv"
|
|
assert data["artifacts"]["string_url"]["path"] == "final/Os.CRR.ppi_url.txt"
|
|
assert data["artifacts"]["network"]["full_path"] == str((final_dir / "Os.CRR.ppi_network.tsv").absolute())
|
|
|
|
|
|
def test_prepare_ppi_reset_inputs_creates_isolated_diff_and_manifest(tmp_path):
|
|
final_dir = tmp_path / "final"
|
|
final_dir.mkdir()
|
|
source = final_dir / "Target.Os.CRR.curated_members.id"
|
|
source.write_text("Os01g0810800\nOs09g0359500\nOs05g0393100\n", encoding="utf-8")
|
|
original_ppi = final_dir / "Os.CRR.ppi_network.tsv"
|
|
original_ppi.write_text("protein1\tprotein2\tscore\nA\tB\t0.9\n", encoding="utf-8")
|
|
|
|
result = backend.prepare_ppi_reset_inputs(
|
|
final_dir=final_dir,
|
|
prefix="Os.CRR",
|
|
source_gene_id_file=source,
|
|
submitted_gene_ids=["Os01g0810800", "Os05g0393100", "Os07g0501800"],
|
|
instruction="remove one member and add one candidate",
|
|
ppi_species="osa",
|
|
ppi_score=200,
|
|
)
|
|
|
|
reset_dir = final_dir / "ppi_reset" / "reset_001"
|
|
assert result["reset_id"] == "reset_001"
|
|
assert result["reset_dir"] == reset_dir
|
|
assert (reset_dir / "Os.CRR.ppi_reset_input.id").read_text(encoding="utf-8").splitlines() == [
|
|
"Os01g0810800",
|
|
"Os05g0393100",
|
|
"Os07g0501800",
|
|
]
|
|
diff_lines = (reset_dir / "Os.CRR.ppi_reset_diff.tsv").read_text(encoding="utf-8").splitlines()
|
|
assert "Os09g0359500\tdeleted" in diff_lines
|
|
assert "Os07g0501800\tadded" in diff_lines
|
|
manifest = json.loads((reset_dir / "Os.CRR.ppi_reset.manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["reset_id"] == "reset_001"
|
|
assert manifest["source_gene_id_file"] == str(source)
|
|
assert manifest["input_gene_count"] == 3
|
|
assert manifest["deleted_count"] == 1
|
|
assert manifest["added_count"] == 1
|
|
assert manifest["parameters"]["ppi_species"] == "osa"
|
|
assert original_ppi.read_text(encoding="utf-8").startswith("protein1\tprotein2")
|
|
|
|
|
|
def test_prepare_ppi_reset_inputs_uses_next_reset_directory(tmp_path):
|
|
final_dir = tmp_path / "final"
|
|
source = final_dir / "Target.Os.CRR.curated_members.id"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_text("A\nB\n", encoding="utf-8")
|
|
(final_dir / "ppi_reset" / "reset_001").mkdir(parents=True)
|
|
|
|
result = backend.prepare_ppi_reset_inputs(
|
|
final_dir=final_dir,
|
|
prefix="Os.CRR",
|
|
source_gene_id_file=source,
|
|
submitted_gene_ids=["A"],
|
|
instruction="second reset",
|
|
ppi_species="osa",
|
|
ppi_score=200,
|
|
)
|
|
|
|
assert result["reset_id"] == "reset_002"
|
|
assert result["reset_dir"] == final_dir / "ppi_reset" / "reset_002"
|
|
|
|
|
|
def test_run_ppi_reset_analysis_writes_outputs_under_reset_dir(tmp_path, monkeypatch):
|
|
job = backend.JobInfo(
|
|
job_id="job_test",
|
|
input_dir=str(tmp_path / "input"),
|
|
workspace_dir=str(tmp_path / "workspace"),
|
|
logs_dir=str(tmp_path / "workspace" / "logs"),
|
|
created_at="",
|
|
)
|
|
final_dir = tmp_path / "workspace" / "final"
|
|
logs_dir = tmp_path / "workspace" / "logs"
|
|
final_dir.mkdir(parents=True)
|
|
logs_dir.mkdir(parents=True)
|
|
source = final_dir / "Target.Os.CRR.curated_members.id"
|
|
source.write_text("A\nB\nC\n", encoding="utf-8")
|
|
|
|
captured = {}
|
|
|
|
def fake_run_cmd_logged(cmd, log_path, cwd=None):
|
|
captured["cmd"] = cmd
|
|
out_dir = cmd[cmd.index("--output") + 1]
|
|
ppi_dir = backend.Path(out_dir)
|
|
ppi_dir.mkdir(parents=True, exist_ok=True)
|
|
_write_fake_ppi_outputs(ppi_dir, prefix="Os.CRR")
|
|
return 0
|
|
|
|
monkeypatch.setattr(backend, "run_cmd_logged", fake_run_cmd_logged)
|
|
monkeypatch.setattr(backend, "download_string_network_image", lambda url, path: None)
|
|
|
|
result = backend.run_ppi_reset_analysis(
|
|
job=job,
|
|
prefix="Os.CRR",
|
|
source_gene_id_file=source,
|
|
submitted_gene_ids=["A", "C"],
|
|
instruction="remove B",
|
|
ppi_species="osa",
|
|
ppi_score=200,
|
|
)
|
|
|
|
reset_dir = final_dir / "ppi_reset" / "reset_001"
|
|
assert result["success"] is True
|
|
assert captured["cmd"][captured["cmd"].index("--gene-ids") + 1] == str(reset_dir / "Os.CRR.ppi_reset_input.id")
|
|
assert (reset_dir / "Os.CRR.ppi_network.tsv").exists()
|
|
assert (reset_dir / "Os.CRR.ppi_report.md").exists()
|
|
assert (reset_dir / "Os.CRR.ppi_id_resolution.tsv").exists()
|
|
assert (reset_dir / "Os.CRR.ppi_url.txt").exists()
|
|
assert (reset_dir / "Os.CRR.ppi_alphafold2_links.tsv").exists()
|
|
assert not (final_dir / "Os.CRR.ppi_network.tsv").exists()
|
|
manifest = json.loads((reset_dir / "Os.CRR.ppi_reset.manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["status"] == "success"
|
|
assert manifest["artifacts"]["network"]["path"] == "final/ppi_reset/reset_001/Os.CRR.ppi_network.tsv"
|