Light1YT
a85e6d8e77
Estimator ADDITIVELY derives expected_sold from asking-median × per-rooms asking_to_sold_ratio (table from #648 S2 / migration 080). Headline asking median, ranges, confidence and actual_deals are UNCHANGED. Graceful: missing/ empty ratio table → expected_sold fields None, estimate still returns 200 (ratio lookup rolls back the shared session on error so the subsequent INSERT isn't poisoned — mirrors sibling cache helpers). - estimator: _get_asking_sold_ratio(db, rooms) cached lookup (per-rooms → global -1 fallback → None, try/except+rollback), applied after repair-coef. - schema: 6 optional expected_sold_* / ratio fields on AggregatedEstimate. - api: rehydrate the new columns in GET /estimate/{id} + /pdf reopen SELECTs and both constructors (the documented reopen-path bug). - migration 081: ADD COLUMN IF NOT EXISTS for the persisted subset. - tests: 10 new (lookup/apply/graceful) + fix pre-existing test_estimator_ source_quota 2→3 tuple unpack of _fetch_analogs (was RED on main).
257 lines
11 KiB
Python
257 lines
11 KiB
Python
"""Tests for Cian Valuation integration in estimator.py (Stage 9)."""
|
|
import os
|
|
|
|
# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
|
|
os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import anyio
|
|
|
|
from app.services.scrapers.cian_valuation import CianValuationResult
|
|
|
|
|
|
def _fake_cian_result(**kwargs) -> CianValuationResult:
|
|
result = CianValuationResult()
|
|
result.sale_price_rub = kwargs.get("sale_price_rub", 7_500_000.0)
|
|
result.sale_accuracy = kwargs.get("sale_accuracy", 85.0)
|
|
result.sale_price_from = kwargs.get("sale_price_from", 7_000_000.0)
|
|
result.sale_price_to = kwargs.get("sale_price_to", 8_000_000.0)
|
|
result.chart = kwargs.get("chart", [{"month_date": "2026-05-01", "price": 145_000}])
|
|
result.chart_change_pct = kwargs.get("chart_change_pct", 1.5)
|
|
result.chart_change_direction = kwargs.get("chart_change_direction", "increase")
|
|
result.external_house_id = kwargs.get("external_house_id", 12345)
|
|
result.is_authenticated = kwargs.get("is_authenticated", True)
|
|
return result
|
|
|
|
|
|
# ── Cache-layer unit tests ────────────────────────────────────────────────────
|
|
|
|
def test_cian_valuation_cache_key_deterministic() -> None:
|
|
"""compute_cache_key produces same hash for identical inputs."""
|
|
from app.services.scrapers.cian_valuation import compute_cache_key
|
|
k1 = compute_cache_key("ЕКБ, ул. Учителей, 18", 38.8, 1, 4, "cosmetic", "sale")
|
|
k2 = compute_cache_key("ЕКБ, ул. Учителей, 18", 38.8, 1, 4, "cosmetic", "sale")
|
|
assert k1 == k2
|
|
assert len(k1) == 64
|
|
|
|
|
|
def test_cian_valuation_cache_key_differs_for_different_inputs() -> None:
|
|
from app.services.scrapers.cian_valuation import compute_cache_key
|
|
a = compute_cache_key("addr A", 38.8, 1, 4, "cosmetic", "sale")
|
|
b = compute_cache_key("addr B", 38.8, 1, 4, "cosmetic", "sale")
|
|
c = compute_cache_key("addr A", 50.0, 1, 4, "cosmetic", "sale")
|
|
d = compute_cache_key("addr A", 38.8, 2, 4, "cosmetic", "sale")
|
|
assert len({a, b, c, d}) == 4
|
|
|
|
|
|
# ── estimate_via_cian_valuation graceful-degradation tests ───────────────────
|
|
|
|
def test_estimate_via_cian_returns_none_when_no_cookies() -> None:
|
|
"""estimate_via_cian_valuation returns None when no cookies in DB (graceful)."""
|
|
db = MagicMock()
|
|
# cache miss
|
|
db.execute.return_value.mappings.return_value.first.return_value = None
|
|
|
|
async def _run() -> None:
|
|
with patch(
|
|
"app.services.scrapers.cian_valuation.load_session",
|
|
return_value=None, # no cookies stored
|
|
):
|
|
from app.services.scrapers.cian_valuation import estimate_via_cian_valuation
|
|
result = await estimate_via_cian_valuation(
|
|
db,
|
|
address="ЕКБ, ул. Учителей, 18",
|
|
total_area=38.8,
|
|
rooms_count=1,
|
|
floor=4,
|
|
total_floors=16,
|
|
use_cache=True,
|
|
)
|
|
assert result is None
|
|
|
|
anyio.run(_run)
|
|
|
|
|
|
# ── Estimator integration tests ───────────────────────────────────────────────
|
|
|
|
def _make_fake_geo():
|
|
"""Return GeocodeResult with provider field (not source)."""
|
|
from app.services.geocoder import GeocodeResult
|
|
return GeocodeResult(
|
|
lat=56.838,
|
|
lon=60.595,
|
|
full_address="Свердловская обл., Екатеринбург, ул. Учителей, 18",
|
|
provider="nominatim",
|
|
)
|
|
|
|
|
|
def _make_payload():
|
|
from app.schemas.trade_in import TradeInEstimateInput
|
|
return TradeInEstimateInput(
|
|
address="ЕКБ, ул. Учителей, 18",
|
|
area_m2=38.8,
|
|
rooms=1,
|
|
floor=4,
|
|
total_floors=16,
|
|
)
|
|
|
|
|
|
def _common_patches(cian_mock):
|
|
"""Return list of context managers for all sources except cian_valuation."""
|
|
return [
|
|
patch("app.services.estimator.geocode", new=AsyncMock(return_value=_make_fake_geo())),
|
|
patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator._fetch_analogs", return_value=([], False, "W")),
|
|
patch("app.services.estimator._fetch_deals", return_value=[]),
|
|
patch(
|
|
"app.services.estimator._get_or_fetch_imv_cached",
|
|
new=AsyncMock(return_value=None),
|
|
),
|
|
patch(
|
|
"app.services.estimator._get_or_fetch_yandex_valuation_cached",
|
|
new=AsyncMock(return_value=None),
|
|
),
|
|
patch(
|
|
"app.services.estimator.estimate_via_cian_valuation",
|
|
new=cian_mock,
|
|
),
|
|
# #648 S3: stub asking→sold lookup off (isolates cian-source assertions).
|
|
patch(
|
|
"app.services.estimator._get_asking_sold_ratio",
|
|
return_value=(None, None),
|
|
),
|
|
]
|
|
|
|
|
|
def test_estimator_includes_cian_valuation_when_available() -> None:
|
|
"""When cian_valuation returns result with sale_price_rub, sources_used includes it."""
|
|
from app.services.estimator import estimate_quality
|
|
|
|
db = MagicMock()
|
|
payload = _make_payload()
|
|
fake_cian = _fake_cian_result()
|
|
cian_mock = AsyncMock(return_value=fake_cian)
|
|
|
|
async def _run() -> None:
|
|
with (
|
|
patch("app.services.estimator.geocode",
|
|
new=AsyncMock(return_value=_make_fake_geo())),
|
|
patch("app.services.estimator.get_house_metadata",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator._fetch_analogs", return_value=([], False, "W")),
|
|
patch("app.services.estimator._fetch_deals", return_value=[]),
|
|
patch("app.services.estimator._get_or_fetch_imv_cached",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator._get_or_fetch_yandex_valuation_cached",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator.estimate_via_cian_valuation",
|
|
new=cian_mock),
|
|
patch("app.services.estimator._get_asking_sold_ratio",
|
|
return_value=(None, None)),
|
|
):
|
|
result = await estimate_quality(payload, db)
|
|
|
|
assert "cian_valuation" in result.sources_used
|
|
|
|
anyio.run(_run)
|
|
|
|
|
|
def test_estimator_graceful_when_cian_returns_none() -> None:
|
|
"""When estimate_via_cian_valuation returns None (no cookies), estimator continues."""
|
|
from app.services.estimator import estimate_quality
|
|
|
|
db = MagicMock()
|
|
payload = _make_payload()
|
|
cian_mock = AsyncMock(return_value=None)
|
|
|
|
async def _run() -> None:
|
|
with (
|
|
patch("app.services.estimator.geocode",
|
|
new=AsyncMock(return_value=_make_fake_geo())),
|
|
patch("app.services.estimator.get_house_metadata",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator._fetch_analogs", return_value=([], False, "W")),
|
|
patch("app.services.estimator._fetch_deals", return_value=[]),
|
|
patch("app.services.estimator._get_or_fetch_imv_cached",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator._get_or_fetch_yandex_valuation_cached",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator.estimate_via_cian_valuation",
|
|
new=cian_mock),
|
|
patch("app.services.estimator._get_asking_sold_ratio",
|
|
return_value=(None, None)),
|
|
):
|
|
result = await estimate_quality(payload, db)
|
|
|
|
assert "cian_valuation" not in result.sources_used
|
|
assert result.estimate_id is not None # estimator still returned a result
|
|
|
|
anyio.run(_run)
|
|
|
|
|
|
def test_estimator_graceful_when_cian_raises() -> None:
|
|
"""When estimate_via_cian_valuation raises, estimator logs warning and continues."""
|
|
from app.services.estimator import estimate_quality
|
|
|
|
db = MagicMock()
|
|
payload = _make_payload()
|
|
cian_mock = AsyncMock(side_effect=RuntimeError("network timeout"))
|
|
|
|
async def _run() -> None:
|
|
with (
|
|
patch("app.services.estimator.geocode",
|
|
new=AsyncMock(return_value=_make_fake_geo())),
|
|
patch("app.services.estimator.get_house_metadata",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator._fetch_analogs", return_value=([], False, "W")),
|
|
patch("app.services.estimator._fetch_deals", return_value=[]),
|
|
patch("app.services.estimator._get_or_fetch_imv_cached",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator._get_or_fetch_yandex_valuation_cached",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator.estimate_via_cian_valuation",
|
|
new=cian_mock),
|
|
patch("app.services.estimator._get_asking_sold_ratio",
|
|
return_value=(None, None)),
|
|
):
|
|
result = await estimate_quality(payload, db)
|
|
|
|
assert "cian_valuation" not in result.sources_used
|
|
assert result.estimate_id is not None
|
|
|
|
anyio.run(_run)
|
|
|
|
|
|
def test_estimator_cian_result_no_sale_price_not_added() -> None:
|
|
"""When cian_valuation returns result with sale_price_rub=None, not added to sources."""
|
|
from app.services.estimator import estimate_quality
|
|
|
|
db = MagicMock()
|
|
payload = _make_payload()
|
|
# Result returned but no sale price (e.g. partial parse)
|
|
fake_cian_no_price = _fake_cian_result(sale_price_rub=None)
|
|
cian_mock = AsyncMock(return_value=fake_cian_no_price)
|
|
|
|
async def _run() -> None:
|
|
with (
|
|
patch("app.services.estimator.geocode",
|
|
new=AsyncMock(return_value=_make_fake_geo())),
|
|
patch("app.services.estimator.get_house_metadata",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator._fetch_analogs", return_value=([], False, "W")),
|
|
patch("app.services.estimator._fetch_deals", return_value=[]),
|
|
patch("app.services.estimator._get_or_fetch_imv_cached",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator._get_or_fetch_yandex_valuation_cached",
|
|
new=AsyncMock(return_value=None)),
|
|
patch("app.services.estimator.estimate_via_cian_valuation",
|
|
new=cian_mock),
|
|
patch("app.services.estimator._get_asking_sold_ratio",
|
|
return_value=(None, None)),
|
|
):
|
|
result = await estimate_quality(payload, db)
|
|
|
|
assert "cian_valuation" not in result.sources_used
|
|
|
|
anyio.run(_run)
|