bioai-platform/backend/tests/test_identifier_resolution.py

"""
Unit tests for the multi-database UniProt identifier resolver.

Offline logic (regex, gene hints, query building, strategy ordering) runs
with mocked HTTP; the strategy ladder against the live UniProt REST API is
covered by a small, marked, network-dependent suite.
"""

import pytest

from app.services import identifier_resolution as ir


# ---------------------------------------------------------------------------
# Offline: accession format detection
# ---------------------------------------------------------------------------

@pytest.mark.parametrize("acc,expected", [
    ("P04637", True),
    ("Q9H3D4", True),
    ("O88898", True),
    ("A0A0S2Z4N5", True),
    ("A0A024RBG1", True),
    ("p04637", True),          # case-insensitive
    ("NP_003713", False),
    ("4X0Z", False),
    ("ENSP00000269305", False),
    ("", False),
    ("BOGUS", False),
    ("P0463", False),          # too short
    ("P046370", False),        # too long
])
def test_is_uniprot_accession(acc, expected):
    assert ir.is_uniprot_accession(acc) is expected


# ---------------------------------------------------------------------------
# Offline: description parsing helpers
# ---------------------------------------------------------------------------

def test_extract_organism_from_ncbi_bracket():
    assert ir.extract_organism("tumor protein p63 isoform 1 [Homo sapiens]") == "Homo sapiens"
    assert ir.extract_organism("no organism tag here") == ""


def test_extract_gene_hint_variants():
    assert ir.extract_gene_hint("gene=TP63") == "TP63"
    assert ir.extract_gene_hint("recName: Full=foo; short=TP63") == "TP63"
    assert ir.extract_gene_hint("tumor protein p63 isoform 1 [Homo sapiens]") == "p63"
    assert ir.extract_gene_hint("hypothetical protein [Mus musculus]") == ""
    assert ir.extract_gene_hint("") == ""


# ---------------------------------------------------------------------------
# Offline: clean fasta / query hygiene
# ---------------------------------------------------------------------------

def test_clean_fasta():
    assert ir._clean_fasta(">hdr\nMSQSI-HQS\nlower") == "MSQSIHQSLOWER"
    assert ir._clean_fasta(">sp|P04637|TP53_HUMAN p53\nMEQPSDK") == "MEQPSDK"


@pytest.mark.parametrize("seq,expected", [
    ("ACGTACGTACGTTGAC", True),
    ("AUGCGAUGCGA", True),
    ("ATGCNNN", True),
    ("MSQSIHQSLOWER", False),
    ("", False),
    ("---ACGT---", True),
])
def test_looks_nucleotide(seq, expected):
    assert ir._looks_nucleotide(seq) is expected


def test_pick_best_prefers_reviewed():
    rows = [
        {"entryType": "UniProtKB unreviewed (TrEMBL)", "primaryAccession": "A0A111"},
        {"entryType": "UniProtKB reviewed (Swiss-Prot)", "primaryAccession": "P04637"},
    ]
    assert ir._pick_best(rows) == "P04637"
    assert ir._pick_best([]) is None
    assert ir._pick_best([{"entryType": "x", "primaryAccession": ""}]) is None


# ---------------------------------------------------------------------------
# Offline: strategy ladder ordering + mocked HTTP
# ---------------------------------------------------------------------------

class FakeResp:
    def __init__(self, status_code, payload):
        self.status_code = status_code
        self._payload = payload

    def json(self):
        return self._payload


class FakeClient:
    """Serves xref:XX -> P99999, but nothing else."""

    def __init__(self, *args, **kwargs):
        self.calls = []

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        return None

    async def get(self, url, params=None, **kwargs):
        self.calls.append((url, params))
        if "idmapping/status" in url:
            # End the (normally slow) polling immediately in tests.
            return FakeResp(200, {"jobStatus": "ERROR"})
        q = (params or {}).get("query", "")
        if "xref:TP63HIT" in q:
            return FakeResp(200, {"results": [
                {"entryType": "UniProtKB reviewed (Swiss-Prot)", "primaryAccession": "Q9H3D4"}
            ]})
        return FakeResp(200, {"results": []})

    async def post(self, url, **kwargs):
        return FakeResp(200, {"jobId": "job-1"})

    async def put(self, url, **kwargs):
        return FakeResp(200, {})


