UniProt MCP Server

from __future__ import annotations from typing import Any import httpx import pytest from uniprot_mcp.adapters import uniprot_client from uniprot_mcp.adapters.uniprot_client import FIELDS_SEQUENCE_MIN, UniProtClientError @pytest.mark.asyncio async def test_new_client_enables_redirects_and_trust_env(): async with uniprot_client.new_client() as client: assert client.follow_redirects is True assert client.trust_env is True timeout = client.timeout assert isinstance(timeout, httpx.Timeout) assert timeout.read == 20.0 @pytest.mark.asyncio async def test_fetch_entry_json_returns_empty_on_404(): def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(404) async with httpx.AsyncClient( transport=httpx.MockTransport(handler), base_url=uniprot_client.DEFAULT_BASE_URL, ) as client: payload = await uniprot_client.fetch_entry_json(client, "NOPE") assert payload == {} @pytest.mark.asyncio async def test_fetch_entry_json_retries_then_succeeds(monkeypatch): attempts = {"count": 0} def handler(request: httpx.Request) -> httpx.Response: if "retry" in request.url.path and attempts["count"] == 0: attempts["count"] += 1 return httpx.Response(500) return httpx.Response(200, json={"primaryAccession": "retry"}) monkeypatch.setenv("UNIPROT_ENABLE_FIELDS", "1") async with httpx.AsyncClient( transport=httpx.MockTransport(handler), base_url=uniprot_client.DEFAULT_BASE_URL, ) as client: payload = await uniprot_client.fetch_entry_json(client, "retry") assert payload["primaryAccession"] == "retry" assert attempts["count"] == 1 @pytest.mark.asyncio async def test_fetch_entry_json_raises_on_unexpected_error(): def handler(request: httpx.Request) -> httpx.Response: return httpx.Response(400) async with httpx.AsyncClient( transport=httpx.MockTransport(handler), base_url=uniprot_client.DEFAULT_BASE_URL ) as client: with pytest.raises(UniProtClientError): await uniprot_client.fetch_entry_json(client, "bad") @pytest.mark.asyncio async def test_start_id_mapping_extracts_job_id(): responses: dict[str, Any] = {"/idmapping/run": httpx.Response(200, json={"jobId": "abc123"})} def handler(request: httpx.Request) -> httpx.Response: return responses[request.url.path] async with httpx.AsyncClient( transport=httpx.MockTransport(handler), base_url=uniprot_client.DEFAULT_BASE_URL ) as client: job_id = await uniprot_client.start_id_mapping( client, from_db="UniProtKB_AC-ID", to_db="Ensembl", ids=["P12345"] ) assert job_id == "abc123" @pytest.mark.asyncio async def test_get_mapping_results_returns_dict(): def handler(request: httpx.Request) -> httpx.Response: if request.url.path.startswith("/idmapping/results"): return httpx.Response(200, json={"results": []}) return httpx.Response(200, json={}) async with httpx.AsyncClient( transport=httpx.MockTransport(handler), base_url=uniprot_client.DEFAULT_BASE_URL ) as client: status = await uniprot_client.get_mapping_status(client, "job") results = await uniprot_client.get_mapping_results(client, "job") assert status == {} assert results == {"results": []} @pytest.mark.asyncio async def test_fetch_entry_json_accepts_fields_and_version(monkeypatch): observed: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: observed.update(request.url.params) return httpx.Response(200, json={"primaryAccession": "P12345"}) monkeypatch.delenv("UNIPROT_ENABLE_FIELDS", raising=False) async with httpx.AsyncClient( transport=httpx.MockTransport(handler), base_url=uniprot_client.DEFAULT_BASE_URL ) as client: await uniprot_client.fetch_entry_json( client, "P12345", fields=["accession", "sequence"], version="last", ) assert observed["fields"] == "accession,sequence" assert observed["version"] == "last" @pytest.mark.asyncio async def test_search_json_supports_optional_parameters(monkeypatch): observed: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: observed.update(request.url.params) return httpx.Response(200, json={"results": []}) monkeypatch.delenv("UNIPROT_ENABLE_FIELDS", raising=False) async with httpx.AsyncClient( transport=httpx.MockTransport(handler), base_url=uniprot_client.DEFAULT_BASE_URL ) as client: await uniprot_client.search_json( client, query="kinase", size=25, reviewed_only=True, fields=["accession"], sort="accession desc", include_isoform=True, ) assert observed["fields"] == "accession" assert observed["sort"] == "accession desc" assert observed["includeIsoform"] == "true" assert "reviewed:true" in observed["query"] assert int(observed["size"]) == 25 @pytest.mark.asyncio async def test_search_json_clamps_size(monkeypatch): observed: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: observed.update(request.url.params) return httpx.Response(200, json={"results": []}) async with httpx.AsyncClient( transport=httpx.MockTransport(handler), base_url=uniprot_client.DEFAULT_BASE_URL ) as client: await uniprot_client.search_json(client, query="kinase", size=999) assert int(observed["size"]) == 500 @pytest.mark.asyncio async def test_fetch_sequence_json_uses_minimal_fields(monkeypatch): observed: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: observed.update(request.url.params) return httpx.Response(200, json={"sequence": {"value": "AA", "length": 2}}) monkeypatch.setenv("UNIPROT_ENABLE_FIELDS", "1") async with httpx.AsyncClient( transport=httpx.MockTransport(handler), base_url=uniprot_client.DEFAULT_BASE_URL ) as client: payload = await uniprot_client.fetch_sequence_json(client, "P12345") assert payload["sequence"]["value"] == "AA" assert observed["fields"] == FIELDS_SEQUENCE_MIN @pytest.mark.asyncio async def test_fetch_sequence_json_retries(monkeypatch): attempts = {"count": 0} def handler(request: httpx.Request) -> httpx.Response: if attempts["count"] == 0: attempts["count"] += 1 return httpx.Response(503) return httpx.Response(200, json={"sequence": {"value": "AA", "length": 2}}) monkeypatch.setenv("UNIPROT_ENABLE_FIELDS", "1") async with httpx.AsyncClient( transport=httpx.MockTransport(handler), base_url=uniprot_client.DEFAULT_BASE_URL ) as client: payload = await uniprot_client.fetch_sequence_json(client, "P12345") assert payload["sequence"]["length"] == 2 assert attempts["count"] == 1 @pytest.mark.asyncio async def test_fetch_entry_flatfile_sets_headers_and_params(monkeypatch): observed: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: observed["accept"] = request.headers.get("accept") observed["params"] = dict(request.url.params) return httpx.Response(200, text="ENTRY DATA") async with httpx.AsyncClient( transport=httpx.MockTransport(handler), base_url=uniprot_client.DEFAULT_BASE_URL ) as client: text = await uniprot_client.fetch_entry_flatfile(client, "P12345", "6") assert text == "ENTRY DATA" assert observed["accept"] == "text/plain;format=txt" assert observed["params"]["version"] == "6" assert observed["params"]["format"] == "txt" @pytest.mark.asyncio async def test_fetch_entry_flatfile_rejects_unknown_format(monkeypatch): async with httpx.AsyncClient( transport=httpx.MockTransport(lambda request: httpx.Response(200, text="ok")), base_url=uniprot_client.DEFAULT_BASE_URL, ) as client: with pytest.raises(ValueError): await uniprot_client.fetch_entry_flatfile(client, "P12345", "6", format="xml")