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)
|