@pytest.mark.asyncio
async def test_direct_accession_short_circuits(monkeypatch):
    called = []

    async def boom(*a, **k):
        called.append(True)
        raise AssertionError("should not hit network")

    monkeypatch.setattr(ir.httpx, "AsyncClient", boom)
    r = await ir.resolve_to_uniprot(accession="P04637")
    assert r == {"accession": "P04637", "method": "direct", "status": "resolved", "confidence": "identified"}
    assert not called


@pytest.mark.asyncio
async def test_xref_strategy_resolves_refseq(monkeypatch):
    fake = FakeClient()
    monkeypatch.setattr(ir.httpx, "AsyncClient", lambda *a, **k: fake)
    r = await ir.resolve_to_uniprot(accession="TP63HIT", description="tumor protein p63 [Homo sapiens]")
    assert r["method"] == "xref"
    assert r["accession"] == "Q9H3D4"


@pytest.mark.asyncio
async def test_sequence_strategy_only_when_requested(monkeypatch):
    """try_sequence=False must skip the EBI BLAST fallback entirely."""
    fake = FakeClient()

    def boom(*a, **k):
        raise AssertionError("sequence fallback must be skipped")

    monkeypatch.setattr(ir.httpx, "AsyncClient", lambda *a, **k: fake)
    monkeypatch.setattr(ir, "resolve_by_sequence", boom)
    # ZZ9999 isn't matched by FakeClient's xref stub, so xref+name both fail.
    r = await ir.resolve_to_uniprot(accession="ZZ9999", description="nothing", try_sequence=False)
    assert r["status"] == "unresolved"
    assert r["confidence"] == "de_novo"
    assert r["accession"] is None


@pytest.mark.asyncio
async def test_pdb_chain_suffix_stripped(monkeypatch):
    """4X0Z:A must be searched as xref:4X0Z (parent entry only)."""
    queries = []

    async def fake_search(query, **kwargs):
        queries.append(query)
        return [{"entryType": "UniProtKB reviewed (Swiss-Prot)", "primaryAccession": "P04637"}]

    monkeypatch.setattr(ir, "search_uniprot", fake_search)
    acc = await ir.resolve_by_xref("4X0Z:A")
    assert acc == "P04637"
    assert any("xref:4X0Z" in q for q in queries)
    assert all("4X0Z:A" not in q for q in queries)


@pytest.mark.asyncio
async def test_nucleotide_sequence_skips_sequence_blast(monkeypatch):
    """A nucleotide query must never trigger the protein BLAST fallback."""

    def boom(*a, **k):
        raise AssertionError("nucleotide query must not run protein BLAST")

    # If the guard fails, the real implementation would import and run BlastTool.
    monkeypatch.setattr("app.tools.blast.BlastTool", boom)
    acc = await ir.resolve_by_sequence("ACGTACGTACGTTGACGG")
    assert acc is None


@pytest.mark.asyncio
async def test_no_input_returns_unresolved():
    r = await ir.resolve_to_uniprot()
    assert r == ir.UNRESOLVED_RESULT
    assert (await ir.resolve_to_uniprot(accession="")) == ir.UNRESOLVED_RESULT


@pytest.mark.asyncio
async def test_sequence_method_maps_to_homolog_confidence(monkeypatch):
    """Tier-4 (EBI BLAST) hits are homologs, not database-grade identity."""

    async def fake_seq(seq):
        return "P04637"

    monkeypatch.setattr(ir, "resolve_by_sequence", fake_seq)
    r = await ir.resolve_to_uniprot(sequence="MEEPQSDPSVEP")
    assert r["method"] == "sequence"
    assert r["status"] == "resolved"
    assert r["confidence"] == "homolog"