7af97c2c59
Stage 3 of AvitoScraper_v2. - _get_or_fetch_imv_cached(db, **params) — TTL 24h по cache_key (sha256 of address+params) * Cache HIT: SELECT avito_imv_evaluations WHERE cache_key AND fetched_at > NOW() - 24h * Cache MISS: evaluate_via_imv + save_imv_evaluation (2 HTTP requests to Avito) * Graceful: на любой error → return None, estimator продолжает без IMV - В estimate_quality после aggregation: * Map house_type/repair_state → Avito IMV формат (_IMV_HOUSE_TYPE_MAP, _IMV_REPAIR_MAP) * Skip IMV если payload без required fields (rooms/area/floor/total_floors/house_type/repair) * Append 'avito_imv' в sources_used (saved + returned) - После INSERT estimate: link estimate_id в avito_imv_evaluations row (для analytics joining) - Logging: imv=<price> в final estimate log line - pytest offline (4 tests): house_type_map / repair_map / cache_miss_fetch / fetch_failure_graceful * anyio.run() вместо pytest-asyncio (не установлен в tradein-mvp venv) * DATABASE_URL dummy через os.environ.setdefault (Settings fail-fast без него) IMV — ON-DEMAND only (per estimator call), НЕ в cron-scrape. Refs: AvitoScraper_v2_Implementation_Plan Stage 3.
108 lines
3.3 KiB
Python
108 lines
3.3 KiB
Python
"""Offline smoke for estimator IMV integration. Mocks IMV fetch + DB."""
|
|
|
|
import os
|
|
|
|
# Settings требует DATABASE_URL при инициализации (fail-fast, C-3).
|
|
# Для offline unit-тестов задаём dummy DSN до любого импорта из app.
|
|
os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import anyio
|
|
|
|
from app.services.estimator import (
|
|
_IMV_HOUSE_TYPE_MAP,
|
|
_IMV_REPAIR_MAP,
|
|
_get_or_fetch_imv_cached,
|
|
)
|
|
|
|
|
|
def test_imv_house_type_map() -> None:
|
|
assert _IMV_HOUSE_TYPE_MAP["panel"] == "panel"
|
|
assert _IMV_HOUSE_TYPE_MAP["monolithic"] == "monolith"
|
|
assert _IMV_HOUSE_TYPE_MAP[None] is None
|
|
assert _IMV_HOUSE_TYPE_MAP.get("unknown") is None
|
|
|
|
|
|
def test_imv_repair_map() -> None:
|
|
assert _IMV_REPAIR_MAP["needs_repair"] == "required"
|
|
assert _IMV_REPAIR_MAP["excellent"] == "designer"
|
|
assert _IMV_REPAIR_MAP[None] is None
|
|
|
|
|
|
def test_imv_cache_miss_calls_evaluate() -> None:
|
|
"""Cache miss → calls evaluate_via_imv + save_imv_evaluation."""
|
|
from app.services.scrapers.avito_imv import IMVEvaluation, IMVGeo
|
|
|
|
mock_db = MagicMock()
|
|
mock_db.execute.return_value.mappings.return_value.first.return_value = None # no cache hit
|
|
|
|
fake_result = IMVEvaluation(
|
|
cache_key="x" * 64,
|
|
address="ЕКБ test",
|
|
rooms=2,
|
|
area_m2=42.0,
|
|
floor=4,
|
|
floor_at_home=5,
|
|
house_type="panel",
|
|
renovation_type="cosmetic",
|
|
has_balcony=True,
|
|
has_loggia=False,
|
|
geo=IMVGeo(geo_hash="JWT"),
|
|
recommended_price=6_290_000,
|
|
lower_price=6_100_000,
|
|
higher_price=6_600_000,
|
|
market_count=800,
|
|
)
|
|
|
|
async def _run() -> None:
|
|
with (
|
|
patch(
|
|
"app.services.estimator.evaluate_via_imv",
|
|
new=AsyncMock(return_value=fake_result),
|
|
),
|
|
patch("app.services.estimator.save_imv_evaluation", return_value=1),
|
|
):
|
|
result = await _get_or_fetch_imv_cached(
|
|
mock_db,
|
|
address="ЕКБ test",
|
|
rooms=2,
|
|
area_m2=42.0,
|
|
floor=4,
|
|
floor_at_home=5,
|
|
house_type="panel",
|
|
renovation_type="cosmetic",
|
|
has_balcony=True,
|
|
has_loggia=False,
|
|
)
|
|
assert result is not None
|
|
assert result.recommended_price == 6_290_000
|
|
|
|
anyio.run(_run)
|
|
|
|
|
|
def test_imv_fetch_failure_returns_none_gracefully() -> None:
|
|
"""Network failure → returns None, не raises."""
|
|
mock_db = MagicMock()
|
|
mock_db.execute.return_value.mappings.return_value.first.return_value = None
|
|
|
|
async def _run() -> None:
|
|
with patch(
|
|
"app.services.estimator.evaluate_via_imv",
|
|
new=AsyncMock(side_effect=RuntimeError("test net error")),
|
|
):
|
|
result = await _get_or_fetch_imv_cached(
|
|
mock_db,
|
|
address="X",
|
|
rooms=2,
|
|
area_m2=42.0,
|
|
floor=4,
|
|
floor_at_home=5,
|
|
house_type="panel",
|
|
renovation_type="cosmetic",
|
|
has_balcony=True,
|
|
has_loggia=False,
|
|
)
|
|
assert result is None
|
|
|
|
anyio.run(_run)
|