import importlib.util from pathlib import Path import yaml from gfa_core.ppi_string_policy import ( STRING_SPECIES_MAP, build_ppi_species_diagnostic, evaluate_string_mapping_quality, normalize_gene_ids_for_string, reference_protein_candidate_paths, recommended_string_reference_species, species_to_string_code, string_taxon_id, ) REPO_ROOT = Path(__file__).resolve().parents[2] def load_ppi_skill_module(): path = REPO_ROOT / "skills/ppi/run.py" spec = importlib.util.spec_from_file_location("ppi_skill_run", path) module = importlib.util.module_from_spec(spec) assert spec and spec.loader spec.loader.exec_module(module) return module def test_shared_species_policy_resolves_common_aliases_without_silent_ath(): assert species_to_string_code("Os", fallback="") == "osa" assert species_to_string_code("rice", fallback="") == "osa" assert species_to_string_code("not_a_species", fallback="") == "" assert STRING_SPECIES_MAP["ath"] == "3702" assert STRING_SPECIES_MAP["osa"] def test_every_declared_string_species_code_resolves_to_itself(): unresolved = { code: species_to_string_code(code, fallback="") for code in STRING_SPECIES_MAP if species_to_string_code(code, fallback="") != code } assert unresolved == {} assert { code: string_taxon_id(code, fallback="") for code in STRING_SPECIES_MAP } == STRING_SPECIES_MAP def test_shared_species_policy_resolves_spinacia_as_string_supported_near_reference(): assert species_to_string_code("Spinacia oleracea", fallback="") == "soe" assert string_taxon_id("Spinacia oleracea", fallback="") == "3562" recommendations = recommended_string_reference_species("Ft") assert recommendations[0]["species_code"] == "soe" assert recommendations[0]["species_name"] == "Spinacia oleracea" assert recommendations[0]["taxon_id"] == "3562" assert recommendations[0]["relationship"] == "same_order_caryophyllales" def test_osa_transcript_ids_normalize_to_gene_locus_ids(): assert normalize_gene_ids_for_string(["Os01t0810800-01", "Os02t0000100"], "osa") == [ "Os01g0810800", "Os02g0000100", ] def test_mapping_quality_small_families_require_all_ids_mapped(): one_missing = evaluate_string_mapping_quality(input_count=2, mapped_count=1) assert one_missing["status"] == "failed" assert one_missing["issue_code"] == "ppi_string_id_mapping_low" all_mapped = evaluate_string_mapping_quality(input_count=2, mapped_count=2) assert all_mapped["status"] == "success" def test_mapping_quality_general_thresholds(): low_rate = evaluate_string_mapping_quality(input_count=20, mapped_count=9) assert low_rate["status"] == "failed" enough = evaluate_string_mapping_quality(input_count=20, mapped_count=10) assert enough["status"] == "success" def test_target_unsupported_with_reference_pep_requires_separate_ortholog_stage(): diagnostic = build_ppi_species_diagnostic( requested_ppi_species="auto", target_species="Ft", target_string_supported=False, reference_species="At", reference_data_source="/data/ref_genome/ath/Arabidopsis.pep.fa", ortholog_mapping_used=False, fallback_level="near_reference", fallback_reason="target_species_not_supported_by_string", ) assert diagnostic["status"] == "failed" assert diagnostic["diagnostics"]["issue_code"] == "ppi_requires_ortholog_mapping" assert diagnostic["diagnostics"]["required_next_stage"] == "ortholog" assert diagnostic["parameters"]["ppi_species_resolution"]["selected_string_species"] == "" def test_target_unsupported_without_reference_data_requests_reference_input(): diagnostic = build_ppi_species_diagnostic( requested_ppi_species="auto", target_species="Ft", target_string_supported=False, reference_species="", reference_data_source="", ortholog_mapping_used=False, fallback_level="reference_data_missing", fallback_reason="target_species_not_supported_by_string", ) assert diagnostic["status"] == "failed" assert diagnostic["diagnostics"]["issue_code"] == "ppi_reference_data_missing" assert diagnostic["diagnostics"]["required_next_stage"] == "ortholog" actions = diagnostic["diagnostics"]["recommended_actions"] assert [action["id"] for action in actions] == [ "upload_reference_protein", "prepare_reference_download", "explicit_arabidopsis_last_resort", ] assert actions[-1]["requires_confirmation"] is True assert actions[-1]["risk"] == "low_confidence_model_reference" def test_reference_protein_candidates_include_data_ref_genome_roots(): candidates = reference_protein_candidate_paths("At", roots=[Path("/data/ref_genome"), Path("/data")]) as_text = [str(path) for path in candidates] assert "/data/ref_genome/At/At.pep.fasta" in as_text assert "/data/ref_genome/ath/ath.pep.fa" in as_text assert "/data/At/At.pep.fasta" in as_text def test_frontend_ppi_defaults_do_not_force_ath(): html = (REPO_ROOT / "apps/web/vue3/index.html").read_text(encoding="utf-8") assert "ppiSpecies: 'auto'" in html assert "ppiSpecies = 'ath'" not in html assert "ppiSpecies || 'ath'" not in html assert "confirmParams.ppiSpecies || 'ath'" not in html assert "stageParams.ppiSpecies || confirmParams.ppiSpecies || 'ath'" not in html def test_ppi_skill_metadata_marks_plots_optional_and_string_png_default(): config = yaml.safe_load((REPO_ROOT / "skills/ppi/config.yaml").read_text(encoding="utf-8")) files = config["output"]["files"] optional_files = config["output"].get("optional_files", []) assert "{prefix}.ppi_network.pdf" not in files assert "{prefix}.ppi_network.pdf" in optional_files assert "{prefix}.ppi_network.string.png" in optional_files skill_doc = (REPO_ROOT / "skills/ppi/SKILL.md").read_text(encoding="utf-8") assert "STRING PNG" in skill_doc assert "可选" in skill_doc def test_legacy_cli_and_skill_use_shared_species_policy(): string_api = (REPO_ROOT / "apps/cli/string_api.py").read_text(encoding="utf-8") skill_run = (REPO_ROOT / "skills/ppi/run.py").read_text(encoding="utf-8") backend = (REPO_ROOT / "apps/web/pipeline_backend.py").read_text(encoding="utf-8") assert "from gfa_core.ppi_string_policy import STRING_SPECIES_MAP" in string_api assert "from gfa_core.ppi_string_policy import" in skill_run assert "STRING_SPECIES_MAP" in skill_run assert "from gfa_core.ppi_string_policy import" in backend assert "species_to_string_code as _shared_species_to_string_code" in backend assert 'f"{prefix}_id_resolution.tsv": f"{prefix}.ppi_id_resolution.tsv"' in backend def test_ppi_skill_writes_id_resolution_report_and_manifest_metrics(tmp_path): module = load_ppi_skill_module() gene_file = tmp_path / "genes.txt" gene_file.write_text("Os01t0810800-01\nOs02t0000100\n", encoding="utf-8") class FakeStringClient: species_id = "39947" def __init__(self): self.mapping_calls = [] self.network_calls = [] def get_string_ids(self, identifiers): self.mapping_calls.append(list(identifiers)) selected = { "Os01g0810800": "39947.Os01g0810800", "Os02g0000100": "39947.Os02g0000100", } return [ { "queryIndex": index, "queryItem": identifier, "stringId": selected[identifier], "preferredName": identifier, "ncbiTaxonId": 39947, } for index, identifier in enumerate(identifiers) if identifier in selected ] def get_network(self, identifiers, required_score=200, network_type="functional"): self.network_calls.append((list(identifiers), required_score, network_type)) return [] client = FakeStringClient() result = module.PpiSkill( string_api_factory=lambda _species: client ).run( gene_ids_file=gene_file, species="osa", prefix="Os.CrR", output_dir=tmp_path / "ppi", required_score=200, ) id_report = tmp_path / "ppi" / "Os.CrR_id_resolution.tsv" assert id_report.exists() text = id_report.read_text(encoding="utf-8") assert "Os01t0810800-01\t\tOs01t0810800-01\t\tOs01g0810800" in text assert result["metrics"]["id_resolution"]["input_member_count"] == 2 assert result["metrics"]["id_resolution"]["raw_resolved_query_count"] == 2 assert result["metrics"]["id_resolution"]["resolved_member_count"] == 2 assert client.network_calls == [ (["39947.Os01g0810800", "39947.Os02g0000100"], 200, "functional") ] assert "id_resolution" in result["artifacts"]