Spaces:
Running
Running
File size: 7,492 Bytes
59ba48b 4939f56 59ba48b 4939f56 59ba48b 4939f56 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 | """
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"
|