PlantAI-dev/tests/unit/test_ppi_target_identity.py

import asyncio
from pathlib import Path

from openpyxl import Workbook

from apps.web import pipeline_backend as backend


def _write_config(input_dir: Path, headers: list[str], rows: list[list[object]]) -> None:
    input_dir.mkdir(parents=True, exist_ok=True)
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = "config"
    sheet.append(headers)
    for row in rows:
        sheet.append(row)
    workbook.save(input_dir / "config.xlsx")


def test_job_identity_reads_unique_setaria_scientific_name(tmp_path: Path) -> None:
    input_dir = tmp_path / "input"
    _write_config(
        input_dir,
        ["Species_Code", "Species_Name", "File_Type"],
        [
            ["Sv", "Setaria_viridis", "pep"],
            ["Sv", "Setaria viridis", "cds"],
        ],
    )

    identity = backend.resolve_ppi_target_identity(input_dir, "Sv")

    assert identity["status"] == "resolved"
    assert identity["scientific_name"] == "Setaria viridis"
    assert identity["tree_taxon_id"] == ""
    assert identity["identity_source"] == "job_scientific_name"


def test_request_model_exposes_independent_target_identity_fields() -> None:
    from apps.web.vue3.main import PpiSpeciesResolutionRequest

    request = PpiSpeciesResolutionRequest(
        job_id="job_identity",
        prefix="Sv.HKL",
        targetScientificName="Setaria viridis",
        ppiTargetTreeTaxonId="4556",
    )

    assert request.targetScientificName == "Setaria viridis"
    assert request.ppiTargetTreeTaxonId == "4556"
    assert request.ppiSpecies == "auto"


def test_explicit_identity_reaches_backend_without_using_ppi_species(tmp_path: Path) -> None:
    input_dir = tmp_path / "input"
    workspace = tmp_path / "workspace"
    input_dir.mkdir()
    (workspace / "final").mkdir(parents=True)
    job = backend.JobInfo(
        job_id="job_identity",
        input_dir=str(input_dir),
        workspace_dir=str(workspace),
        logs_dir=str(workspace / "logs"),
        created_at="",
    )

    result = backend.resolve_ppi_species_resolution(
        job=job,
        prefix="Sv.HKL",
        ppi_species="auto",
        target_species=["Sv"],
        requested_stages=["ppi"],
        target_scientific_name="Setaria viridis",
        ppi_target_tree_taxon_id="4556",
    )

    assert result["requested_ppi_species"] == "auto"
    assert result["target_identity"]["scientific_name"] == "Setaria viridis"
    assert result["target_identity"]["tree_taxon_id"] == "4556"
    assert result["target_identity"]["identity_source"] == "explicit_taxon_id"


def test_taxon_id_column_aliases_and_priority(tmp_path: Path) -> None:
    expected_by_header = {
        "Taxon_ID": "4556",
        "TaxID": "4556",
        "NCBI_TaxID": "4556",
    }
    for header, expected in expected_by_header.items():
        input_dir = tmp_path / header
        _write_config(
            input_dir,
            ["Species_Code", header],
            [["Sv", 4556]],
        )
        identity = backend.resolve_ppi_target_identity(input_dir, "Sv")
        assert identity["tree_taxon_id"] == expected
        assert identity["identity_source"] == "job_taxon_id"

    priority_dir = tmp_path / "priority"
    _write_config(
        priority_dir,
        ["Species_Code", "Taxon_ID", "TaxID", "NCBI_TaxID"],
        [["Sv", 4556, 4555, 4540]],
    )
    priority = backend.resolve_ppi_target_identity(priority_dir, "Sv")
    assert priority["tree_taxon_id"] == "4556"


def test_explicit_taxon_conflict_with_job_config_is_structured(tmp_path: Path) -> None:
    input_dir = tmp_path / "input"
    _write_config(
        input_dir,
        ["Species_Code", "Species_Name", "TaxID"],
        [["Sv", "Setaria viridis", 4556]],
    )

    identity = backend.resolve_ppi_target_identity(
        input_dir,
        "Sv",
        explicit_tree_taxon_id="4555",
    )

    assert identity["status"] == "conflict"
    assert identity["issue_code"] == "target_taxon_id_conflict"
    assert identity["tree_taxon_id"] == ""
    assert identity["explicit_tree_taxon_id"] == "4555"
    assert identity["config_tree_taxon_id"] == "4556"


def test_explicit_scientific_name_conflict_never_mixes_with_job_taxon(tmp_path: Path) -> None:
    input_dir = tmp_path / "input"
    _write_config(
        input_dir,
        ["Species_Code", "Species_Name", "TaxID"],
        [["Sv", "Setaria viridis", 4556]],
    )

    identity = backend.resolve_ppi_target_identity(
        input_dir,
        "Sv",
        explicit_scientific_name="Setaria_italica",
    )

    assert identity["status"] == "conflict"
    assert identity["issue_code"] == "target_scientific_name_conflict"
    assert identity["scientific_name"] == ""
    assert identity["tree_taxon_id"] == ""
    assert identity["explicit_scientific_name"] == "Setaria italica"
    assert identity["config_scientific_name"] == "Setaria viridis"


def test_species_resolution_api_surfaces_scientific_name_conflict(
    tmp_path: Path,
    monkeypatch,
) -> None:
    from apps.web.vue3 import main

    job_id = "job_identity_conflict"
    input_root = tmp_path / "input_root"
    workspace_root = tmp_path / "workspace_root"
    input_dir = input_root / job_id
    (workspace_root / job_id / "final").mkdir(parents=True)
    _write_config(
        input_dir,
        ["Species_Code", "Species_Name", "TaxID"],
        [["Sv", "Setaria viridis", 4556]],
    )
    monkeypatch.setattr(main, "INPUT_ROOT", input_root)
    monkeypatch.setattr(main, "WORKSPACE_ROOT", workspace_root)

    response = asyncio.run(
        main.resolve_ppi_species_resolution_api(
            main.PpiSpeciesResolutionRequest(
                job_id=job_id,
                prefix="Sv.HKL",
                targetSpecies=["Sv"],
                targetScientificName="Setaria italica",
                requestedStages=["ppi"],
            )
        )
    )

    assert response["target_identity"]["status"] == "conflict"
    assert response["target_identity"]["issue_code"] == "target_scientific_name_conflict"
    assert response["target_identity"]["scientific_name"] == ""
    assert response["target_identity"]["tree_taxon_id"] == ""


def test_only_unknown_short_code_does_not_guess_identity(tmp_path: Path) -> None:
    input_dir = tmp_path / "input"
    input_dir.mkdir()

    identity = backend.resolve_ppi_target_identity(input_dir, "Sv")

    assert identity["status"] == "unresolved"
    assert identity["issue_code"] == "target_string_tree_unresolved"
    assert identity["scientific_name"] == ""
    assert identity["tree_taxon_id"] == ""


def test_os_and_ft_keep_curated_compatibility_identity(tmp_path: Path) -> None:
    input_dir = tmp_path / "input"
    input_dir.mkdir()

    rice = backend.resolve_ppi_target_identity(input_dir, "Os")
    buckwheat = backend.resolve_ppi_target_identity(input_dir, "Ft")

    assert rice["tree_taxon_id"] == "39947"
    assert rice["identity_source"] == "project_alias"
    assert buckwheat["tree_taxon_id"] == "62330"
    assert buckwheat["identity_source"] == "project_alias"