134 lines
4.9 KiB
Python
Executable File
134 lines
4.9 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def test_subfamily_confirm_inherits_scope_from_source_prediction_manifest(tmp_path):
|
|
workspace = tmp_path / "job_scope_confirm"
|
|
final_dir = workspace / "final"
|
|
input_dir = tmp_path / "input" / "job_scope_confirm"
|
|
final_dir.mkdir(parents=True)
|
|
input_dir.mkdir(parents=True)
|
|
|
|
edited = final_dir / "Lc.SPL.confirmed_subfamily.edited.tsv"
|
|
edited.write_text(
|
|
"gene_id\tspecies\tpredicted_subfamily\tconfirmed_subfamily\tconfidence\tnote\n"
|
|
"LcGene1\tLc\tI\tI\thigh\tkept\n"
|
|
"LcGene2\tLc\tII\tIII\tmedium\tmanual\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
source_manifest = final_dir / "Target.SPL.target-Lc.ref-At.subfamily_prediction.reference_classification.manifest.json"
|
|
source_manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": "reference_classification",
|
|
"status": "success",
|
|
"prefix": "Lc.SPL",
|
|
"output_prefix": "Target.SPL.target-Lc.ref-At.subfamily_prediction",
|
|
"scope_id": "SPL.target-Lc.ref-At.target_plus_reference",
|
|
"scope_mode": "target_plus_reference",
|
|
"target_species": ["Lc"],
|
|
"reference_species": ["At"],
|
|
"comparative_species": [],
|
|
"selected_species": ["Lc", "At"],
|
|
"analysis_route": "subfamily_prediction",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
proc = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(ROOT / "apps/cli/run_subfamily_confirm.py"),
|
|
"--workspace",
|
|
str(workspace),
|
|
"--input-dir",
|
|
str(input_dir),
|
|
"--prefix",
|
|
"Lc.SPL",
|
|
"--edited-subfamily",
|
|
str(edited),
|
|
"--source-prediction-manifest",
|
|
str(source_manifest),
|
|
],
|
|
cwd=str(ROOT),
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert proc.returncode == 0, proc.stderr or proc.stdout
|
|
manifest_path = final_dir / "Lc.SPL.subfamily_confirm.manifest.json"
|
|
assert manifest_path.exists()
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
|
|
assert manifest["output_prefix"] == "Lc.SPL"
|
|
assert manifest["scope_id"] == "SPL.target-Lc.ref-At.target_plus_reference"
|
|
assert manifest["target_species"] == ["Lc"]
|
|
assert manifest["reference_species"] == ["At"]
|
|
assert manifest["parameters"]["source_prediction_manifest"] == str(source_manifest)
|
|
assert manifest["parameters"]["source_prediction_prefix"] == "Target.SPL.target-Lc.ref-At.subfamily_prediction"
|
|
assert (final_dir / "Lc.SPL.confirmed_subfamily.tsv").exists()
|
|
assert not (final_dir / "Lc.SPL.confirmed_subfamily.subfamily_confirm.manifest.json").exists()
|
|
|
|
|
|
def test_subfamily_confirm_promotes_curated_members_to_final_members(tmp_path):
|
|
workspace = tmp_path / "job_confirm_members"
|
|
final_dir = workspace / "final"
|
|
input_dir = tmp_path / "input" / "job_confirm_members"
|
|
final_dir.mkdir(parents=True)
|
|
input_dir.mkdir(parents=True)
|
|
|
|
(final_dir / "Ft.SPL.final_members.id").write_text("FtGene1\nFtGene2\nFtGene3\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.final_members.pep.fa").write_text(
|
|
">FtGene1\nMA\n>FtGene2\nMT\n>FtGene3\nMG\n",
|
|
encoding="utf-8",
|
|
)
|
|
(final_dir / "Ft.SPL.curated_members.id").write_text("FtGene1\nFtGene3\n", encoding="utf-8")
|
|
(final_dir / "Ft.SPL.curated_members.pep.fa").write_text(
|
|
">FtGene1\nMA\n>FtGene3\nMG\n",
|
|
encoding="utf-8",
|
|
)
|
|
edited = final_dir / "Ft.SPL.confirmed_subfamily.edited.tsv"
|
|
edited.write_text(
|
|
"gene_id\tconfirmed_subfamily\n"
|
|
"FtGene1\tI\n"
|
|
"FtGene3\tII\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
proc = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(ROOT / "apps/cli/run_subfamily_confirm.py"),
|
|
"--workspace",
|
|
str(workspace),
|
|
"--input-dir",
|
|
str(input_dir),
|
|
"--prefix",
|
|
"Ft.SPL",
|
|
"--edited-subfamily",
|
|
str(edited),
|
|
],
|
|
cwd=str(ROOT),
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
assert proc.returncode == 0, proc.stderr or proc.stdout
|
|
assert (final_dir / "Ft.SPL.final_members.id").read_text(encoding="utf-8") == "FtGene1\nFtGene3\n"
|
|
assert (final_dir / "Ft.SPL.final_members.pep.fa").read_text(encoding="utf-8") == ">FtGene1\nMA\n>FtGene3\nMG\n"
|
|
manifest = json.loads((final_dir / "Ft.SPL.subfamily_confirm.manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["metrics"]["final_members_count"] == 2
|
|
assert manifest["parameters"]["final_members_source_stage"] == "subfamily_confirm"
|
|
assert manifest["artifacts"]["final_members_id"]["path"] == "final/Ft.SPL.final_members.id"
|