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"