PubMed MCP Server

""" Extended test cases for PubMed client functionality. """ from datetime import datetime from unittest.mock import Mock, patch import httpx import pytest from src.models import ArticleType, DateRange, SortOrder from src.pubmed_client import PubMedClient class TestPubMedClientExtended: """Extended tests for PubMed client to improve coverage.""" @pytest.fixture def client(self): """Create a PubMed client for testing.""" return PubMedClient( api_key="test_api_key", email="test@example.com", rate_limit=10.0, # High rate limit for tests ) @pytest.fixture def mock_response_xml(self): """Create a mock XML response for testing.""" return """<?xml version="1.0" ?> <eSearchResult> <Count>1</Count> <RetMax>1</RetMax> <RetStart>0</RetStart> <QueryKey>1</QueryKey> <WebEnv>test_webenv</WebEnv> <IdList> <Id>12345678</Id> </IdList> </eSearchResult>""" @pytest.fixture def mock_detailed_xml(self): """Create a mock detailed XML response for testing.""" return """<?xml version="1.0" ?> <PubmedArticleSet> <PubmedArticle> <MedlineCitation> <PMID>12345678</PMID> <Article PubModel="Electronic"> <Journal> <ISSN IssnType="Electronic">1234-5678</ISSN> <JournalIssue CitedMedium="Internet"> <Volume>25</Volume> <Issue>3</Issue> <PubDate> <Year>2023</Year> <Month>03</Month> <Day>15</Day> </PubDate> </JournalIssue> <Title>Journal of Cancer Research</Title> <ISOAbbreviation>J Cancer Res</ISOAbbreviation> </Journal> <ArticleTitle>A Sample Research Article on Cancer Treatment</ArticleTitle> <Abstract> <AbstractText>This is a sample abstract describing cancer treatment research.</AbstractText> </Abstract> <AuthorList CompleteYN="Y"> <Author ValidYN="Y"> <LastName>Smith</LastName> <ForeName>John</ForeName> <Initials>J</Initials> <AffiliationInfo> <Affiliation>University Hospital</Affiliation> </AffiliationInfo> </Author> <Author ValidYN="Y"> <LastName>Doe</LastName> <ForeName>Jane</ForeName> <Initials>J</Initials> </Author> </AuthorList> <Language>eng</Language> <PublicationTypeList> <PublicationType UI="D016428">Journal Article</PublicationType> </PublicationTypeList> </Article> </MedlineCitation> <PubmedData> <History> <PubMedPubDate PubStatus="received"> <Year>2023</Year> <Month>01</Month> <Day>01</Day> </PubMedPubDate> </History> <ArticleIdList> <ArticleId IdType="pubmed">12345678</ArticleId> <ArticleId IdType="doi">10.1000/example.doi</ArticleId> <ArticleId IdType="pmc">PMC1234567</ArticleId> </ArticleIdList> </PubmedData> </PubmedArticle> </PubmedArticleSet>""" def create_mock_search_response(self, pmids=None, count="1"): """Helper to create mock search response.""" if pmids is None: pmids = ["12345678"] mock_response = Mock() mock_response.json.return_value = {"esearchresult": {"idlist": pmids, "count": count}} return mock_response def create_mock_link_response(self, pmids=None): """Helper to create mock link response.""" if pmids is None: pmids = ["12345678"] mock_response = Mock() mock_response.json.return_value = { "linksets": [{"linksetdbs": [{"linkname": "pubmed_pubmed", "links": pmids}]}] } return mock_response @pytest.mark.asyncio async def test_search_with_date_range_enum(self, client, mock_response_xml): """Test search with DateRange enum values.""" # Create a proper mock response object mock_response = Mock() mock_response.json.return_value = {"esearchresult": {"idlist": ["12345678"], "count": "1"}} with patch.object(client, "_make_request", return_value=mock_response): with patch.object(client, "_fetch_article_details", return_value=[]): result = await client.search_articles( query="cancer", date_range=DateRange.LAST_5_YEARS, max_results=10, ) assert result.total_results == 1 assert result.returned_results == 0 # No articles since we mock empty list @pytest.mark.asyncio async def test_search_with_custom_date_range(self, client, mock_response_xml): """Test search with custom date range.""" mock_response = self.create_mock_search_response() with patch.object(client, "_make_request", return_value=mock_response): with patch.object(client, "_fetch_article_details", return_value=[]): result = await client.search_articles( query="cancer", date_from="2020/01/01", date_to="2023/12/31", max_results=10, ) assert result.total_results == 1 @pytest.mark.asyncio async def test_search_with_article_types(self, client, mock_response_xml): """Test search with specific article types.""" mock_response = self.create_mock_search_response() with patch.object(client, "_make_request", return_value=mock_response): with patch.object(client, "_fetch_article_details", return_value=[]): result = await client.search_articles( query="cancer", article_types=[ArticleType.REVIEW, ArticleType.META_ANALYSIS], max_results=10, ) assert result.total_results == 1 @pytest.mark.asyncio async def test_search_with_sort_order(self, client, mock_response_xml): """Test search with different sort orders.""" mock_response = self.create_mock_search_response() with patch.object(client, "_make_request", return_value=mock_response): with patch.object(client, "_fetch_article_details", return_value=[]): result = await client.search_articles( query="cancer", sort_order=SortOrder.PUBLICATION_DATE, max_results=10, ) assert result.total_results == 1 @pytest.mark.asyncio async def test_search_with_all_parameters(self, client, mock_response_xml): """Test search with all parameters specified.""" mock_response = self.create_mock_search_response() with patch.object(client, "_make_request", return_value=mock_response): with patch.object(client, "_fetch_article_details", return_value=[]): result = await client.search_articles( query="cancer treatment", max_results=50, sort_order=SortOrder.RELEVANCE, date_range=DateRange.LAST_YEAR, article_types=[ArticleType.JOURNAL_ARTICLE], authors=["Smith J"], journals=["Nature"], mesh_terms=["Neoplasms"], humans_only=True, has_abstract=True, has_full_text=False, language="eng", ) assert result.total_results == 1 @pytest.mark.asyncio async def test_get_article_details_comprehensive(self, client, mock_detailed_xml): """Test getting comprehensive article details.""" # Create a proper mock response object with .text property mock_response = Mock() mock_response.text = mock_detailed_xml with patch.object(client, "_make_request", return_value=mock_response): articles = await client.get_article_details( pmids=["12345678"], include_abstracts=True, include_citations=True, ) assert len(articles) == 1 article = articles[0] assert article.pmid == "12345678" assert article.title == "A Sample Research Article on Cancer Treatment" assert len(article.authors) == 2 assert article.authors[0].last_name == "Smith" assert article.authors[0].affiliation == "University Hospital" @pytest.mark.asyncio async def test_search_by_author_comprehensive(self, client, mock_response_xml): """Test comprehensive author search.""" mock_response = self.create_mock_search_response() with patch.object(client, "_make_request", return_value=mock_response): with patch.object(client, "_fetch_article_details", return_value=[]): result = await client.search_by_author( author_name="Smith J", max_results=20, include_coauthors=True, ) assert result.total_results == 1 @pytest.mark.asyncio async def test_find_related_articles(self, client, mock_response_xml): """Test finding related articles.""" mock_response = self.create_mock_link_response() with patch.object(client, "_make_request", return_value=mock_response): with patch.object(client, "_fetch_article_details", return_value=[]): result = await client.find_related_articles( pmid="12345678", max_results=10, ) assert result.total_results == 1 @pytest.mark.asyncio async def test_make_request_with_retry(self, client): """Test make_request error handling.""" with patch("httpx.AsyncClient.get") as mock_get: # Create a mock response that will raise HTTPStatusError when raise_for_status is called mock_response = Mock() mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( "Rate limited", request=Mock(), response=Mock(status_code=429) ) mock_get.return_value = mock_response # Test that the error is properly raised with pytest.raises(httpx.HTTPStatusError, match="Rate limited"): await client._make_request("esearch.fcgi", {"param": "value"}) @pytest.mark.asyncio async def test_make_request_with_timeout(self, client): """Test make_request with timeout.""" with patch("httpx.AsyncClient.get") as mock_get: mock_get.side_effect = httpx.TimeoutException("Request timeout") with pytest.raises(httpx.TimeoutException): await client._make_request("esearch.fcgi", {"param": "value"}) @pytest.mark.asyncio async def test_make_request_with_connection_error(self, client): """Test make_request with connection error.""" with patch("httpx.AsyncClient.get") as mock_get: mock_get.side_effect = httpx.ConnectError("Connection failed") with pytest.raises(httpx.ConnectError): await client._make_request("esearch.fcgi", {"param": "value"}) @pytest.mark.asyncio async def test_close_client(self, client): """Test closing the HTTP client.""" with patch.object(client.client, "aclose") as mock_close: await client.close() mock_close.assert_called_once() @pytest.mark.asyncio async def test_close_client_with_error(self, client): """Test closing client when aclose raises an error.""" with patch.object(client.client, "aclose", side_effect=Exception("Close error")): # Should not raise exception - the close method should handle errors gracefully try: await client.close() except Exception: pytest.fail("close() should handle errors gracefully") @pytest.mark.asyncio async def test_rate_limiting_functionality(self, client): """Test that rate limiting is enforced.""" # Create a client with very low rate limit for testing slow_client = PubMedClient( api_key="test_key", email="test@example.com", rate_limit=10.0, # 10 requests per second - fast enough for testing ) with patch.object(slow_client, "_make_request", return_value=Mock(text="<test>")): start_time = datetime.now() # Make two requests await slow_client._make_request("esearch.fcgi", {}) await slow_client._make_request("esearch.fcgi", {}) elapsed = (datetime.now() - start_time).total_seconds() # Should take some time due to rate limiting, but not too much for testing assert elapsed >= 0 # Just ensure it doesn't fail