Justin Visser
87aa89b58c
- Remove unused type:ignore comment in scholars.py - Fix 51 mypy errors across test files (Any return types, cast(), protocol stubs) - Fix arxiv unit test failures caused by session isolation (patch_session_factory fixture) - Fix integration test rate-limit bleed between search and create - Add scripts/premerge.sh for local CI verification - Add docs/website/.docusaurus/ to .gitignore Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
278 lines
8.8 KiB
Python
278 lines
8.8 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.services.scholar.parser import ParseState
|
|
from app.services.scholar.source import FetchResult
|
|
from app.services.scholars import application as scholar_service
|
|
|
|
pytestmark = [pytest.mark.integration, pytest.mark.db]
|
|
|
|
|
|
class StubScholarSource:
|
|
def __init__(self, fetch_results: list[FetchResult]) -> None:
|
|
self._fetch_results = list(fetch_results)
|
|
self.calls = 0
|
|
|
|
async def fetch_author_search_html(self, query: str, *, start: int = 0) -> FetchResult:
|
|
assert start == 0
|
|
self.calls += 1
|
|
if not self._fetch_results:
|
|
raise RuntimeError("No stub fetch results configured.")
|
|
index = min(self.calls - 1, len(self._fetch_results) - 1)
|
|
return self._fetch_results[index]
|
|
|
|
async def fetch_profile_html(self, scholar_id: str) -> FetchResult:
|
|
raise NotImplementedError
|
|
|
|
async def fetch_profile_page_html(self, scholar_id: str, *, cstart: int, pagesize: int) -> FetchResult:
|
|
raise NotImplementedError
|
|
|
|
|
|
def _ok_author_search_fetch() -> FetchResult:
|
|
body = (
|
|
"<html><body>"
|
|
'<div class="gsc_1usr">'
|
|
'<img src="/citations/images/avatar_scholar_256.png" />'
|
|
'<a class="gs_ai_name" href="/citations?hl=en&user=abcDEF123456">Ada Lovelace</a>'
|
|
'<div class="gs_ai_aff">Analytical Engine</div>'
|
|
'<div class="gs_ai_eml">Verified email at computing.example</div>'
|
|
'<div class="gs_ai_cby">Cited by 42</div>'
|
|
'<a class="gs_ai_one_int">Mathematics</a>'
|
|
"</div>"
|
|
"</body></html>"
|
|
)
|
|
return FetchResult(
|
|
requested_url="https://scholar.google.com/citations?hl=en&view_op=search_authors&mauthors=ada",
|
|
status_code=200,
|
|
final_url="https://scholar.google.com/citations?hl=en&view_op=search_authors&mauthors=ada",
|
|
body=body,
|
|
error=None,
|
|
)
|
|
|
|
|
|
def _blocked_author_search_fetch() -> FetchResult:
|
|
return FetchResult(
|
|
requested_url="https://scholar.google.com/citations?hl=en&view_op=search_authors&mauthors=ada",
|
|
status_code=200,
|
|
final_url=(
|
|
"https://accounts.google.com/v3/signin/identifier?continue=https%3A%2F%2Fscholar.google.com%2Fcitations"
|
|
),
|
|
body="<html><body>Sign in</body></html>",
|
|
error=None,
|
|
)
|
|
|
|
|
|
def _network_timeout_fetch() -> FetchResult:
|
|
return FetchResult(
|
|
requested_url="https://scholar.google.com/citations?hl=en&view_op=search_authors&mauthors=ada",
|
|
status_code=None,
|
|
final_url=None,
|
|
body="",
|
|
error="timed out",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_author_candidates_serves_cached_response_for_same_query(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
source = StubScholarSource([_ok_author_search_fetch()])
|
|
|
|
first = await scholar_service.search_author_candidates(
|
|
source=source,
|
|
db_session=db_session,
|
|
query="Ada Lovelace",
|
|
limit=10,
|
|
network_error_retries=0,
|
|
retry_backoff_seconds=0.0,
|
|
search_enabled=True,
|
|
cache_ttl_seconds=600,
|
|
blocked_cache_ttl_seconds=60,
|
|
cache_max_entries=64,
|
|
min_interval_seconds=0.0,
|
|
interval_jitter_seconds=0.0,
|
|
cooldown_block_threshold=3,
|
|
cooldown_seconds=300,
|
|
)
|
|
second = await scholar_service.search_author_candidates(
|
|
source=source,
|
|
db_session=db_session,
|
|
query="Ada Lovelace",
|
|
limit=10,
|
|
network_error_retries=0,
|
|
retry_backoff_seconds=0.0,
|
|
search_enabled=True,
|
|
cache_ttl_seconds=600,
|
|
blocked_cache_ttl_seconds=60,
|
|
cache_max_entries=64,
|
|
min_interval_seconds=0.0,
|
|
interval_jitter_seconds=0.0,
|
|
cooldown_block_threshold=3,
|
|
cooldown_seconds=300,
|
|
)
|
|
|
|
assert first.state == ParseState.OK
|
|
assert second.state == ParseState.OK
|
|
assert source.calls == 1
|
|
assert len(second.candidates) == 1
|
|
assert second.candidates[0].scholar_id == "abcDEF123456"
|
|
assert "author_search_served_from_cache" in second.warnings
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_author_candidates_trips_cooldown_after_blocked_responses(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
source = StubScholarSource([_blocked_author_search_fetch()])
|
|
|
|
first = await scholar_service.search_author_candidates(
|
|
source=source,
|
|
db_session=db_session,
|
|
query="Blocked Query",
|
|
limit=10,
|
|
network_error_retries=0,
|
|
retry_backoff_seconds=0.0,
|
|
search_enabled=True,
|
|
cache_ttl_seconds=0,
|
|
blocked_cache_ttl_seconds=0,
|
|
cache_max_entries=64,
|
|
min_interval_seconds=0.0,
|
|
interval_jitter_seconds=0.0,
|
|
cooldown_block_threshold=1,
|
|
cooldown_seconds=300,
|
|
)
|
|
second = await scholar_service.search_author_candidates(
|
|
source=source,
|
|
db_session=db_session,
|
|
query="Blocked Query",
|
|
limit=10,
|
|
network_error_retries=0,
|
|
retry_backoff_seconds=0.0,
|
|
search_enabled=True,
|
|
cache_ttl_seconds=0,
|
|
blocked_cache_ttl_seconds=0,
|
|
cache_max_entries=64,
|
|
min_interval_seconds=0.0,
|
|
interval_jitter_seconds=0.0,
|
|
cooldown_block_threshold=1,
|
|
cooldown_seconds=300,
|
|
)
|
|
|
|
assert first.state == ParseState.BLOCKED_OR_CAPTCHA
|
|
assert "author_search_circuit_breaker_armed" in first.warnings
|
|
assert source.calls == 1
|
|
|
|
assert second.state == ParseState.BLOCKED_OR_CAPTCHA
|
|
assert second.state_reason == scholar_service.SEARCH_COOLDOWN_REASON
|
|
assert "author_search_cooldown_active" in second.warnings
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_author_candidates_emits_cooldown_alert_warning_after_threshold(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
source = StubScholarSource([_blocked_author_search_fetch()])
|
|
|
|
_ = await scholar_service.search_author_candidates(
|
|
source=source,
|
|
db_session=db_session,
|
|
query="Blocked Query",
|
|
limit=10,
|
|
network_error_retries=0,
|
|
retry_backoff_seconds=0.0,
|
|
search_enabled=True,
|
|
cache_ttl_seconds=0,
|
|
blocked_cache_ttl_seconds=0,
|
|
cache_max_entries=64,
|
|
min_interval_seconds=0.0,
|
|
interval_jitter_seconds=0.0,
|
|
cooldown_block_threshold=1,
|
|
cooldown_seconds=300,
|
|
cooldown_rejection_alert_threshold=1,
|
|
)
|
|
second = await scholar_service.search_author_candidates(
|
|
source=source,
|
|
db_session=db_session,
|
|
query="Blocked Query",
|
|
limit=10,
|
|
network_error_retries=0,
|
|
retry_backoff_seconds=0.0,
|
|
search_enabled=True,
|
|
cache_ttl_seconds=0,
|
|
blocked_cache_ttl_seconds=0,
|
|
cache_max_entries=64,
|
|
min_interval_seconds=0.0,
|
|
interval_jitter_seconds=0.0,
|
|
cooldown_block_threshold=1,
|
|
cooldown_seconds=300,
|
|
cooldown_rejection_alert_threshold=1,
|
|
)
|
|
|
|
assert second.state == ParseState.BLOCKED_OR_CAPTCHA
|
|
assert "author_search_cooldown_alert_threshold_exceeded" in second.warnings
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_author_candidates_adds_retry_threshold_warning(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
source = StubScholarSource(
|
|
[
|
|
_network_timeout_fetch(),
|
|
_network_timeout_fetch(),
|
|
_ok_author_search_fetch(),
|
|
]
|
|
)
|
|
|
|
result = await scholar_service.search_author_candidates(
|
|
source=source,
|
|
db_session=db_session,
|
|
query="Ada Lovelace",
|
|
limit=10,
|
|
network_error_retries=2,
|
|
retry_backoff_seconds=0.0,
|
|
search_enabled=True,
|
|
cache_ttl_seconds=0,
|
|
blocked_cache_ttl_seconds=0,
|
|
cache_max_entries=64,
|
|
min_interval_seconds=0.0,
|
|
interval_jitter_seconds=0.0,
|
|
cooldown_block_threshold=3,
|
|
cooldown_seconds=300,
|
|
retry_alert_threshold=2,
|
|
)
|
|
|
|
assert result.state == ParseState.OK
|
|
assert "author_search_retry_threshold_exceeded_2" in result.warnings
|
|
assert source.calls == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_author_candidates_can_be_disabled_by_configuration(
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
source = StubScholarSource([_ok_author_search_fetch()])
|
|
|
|
parsed = await scholar_service.search_author_candidates(
|
|
source=source,
|
|
db_session=db_session,
|
|
query="Ada Lovelace",
|
|
limit=5,
|
|
network_error_retries=0,
|
|
retry_backoff_seconds=0.0,
|
|
search_enabled=False,
|
|
cache_ttl_seconds=600,
|
|
blocked_cache_ttl_seconds=60,
|
|
cache_max_entries=64,
|
|
min_interval_seconds=0.0,
|
|
interval_jitter_seconds=0.0,
|
|
cooldown_block_threshold=1,
|
|
cooldown_seconds=300,
|
|
)
|
|
|
|
assert parsed.state == ParseState.BLOCKED_OR_CAPTCHA
|
|
assert parsed.state_reason == scholar_service.SEARCH_DISABLED_REASON
|
|
assert source.calls == 0
|
|
assert "author_search_disabled_by_configuration" in parsed.warnings
|