34047007@qq.com
62ca8fa6b8
Includes search engine improvements, Alembic migrations, new services (pubmed_daily_update, query_expansion), frontend updates, and documentation sync.
266 lines
10 KiB
Python
266 lines
10 KiB
Python
"""Advanced search engine service tests"""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
def _empty_mock():
|
|
"""Create a mock result that returns empty when iterated"""
|
|
m = MagicMock()
|
|
m.__iter__.return_value = iter([])
|
|
return m
|
|
|
|
|
|
def _atm_mocks(count: int = 2) -> list:
|
|
"""Return N empty mock results for ATM expansion queries"""
|
|
return [_empty_mock() for _ in range(count)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_empty_db():
|
|
"""Empty query on empty DB -> returns empty results"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
|
|
# Mock empty count and empty items
|
|
mock_count = MagicMock()
|
|
mock_count.scalar.return_value = 0
|
|
mock_items = MagicMock()
|
|
mock_items.scalars.return_value.all.return_value = []
|
|
|
|
db.execute.side_effect = [mock_count, mock_items]
|
|
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(db, query="", page=1, page_size=20)
|
|
assert result["total"] == 0
|
|
assert result["items"] == []
|
|
assert result["page"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_title_field():
|
|
"""Title field search generates correct ilike condition"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
condition = AdvancedSearchEngine._field_condition("title", "cancer", False)
|
|
assert condition is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_abstract_field():
|
|
"""Abstract field search generates correct ilike condition"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
condition = AdvancedSearchEngine._field_condition("abstract", "test", False)
|
|
assert condition is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_journal_field():
|
|
"""Journal field search"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
condition = AdvancedSearchEngine._field_condition("journal", "Nature", False)
|
|
assert condition is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_all_field():
|
|
"""'all' field searches title OR abstract"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
condition = AdvancedSearchEngine._field_condition("all", "cancer", False)
|
|
assert condition is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_exact_phrase():
|
|
"""Exact phrase disables % wildcards"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
# exact=True -> no % wrapping
|
|
condition = AdvancedSearchEngine._field_condition("title", "EGFR mutation", True)
|
|
assert condition is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_year_filter():
|
|
"""Year_from and year_to generate bounds"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
|
|
mock_count = MagicMock()
|
|
mock_count.scalar.return_value = 0
|
|
mock_items = MagicMock()
|
|
mock_items.scalars.return_value.all.return_value = []
|
|
|
|
db.execute.side_effect = [*_atm_mocks(), _empty_mock(), mock_count, mock_items]
|
|
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(
|
|
db, query="test", year_from=2020, year_to=2024, page=1, page_size=20
|
|
)
|
|
assert result["total"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_boolean_or():
|
|
"""OR boolean combines terms with or_"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
|
|
mock_count = MagicMock()
|
|
mock_count.scalar.return_value = 0
|
|
mock_items = MagicMock()
|
|
mock_items.scalars.return_value.all.return_value = []
|
|
|
|
db.execute.side_effect = [*_atm_mocks(6), _empty_mock(), mock_count, mock_items]
|
|
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(
|
|
db, query="cancer lung", boolean="or", page=1, page_size=20
|
|
)
|
|
assert result["total"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_date_range():
|
|
"""Date_from and date_to filters"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
|
|
mock_count = MagicMock()
|
|
mock_count.scalar.return_value = 0
|
|
mock_items = MagicMock()
|
|
mock_items.scalars.return_value.all.return_value = []
|
|
|
|
db.execute.side_effect = [*_atm_mocks(), _empty_mock(), mock_count, mock_items]
|
|
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(
|
|
db, query="test",
|
|
date_from="2024-01-01", date_to="2024-12-31",
|
|
page=1, page_size=20
|
|
)
|
|
assert result["total"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_page_size():
|
|
"""Custom page_size is respected"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
|
|
mock_count = MagicMock()
|
|
mock_count.scalar.return_value = 100
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.all.return_value = []
|
|
mock_items = MagicMock()
|
|
mock_items.scalars.return_value.all.return_value = []
|
|
|
|
# ATM (2), count, tag lookups (0), journal lookups (0), items
|
|
db.execute.side_effect = [*_atm_mocks(), _empty_mock(), mock_count, mock_result, mock_result, mock_items]
|
|
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(
|
|
db, query="test", page=2, page_size=5
|
|
)
|
|
assert result["page"] == 2
|
|
assert result["page_size"] == 5
|
|
|
|
|
|
|
|
# ── Retracted / negative result filter tests ──
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_retracted_yes():
|
|
"""retracted='yes' filter does not error"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
mock_count = MagicMock(); mock_count.scalar.return_value = 0
|
|
mock_items = MagicMock(); mock_items.scalars.return_value.all.return_value = []
|
|
db.execute.side_effect = [_empty_mock(), mock_count, mock_items]
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(db, retracted="yes", page=1, page_size=20)
|
|
assert result["total"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_retracted_no():
|
|
"""retracted='no' filter does not error"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
mock_count = MagicMock(); mock_count.scalar.return_value = 0
|
|
mock_items = MagicMock(); mock_items.scalars.return_value.all.return_value = []
|
|
db.execute.side_effect = [_empty_mock(), mock_count, mock_items]
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(db, retracted="no", page=1, page_size=20)
|
|
assert result["total"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_retracted_only():
|
|
"""retracted='only' filter does not error"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
mock_count = MagicMock(); mock_count.scalar.return_value = 0
|
|
mock_items = MagicMock(); mock_items.scalars.return_value.all.return_value = []
|
|
db.execute.side_effect = [_empty_mock(), mock_count, mock_items]
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(db, retracted="only", page=1, page_size=20)
|
|
assert result["total"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_negative_yes():
|
|
"""negative_result='yes' filter does not error"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
mock_count = MagicMock(); mock_count.scalar.return_value = 0
|
|
mock_items = MagicMock(); mock_items.scalars.return_value.all.return_value = []
|
|
db.execute.side_effect = [_empty_mock(), mock_count, mock_items]
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(db, negative_result="yes", page=1, page_size=20)
|
|
assert result["total"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_negative_no():
|
|
"""negative_result='no' filter does not error"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
mock_count = MagicMock(); mock_count.scalar.return_value = 0
|
|
mock_items = MagicMock(); mock_items.scalars.return_value.all.return_value = []
|
|
db.execute.side_effect = [_empty_mock(), mock_count, mock_items]
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(db, negative_result="no", page=1, page_size=20)
|
|
assert result["total"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_negative_only():
|
|
"""negative_result='only' filter does not error"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
mock_count = MagicMock(); mock_count.scalar.return_value = 0
|
|
mock_items = MagicMock(); mock_items.scalars.return_value.all.return_value = []
|
|
db.execute.side_effect = [_empty_mock(), mock_count, mock_items]
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(db, negative_result="only", page=1, page_size=20)
|
|
assert result["total"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_retracted_and_negative():
|
|
"""Combined retracted + negative_result filters do not error"""
|
|
from app.services.search_engine import AdvancedSearchEngine
|
|
db = AsyncMock()
|
|
mock_count = MagicMock(); mock_count.scalar.return_value = 0
|
|
mock_items = MagicMock(); mock_items.scalars.return_value.all.return_value = []
|
|
db.execute.side_effect = [_empty_mock(), mock_count, mock_items]
|
|
with patch("app.services.tag_loader.load_tags_for_literature", AsyncMock(return_value={})):
|
|
result = await AdvancedSearchEngine.search(
|
|
db, retracted="yes", negative_result="only", page=1, page_size=20
|
|
)
|
|
assert result["total"] == 0
|
|
|
|
|
|
__all__ = []
|