PlantAI-dev/tests/unit/test_ppi_numeric_taxon_task4.py

from __future__ import annotations

import json
from pathlib import Path

import pytest
from openpyxl import Workbook

from apps.web import pipeline_backend as backend


def _job(tmp_path: Path, *, code: str = "", name: str = "", taxon_id: str = "") -> backend.JobInfo:
    input_dir = tmp_path / "input"
    workspace = tmp_path / "workspace"
    input_dir.mkdir(parents=True)
    (workspace / "final").mkdir(parents=True)
    (workspace / "logs").mkdir(parents=True)
    if code:
        workbook = Workbook()
        sheet = workbook.active
        sheet.title = "config"
        sheet.append(["Species_Code", "Species_Name", "TaxID"])
        sheet.append([code, name, taxon_id])
        workbook.save(input_dir / "config.xlsx")
    return backend.JobInfo(
        job_id="job_task4",
        input_dir=str(input_dir),
        workspace_dir=str(workspace),
        logs_dir=str(workspace / "logs"),
        created_at="",
    )


def test_numeric_taxon_requires_snapshot_registry_membership() -> None:
    from skills.ppi.run import StringAPI

    assert StringAPI.resolve_species_id("3708") == "3708"
    with pytest.raises(ValueError, match="STRING snapshot registry"):
        StringAPI.resolve_species_id("999999999")


def test_preflight_taxon_survives_execute_worker_and_backend_command(
    tmp_path: Path,
    monkeypatch,
) -> None:
    from apps.web.vue3.main import ExecuteRequest
    from tasks import normalize_stage_params

    job = _job(tmp_path, code="Bn", name="Brassica napus", taxon_id="3708")
    final_dir = Path(job.workspace_dir) / "final"
    (final_dir / "Bn.HKL.final_members.id").write_text("BnaA01g01000D\n", encoding="utf-8")
    scope_file = final_dir / "Bn.HKL.target.scope.json"
    scope_file.write_text(
        json.dumps(
            {
                "family": "HKL",
                "prefix": "Bn.HKL",
                "scope_id": "HKL.target-Bn.target_only",
                "scope_mode": "target_only",
                "target_species": ["Bn"],
                "reference_species": [],
                "selected_species": ["Bn"],
            }
        ),
        encoding="utf-8",
    )
    preflight = backend.resolve_ppi_species_resolution(
        job=job,
        prefix="Bn.HKL",
        target_species=["Bn"],
        requested_stages=["ppi"],
    )
    request = ExecuteRequest(
        stage="ppi",
        job_id=job.job_id,
        prefix="Bn.HKL",
        scopeFile=str(scope_file),
        ppiSpecies=preflight["selected_taxon_id"],
        normalizationSpeciesLabel="Bn",
    )
    worker_params, _profile = normalize_stage_params("ppi", request.model_dump())
    assert preflight["selected_taxon_id"] == "3708"
    assert request.ppiSpecies == "3708"
    assert worker_params["ppiSpecies"] == "3708"
    assert worker_params["normalizationSpeciesLabel"] == "Bn"

    captured: dict[str, list[str]] = {}

    def fake_run_cmd_logged(cmd, _log_path, cwd=None):
        captured["cmd"] = list(cmd)
        return 1

    monkeypatch.setattr(backend, "can_start_new_task", lambda: (True, "ok"))
    monkeypatch.setattr(backend, "run_cmd_logged", fake_run_cmd_logged)
    backend.run_stage(
        "ppi",
        job,
        "Bn.HKL",
        scope_file=str(scope_file),
        ppi_species=worker_params["ppiSpecies"],
        normalization_species_label=worker_params["normalizationSpeciesLabel"],
    )

    command = captured["cmd"]
    assert command[command.index("--species") + 1] == "3708"
    assert command[command.index("--normalization-species-label") + 1] == "Bn"


def test_numeric_taxon_uses_independent_os_normalization_and_records_manifest(
    tmp_path: Path,
) -> None:
    from skills.ppi.run import PpiSkill

    probes: list[str] = []

    class FakeApi:
        species_id = "39947"

        def get_string_ids(self, identifiers):
            probes.extend(identifiers)
            normalized_index = identifiers.index("Os01g0810800")
            return [
                {
                    "queryIndex": normalized_index,
                    "queryItem": "Os01g0810800",
                    "stringId": "39947.Os01g0810800",
                    "preferredName": "Os01g0810800",
                    "ncbiTaxonId": 39947,
                }
            ]

        def get_network(self, proteins, required_score=200):
            assert proteins == ["39947.Os01g0810800"]
            return []

    gene_file = tmp_path / "genes.txt"
    output_dir = tmp_path / "output"
    gene_file.write_text("Os01t0810800-01\n", encoding="utf-8")
    skill = PpiSkill(string_api_factory=lambda _species: FakeApi())

    manifest = skill.run(
        gene_ids_file=gene_file,
        species="39947",
        normalization_species_label="osa",
        prefix="Os.CRR",
        output_dir=output_dir,
    )

    assert "Os01t0810800-01" in probes
    assert "Os01g0810800" in probes
    assert manifest["status"] == "success"
    assert manifest["input"]["species"] == "39947"
    assert manifest["parameters"]["string_taxon_id"] == "39947"
    assert manifest["parameters"]["normalization_species_label"] == "osa"
    saved = json.loads((output_dir / "Os.CRR.manifest.json").read_text(encoding="utf-8"))
    assert saved["parameters"] == manifest["parameters"]


def test_legacy_species_code_remains_compatible() -> None:
    from skills.ppi.run import StringAPI

    assert StringAPI.resolve_species_id("ath") == "3702"
    assert StringAPI.resolve_species_id("osa") == "39947"


def test_skill_config_and_contract_describe_taxon_and_normalization_channels() -> None:
    project_root = Path(__file__).resolve().parents[2]
    config = (project_root / "skills/ppi/config.yaml").read_text(encoding="utf-8")
    contract = (project_root / "gfa_core/schemas/stages/ppi_contract_v1.md").read_text(encoding="utf-8")

    assert "normalization_species_label" in config
    assert "STRING TaxID" in config
    assert "normalizationSpeciesLabel" in contract
    assert "任意数字" in contract