PlantAI-dev/tests/unit/test_effective_member_resolver.py

import json
from pathlib import Path

import pytest

from gfa_core.mas.effective_members import (
    MemberResolutionError,
    resolve_effective_members,
)


def _write(path: Path, text: str) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")
    return path


def _pep(ids: list[str]) -> str:
    return "".join(f">{gene_id}\nM{idx}A\n" for idx, gene_id in enumerate(ids, start=1))


def _confirmed_manifest(final_dir: Path, prefix: str, ids: list[str]) -> None:
    manifest = {
        "stage": "subfamily_confirm",
        "status": "success",
        "prefix": prefix,
        "parameters": {
            "final_members_source_stage": "subfamily_confirm",
            "final_members_promoted": True,
        },
        "artifacts": {
            "final_members_id": {"full_path": str(final_dir / f"{prefix}.final_members.id")},
            "final_members_pep": {"full_path": str(final_dir / f"{prefix}.final_members.pep.fa")},
        },
        "metrics": {"final_members_count": len(ids)},
    }
    _write(final_dir / f"{prefix}.subfamily_confirm.manifest.json", json.dumps(manifest))


def test_post_confirm_required_uses_confirmed_final_not_legacy_identify(tmp_path: Path):
    ws = tmp_path / "workspace" / "job1"
    final_dir = ws / "final"
    input_dir = tmp_path / "input" / "job1"
    prefix = "Target.Os.CrR"

    legacy_ids = [f"legacy{i}" for i in range(500)]
    confirmed_ids = [f"confirmed{i}" for i in range(20)]
    _write(final_dir / f"{prefix}.identified_members.id", "\n".join(legacy_ids) + "\n")
    _write(final_dir / f"{prefix}.identified_members.pep.fa", _pep(legacy_ids))
    _write(final_dir / f"{prefix}.final_members.id", "\n".join(confirmed_ids) + "\n")
    _write(final_dir / f"{prefix}.final_members.pep.fa", _pep(confirmed_ids))
    _confirmed_manifest(final_dir, prefix, confirmed_ids)

    result = resolve_effective_members(
        workspace=ws,
        final_dir=final_dir,
        input_dir=input_dir,
        prefix=prefix,
        phase="post_confirm_required",
        require="pair",
    )

    assert result.path_id == final_dir / f"{prefix}.final_members.id"
    assert result.path_pep == final_dir / f"{prefix}.final_members.pep.fa"
    assert result.source_stage == "subfamily_confirm"
    assert result.source_kind == "confirmed_final_members"
    assert result.member_count == 20


def test_post_confirm_required_fails_without_confirmed_provenance(tmp_path: Path):
    ws = tmp_path / "workspace" / "job1"
    final_dir = ws / "final"
    input_dir = tmp_path / "input" / "job1"
    prefix = "Target.Os.CrR"

    ids = ["a", "b"]
    _write(final_dir / f"{prefix}.identified_members.id", "\n".join(ids) + "\n")
    _write(final_dir / f"{prefix}.identified_members.pep.fa", _pep(ids))

    with pytest.raises(MemberResolutionError, match="confirmed final members"):
        resolve_effective_members(
            workspace=ws,
            final_dir=final_dir,
            input_dir=input_dir,
            prefix=prefix,
            phase="post_confirm_required",
            require="pair",
        )


def test_pre_confirm_prefers_curated_before_identified(tmp_path: Path):
    ws = tmp_path / "workspace" / "job1"
    final_dir = ws / "final"
    input_dir = tmp_path / "input" / "job1"
    prefix = "Target.Os.CrR"

    identified_ids = ["id1", "id2", "id3"]
    curated_ids = ["id1", "id3"]
    _write(final_dir / f"{prefix}.identified_members.id", "\n".join(identified_ids) + "\n")
    _write(final_dir / f"{prefix}.identified_members.pep.fa", _pep(identified_ids))
    _write(final_dir / f"{prefix}.curated_members.id", "\n".join(curated_ids) + "\n")
    _write(final_dir / f"{prefix}.curated_members.pep.fa", _pep(curated_ids))

    result = resolve_effective_members(
        workspace=ws,
        final_dir=final_dir,
        input_dir=input_dir,
        prefix=prefix,
        phase="pre_confirm",
        require="pair",
    )

    assert result.path_id == final_dir / f"{prefix}.curated_members.id"
    assert result.path_pep == final_dir / f"{prefix}.curated_members.pep.fa"
    assert result.source_kind == "curated_members"
    assert result.member_count == 2


def test_pre_confirm_accepts_same_curated_members_in_different_order(tmp_path: Path):
    ws = tmp_path / "workspace" / "job1"
    final_dir = ws / "final"
    input_dir = tmp_path / "input" / "job1"
    prefix = "Target.Sl.GRAS"

    _write(final_dir / f"{prefix}.curated_members.id", "gene1\ngene2\n")
    _write(final_dir / f"{prefix}.curated_members.pep.fa", _pep(["gene2", "gene1"]))

    result = resolve_effective_members(
        workspace=ws,
        final_dir=final_dir,
        input_dir=input_dir,
        prefix=prefix,
        phase="pre_confirm",
        require="pair",
    )

    assert result.member_count == 2
    assert result.path_id == final_dir / f"{prefix}.curated_members.id"
    assert result.path_pep == final_dir / f"{prefix}.curated_members.pep.fa"


