e5fb209ced
Backend foundation for the Yandex admin scraper page (UI follows in next PR).
scraper_settings (mig 053):
- New table source PK + request_delay_sec numeric(5,2) + updated_at + description
- Seeded with row 'yandex' = 5.0s (umbrella key for all 4 Yandex scrapers)
scraper_settings.py (new):
- get_scraper_delay(source) -> float, with 60s in-process cache + DB fallback
- Yandex sub-scrapers (yandex_detail / yandex_newbuilding / yandex_valuation)
resolve to umbrella 'yandex' key via _KEY_ALIASES
- invalidate_cache(source | None) for admin PUT side-effect
- Deferred SessionLocal import (_open_session helper) for unit-test compat
4 Yandex scrapers wired to load delay at __init__:
- yandex_realty.py / yandex_detail.py / yandex_newbuilding.py / yandex_valuation.py
- Class default bumped to 5.0; instance value overridden via get_scraper_delay(self.name)
admin.py -- 5 new endpoints:
- GET /scraper-settings list[ScraperSetting]
- PUT /scraper-settings/{source} upsert + invalidate_cache
- POST /scrape/yandex-detail offer_url= -> detail snapshot
- POST /scrape/yandex-newbuilding slug=, id=, city=ekaterinburg -> JK snapshot
- POST /scrape/yandex-valuation address=, offer_category=, offer_type=, page= -> house meta
Tests: 10 + 4 = 14 new unit tests (loader cache/alias/error paths + scraper wiring).
Ruff clean.
118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
"""Tests for scraper_settings loader (cache + DB fallback + alias mapping)."""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.services import scraper_settings as ss
|
|
|
|
# Patch target: _open_session is a module-level wrapper so tests can mock DB
|
|
_PATCH_TARGET = "app.services.scraper_settings._open_session"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_cache():
|
|
ss._CACHE.clear()
|
|
yield
|
|
ss._CACHE.clear()
|
|
|
|
|
|
def _mock_session(row_value: float | None) -> MagicMock:
|
|
"""Return a mock that behaves like SessionLocal() context-manager."""
|
|
session = MagicMock()
|
|
session.__enter__ = MagicMock(return_value=session)
|
|
session.__exit__ = MagicMock(return_value=False)
|
|
if row_value is None:
|
|
session.execute.return_value.first.return_value = None
|
|
else:
|
|
session.execute.return_value.first.return_value = (row_value,)
|
|
# _open_session() returns the session directly (no extra call)
|
|
mock_open = MagicMock(return_value=session)
|
|
return mock_open, session
|
|
|
|
|
|
def test_returns_db_value_for_yandex():
|
|
mock_open, _ = _mock_session(7.5)
|
|
with patch(_PATCH_TARGET, mock_open):
|
|
assert ss.get_scraper_delay("yandex") == 7.5
|
|
|
|
|
|
def test_yandex_subkeys_map_to_yandex_umbrella():
|
|
"""yandex_detail / yandex_newbuilding / yandex_valuation all read 'yandex' row."""
|
|
mock_open, _session = _mock_session(8.0)
|
|
with patch(_PATCH_TARGET, mock_open):
|
|
assert ss.get_scraper_delay("yandex_detail") == 8.0
|
|
assert ss.get_scraper_delay("yandex_newbuilding") == 8.0
|
|
assert ss.get_scraper_delay("yandex_valuation") == 8.0
|
|
# All 3 lookups hit the cache after the first; _open_session called once
|
|
assert mock_open.call_count == 1
|
|
|
|
|
|
def test_cache_hit_within_ttl():
|
|
mock_open, _session = _mock_session(6.0)
|
|
with patch(_PATCH_TARGET, mock_open):
|
|
ss.get_scraper_delay("yandex")
|
|
ss.get_scraper_delay("yandex")
|
|
ss.get_scraper_delay("yandex")
|
|
assert mock_open.call_count == 1 # cached after first
|
|
|
|
|
|
def test_cache_expires_after_ttl():
|
|
mock_open, _session = _mock_session(6.0)
|
|
with patch(_PATCH_TARGET, mock_open):
|
|
ss.get_scraper_delay("yandex")
|
|
# Force-age the cache
|
|
with ss._CACHE_LOCK:
|
|
v, t = ss._CACHE["yandex"]
|
|
ss._CACHE["yandex"] = (v, t - 999)
|
|
ss.get_scraper_delay("yandex")
|
|
assert mock_open.call_count == 2
|
|
|
|
|
|
def test_db_missing_row_returns_default():
|
|
mock_open, _ = _mock_session(None)
|
|
with patch(_PATCH_TARGET, mock_open):
|
|
assert ss.get_scraper_delay("yandex") == ss._DEFAULT_DELAY_BY_SOURCE["yandex"]
|
|
|
|
|
|
def test_db_error_returns_default():
|
|
bad_session = MagicMock()
|
|
bad_session.__enter__ = MagicMock(return_value=bad_session)
|
|
bad_session.__exit__ = MagicMock(return_value=False)
|
|
bad_session.execute.side_effect = RuntimeError("db down")
|
|
bad_open = MagicMock(return_value=bad_session)
|
|
with patch(_PATCH_TARGET, bad_open):
|
|
assert ss.get_scraper_delay("yandex") == ss._DEFAULT_DELAY_BY_SOURCE["yandex"]
|
|
|
|
|
|
def test_unknown_source_uses_global_default():
|
|
mock_open, _ = _mock_session(None)
|
|
with patch(_PATCH_TARGET, mock_open):
|
|
assert ss.get_scraper_delay("some_new_source") == ss._GLOBAL_DEFAULT_DELAY
|
|
|
|
|
|
def test_invalidate_cache_specific_source():
|
|
mock_open, _ = _mock_session(6.0)
|
|
with patch(_PATCH_TARGET, mock_open):
|
|
ss.get_scraper_delay("yandex")
|
|
ss.invalidate_cache("yandex")
|
|
ss.get_scraper_delay("yandex")
|
|
assert mock_open.call_count == 2
|
|
|
|
|
|
def test_invalidate_cache_all():
|
|
mock_open, _ = _mock_session(6.0)
|
|
with patch(_PATCH_TARGET, mock_open):
|
|
ss.get_scraper_delay("yandex")
|
|
ss.invalidate_cache()
|
|
ss.get_scraper_delay("yandex")
|
|
assert mock_open.call_count == 2
|
|
|
|
|
|
def test_invalidate_alias_invalidates_umbrella():
|
|
"""invalidate_cache('yandex_detail') should clear the 'yandex' umbrella key."""
|
|
mock_open, _ = _mock_session(6.0)
|
|
with patch(_PATCH_TARGET, mock_open):
|
|
ss.get_scraper_delay("yandex_detail")
|
|
ss.invalidate_cache("yandex_detail")
|
|
ss.get_scraper_delay("yandex_detail")
|
|
assert mock_open.call_count == 2
|