Justin Visser
bf04c77aa9
Add ruff and mypy to dev dependencies with configuration in pyproject.toml. Add a lint CI job that runs ruff check, ruff format --check, and mypy. Auto-fix import sorting and formatting across the codebase. Exclude alembic/versions from linting (auto-generated migrations). Ignore B008 (FastAPI Depends pattern) and RUF001 (unicode in user-facing strings). 21 ruff lint errors and 50 mypy errors remain for manual review. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
272 lines
8.6 KiB
Python
272 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.services.domains.scholar.parser import ParseState
|
|
from app.services.domains.scholar.source import FetchResult
|
|
from app.services.domains.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]
|
|
|
|
|
|
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
|