def test_pre_confirm_rejects_different_curated_members_with_equal_counts(tmp_path: Path):
    ws = tmp_path / "workspace" / "job1"
    final_dir = ws / "final"
    input_dir = tmp_path / "input" / "job1"
    prefix = "Target.Sl.GRAS"

    _write(final_dir / f"{prefix}.curated_members.id", "gene1\ngene2\n")
    _write(final_dir / f"{prefix}.curated_members.pep.fa", _pep(["gene1", "gene3"]))

    with pytest.raises(MemberResolutionError, match="ID/FASTA mismatch"):
        resolve_effective_members(
            workspace=ws,
            final_dir=final_dir,
            input_dir=input_dir,
            prefix=prefix,
            phase="pre_confirm",
            require="pair",
        )


def test_subfamily_evidence_required_rejects_unreviewed_identified_members(tmp_path: Path):
    ws = tmp_path / "workspace" / "job1"
    final_dir = ws / "final"
    input_dir = tmp_path / "input" / "job1"
    prefix = "Target.Os.CrR"

    identified_ids = ["id1", "id2", "id3"]
    _write(final_dir / f"{prefix}.identified_members.id", "\n".join(identified_ids) + "\n")
    _write(final_dir / f"{prefix}.identified_members.pep.fa", _pep(identified_ids))

    with pytest.raises(MemberResolutionError, match="member review"):
        resolve_effective_members(
            workspace=ws,
            final_dir=final_dir,
            input_dir=input_dir,
            prefix=prefix,
            phase="subfamily_evidence_required",  # type: ignore[arg-type]
            require="pair",
        )


def test_subfamily_evidence_required_uses_only_curated_members(tmp_path: Path):
    ws = tmp_path / "workspace" / "job1"
    final_dir = ws / "final"
    input_dir = tmp_path / "input" / "job1"
    prefix = "Target.Os.CrR"

    curated_ids = ["id1", "id3"]
    final_ids = ["id1", "id2", "id3", "id4"]
    _write(final_dir / f"{prefix}.curated_members.id", "\n".join(curated_ids) + "\n")
    _write(final_dir / f"{prefix}.curated_members.pep.fa", _pep(curated_ids))
    _write(final_dir / f"{prefix}.final_members.id", "\n".join(final_ids) + "\n")
    _write(final_dir / f"{prefix}.final_members.pep.fa", _pep(final_ids))

    result = resolve_effective_members(
        workspace=ws,
        final_dir=final_dir,
        input_dir=input_dir,
        prefix=prefix,
        phase="subfamily_evidence_required",  # type: ignore[arg-type]
        require="pair",
    )

    assert result.path_id == final_dir / f"{prefix}.curated_members.id"
    assert result.path_pep == final_dir / f"{prefix}.curated_members.pep.fa"
    assert result.source_kind == "curated_members"
    assert result.member_count == 2


def test_legacy_compatible_reads_old_final_members_with_warning(tmp_path: Path):
    ws = tmp_path / "workspace" / "job1"
    final_dir = ws / "final"
    input_dir = tmp_path / "input" / "job1"
    prefix = "Target.Os.CrR"

    ids = ["old1", "old2"]
    _write(final_dir / f"{prefix}.final_members.id", "\n".join(ids) + "\n")
    _write(final_dir / f"{prefix}.final_members.pep.fa", _pep(ids))

    result = resolve_effective_members(
        workspace=ws,
        final_dir=final_dir,
        input_dir=input_dir,
        prefix=prefix,
        phase="legacy_compatible",
        require="pair",
    )

    assert result.source_kind == "legacy_identify_final_members"
    assert result.member_count == 2
    assert any("legacy" in warning for warning in result.warnings)


def test_pipeline_backend_member_helpers_use_effective_resolver(tmp_path: Path):
    from apps.web.pipeline_backend import resolve_single_target_gene_id_file, resolve_species_family_pep

    ws = tmp_path / "workspace" / "job1"
    final_dir = ws / "final"
    input_dir = tmp_path / "input" / "job1"
    prefix = "Target.Os.CrR"
    species = "Os"
    family = "CrR"

    identified_ids = [f"identified{i}" for i in range(500)]
    confirmed_ids = [f"confirmed{i}" for i in range(20)]
    _write(final_dir / f"{prefix}.identified_members.id", "\n".join(identified_ids) + "\n")
    _write(final_dir / f"{prefix}.identified_members.pep.fa", _pep(identified_ids))
    _write(final_dir / f"{prefix}.final_members.id", "\n".join(confirmed_ids) + "\n")
    _write(final_dir / f"{prefix}.final_members.pep.fa", _pep(confirmed_ids))
    _confirmed_manifest(final_dir, prefix, confirmed_ids)

    assert resolve_single_target_gene_id_file(ws, input_dir, final_dir, species, prefix) == (
        final_dir / f"{prefix}.final_members.id"
    )
    assert resolve_species_family_pep(ws, final_dir, input_dir, species, family) == (
        final_dir / f"{prefix}.final_members.pep.fa"
    )


def test_pipeline_backend_post_confirm_does_not_fallback_to_curated(tmp_path: Path):
    from apps.web.pipeline_backend import resolve_single_target_gene_id_file

    ws = tmp_path / "workspace" / "job1"
    final_dir = ws / "final"
    input_dir = tmp_path / "input" / "job1"
    prefix = "Target.Os.CrR"
    species = "Os"
    family = "CrR"

    curated_ids = ["c1", "c2"]
    _write(final_dir / f"{prefix}.curated_members.id", "\n".join(curated_ids) + "\n")
    _write(final_dir / f"{prefix}.curated_members.pep.fa", _pep(curated_ids))
    _confirmed_manifest(final_dir, prefix, curated_ids)

    assert resolve_single_target_gene_id_file(ws, input_dir, final_dir, species, prefix) is None