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
|