test_server_tools.py•8.41 kB
from __future__ import annotations
import pytest
from uniprot_mcp import server
from uniprot_mcp.models.domain import MappingResult
@pytest.mark.asyncio
async def test_fetch_entry_tool_uses_parser(monkeypatch, load_fixture):
payload = load_fixture("entry_reviewed.json")
async def fake_load(accession: str, **_kwargs): # pragma: no cover - trivial
assert accession == "P12345"
return "P12345", payload
monkeypatch.setattr(server, "_load_entry", fake_load)
entry = await server.fetch_entry("P12345")
assert entry.protein_name == "Tumor protein p53"
assert entry.sequence and entry.sequence.length == 10
@pytest.mark.asyncio
async def test_fetch_entry_missing_returns_stub(monkeypatch):
async def fake_load(accession: str, **_kwargs):
return "P12345", {}
monkeypatch.setattr(server, "_load_entry", fake_load)
entry = await server.fetch_entry("P12345")
assert entry.accession == "P12345"
assert entry.reviewed is False
@pytest.mark.asyncio
async def test_fetch_entry_respects_optional_arguments(monkeypatch, load_fixture):
observed = {}
async def fake_load(accession: str, *, fields=None):
observed["fields"] = fields
return "P12345", load_fixture("entry_reviewed.json")
monkeypatch.setattr(server, "_load_entry", fake_load)
await server.fetch_entry("P12345", fields=["accession"])
assert observed["fields"] == ["accession"]
@pytest.mark.asyncio
async def test_fetch_entry_normalizes_input(monkeypatch):
def fake_validate(accession: str) -> str:
assert accession.strip() == "p12345"
return "P12345"
async def fake_fetch_entry_json(
client, accession: str, fields=None
): # pragma: no cover - trivial
assert accession == "P12345"
return {}
class DummyClient:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return False
monkeypatch.setattr(server, "_validate_accession", fake_validate)
monkeypatch.setattr(server, "new_client", lambda: DummyClient())
monkeypatch.setattr(server, "fetch_entry_json", fake_fetch_entry_json)
entry = await server.fetch_entry(" p12345 ")
assert entry.accession == "P12345"
@pytest.mark.asyncio
async def test_search_uniprot_tool(monkeypatch, load_fixture):
payload = load_fixture("search_results.json")
class DummyClient:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return False
async def fake_search_json(
_client,
*,
query: str,
size: int,
reviewed_only: bool,
fields=None,
sort=None,
include_isoform=None,
):
assert query == "kinase"
assert size == 5
assert reviewed_only is True
assert fields == ["accession"]
assert sort == "accession desc"
assert include_isoform is True
return payload
monkeypatch.setattr(server, "new_client", lambda: DummyClient())
monkeypatch.setattr(server, "search_json", fake_search_json)
hits = await server.search_uniprot(
"kinase",
size=5,
reviewed_only=True,
fields=["accession"],
sort="accession desc",
include_isoform=True,
)
assert len(hits) == 2
assert hits[0].accession == "P12345"
@pytest.mark.asyncio
async def test_search_uniprot_uses_light_fields_when_env(monkeypatch, load_fixture):
payload = load_fixture("search_results.json")
monkeypatch.setenv("UNIPROT_ENABLE_FIELDS", "1")
class DummyClient:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return False
async def fake_search_json(
_client,
*,
query: str,
size: int,
reviewed_only: bool,
fields=None,
sort=None,
include_isoform=None,
):
assert fields == [server.FIELDS_SEARCH_LIGHT]
return payload
monkeypatch.setattr(server, "new_client", lambda: DummyClient())
monkeypatch.setattr(server, "search_json", fake_search_json)
hits = await server.search_uniprot("kinase")
assert len(hits) == 2
@pytest.mark.asyncio
async def test_map_ids_tool(monkeypatch, load_fixture):
payload = load_fixture("mapping_result.json")
async def fake_start(*args, **kwargs):
return "job"
async def fake_poll(job_id: str, *, ctx=None):
assert job_id == "job"
return payload
monkeypatch.setattr(server, "start_id_mapping", fake_start)
monkeypatch.setattr(server, "_poll_mapping_job", fake_poll)
result = await server.map_ids("UniProtKB_AC-ID", "Ensembl", ["P12345", "Q00000"])
assert isinstance(result, MappingResult)
assert "P12345" in result.results
assert result.results["NO_MATCH"] == []
@pytest.mark.asyncio
async def test_map_ids_tool_with_empty_ids():
result = await server.map_ids("UniProtKB_AC-ID", "Ensembl", [])
assert result.results == {}
@pytest.mark.asyncio
async def test_fetch_entry_invalid_accession_raises():
with pytest.raises(ValueError):
await server.fetch_entry("not-a-valid-accession")
@pytest.mark.asyncio
async def test_fetch_entry_version_raises_guidance():
with pytest.raises(ValueError) as excinfo:
await server.fetch_entry("P12345", version="5")
assert "fetch_entry_flatfile" in str(excinfo.value)
@pytest.mark.asyncio
async def test_fetch_entry_flatfile_tool(monkeypatch):
class DummyClient:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb): # pragma: no cover - trivial
return False
async def fake_fetch_entry_flatfile(
_client, accession: str, version: str, *, format: str = "txt"
):
assert accession == "P12345"
assert version == "5"
assert format == "txt"
return "FLATFILE"
monkeypatch.setattr(server, "new_client", lambda: DummyClient())
monkeypatch.setattr(server, "fetch_entry_flatfile_raw", fake_fetch_entry_flatfile)
text = await server.fetch_entry_flatfile("P12345", "5")
assert text == "FLATFILE"
@pytest.mark.asyncio
async def test_fetch_entry_flatfile_tool_handles_missing(monkeypatch):
class DummyClient:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb): # pragma: no cover - trivial
return False
async def fake_fetch_entry_flatfile(
_client, accession: str, version: str, *, format: str = "txt"
):
return ""
monkeypatch.setattr(server, "new_client", lambda: DummyClient())
monkeypatch.setattr(server, "fetch_entry_flatfile_raw", fake_fetch_entry_flatfile)
text = await server.fetch_entry_flatfile("P12345", "5")
assert "No UniProt entry" in text
@pytest.mark.asyncio
async def test_map_ids_failure_status(monkeypatch):
async def fake_start(*args, **kwargs):
return "job"
async def fake_status(*args, **kwargs):
return {"status": "FAILED"}
async def fake_results(*args, **kwargs):
return {}
monkeypatch.setattr(server, "start_id_mapping", fake_start)
monkeypatch.setattr(server, "get_mapping_status", fake_status)
monkeypatch.setattr(server, "get_mapping_results", fake_results)
with pytest.raises(server.UniProtClientError):
await server.map_ids("UniProtKB_AC-ID", "Ensembl", ["P12345"])
@pytest.mark.asyncio
async def test_get_sequence_invalid_accession():
with pytest.raises(ValueError):
await server.get_sequence("invalid")
@pytest.mark.asyncio
async def test_get_sequence_normalizes_accession(monkeypatch):
class DummyClient:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return False
async def fake_fetch_sequence_json(_client, accession: str):
assert accession == "P12345"
return {"sequence": {"value": "AA", "length": 2}}
monkeypatch.setattr(server, "new_client", lambda: DummyClient())
monkeypatch.setattr(server, "fetch_sequence_json", fake_fetch_sequence_json)
sequence = await server.get_sequence("p12345")
assert sequence is not None
assert sequence.length == 2