"""Unit-тесты для get_best_layouts (Fix SF-01: honest time_window velocity). Проверяет, что разные time_window → разные deals_window → разный velocity_per_month. Mock-стратегия: патчим db.execute с side_effect, повторяя порядок вызовов в get_best_layouts: 1. _PARCEL_CENTROID_SQL → .mappings().first() 2. _COMPETITORS_IN_RADIUS_SQL → .mappings().all() 3. _INLINE_VELOCITY_SQL → .mappings().all() 4. db.scalar() → MAX(snapshot_date) — через .return_value 5. _SUPPLY_BATCH_SQL → .mappings().all() Ключевые asserts: - last_month (1 мес) → velocity = deals_window / 1.0 - last_quarter (3 мес) → velocity = deals_window / 3.0 - last_year (12 мес) → velocity = deals_window / 12.0 - Разный deals_window при разных time_window → разный mix. """ from __future__ import annotations import datetime as dt from unittest.mock import MagicMock import pytest from app.schemas.parcel import BestLayoutsRequest from app.services.site_finder.best_layouts import _TIME_WINDOW_PARAMS, get_best_layouts _TODAY = dt.date.today() CAD_NUM = "66:41:0303161:123" # ── Фабрики mock-строк ──────────────────────────────────────────────────────── def _coord_row(lon: float = 60.6, lat: float = 56.85) -> MagicMock: r = MagicMock() r.__getitem__ = lambda self, k: {"center_lon": lon, "center_lat": lat}[k] return r def _obj_id_row(obj_id: int) -> MagicMock: r = MagicMock() r.__getitem__ = lambda self, k: {"obj_id": obj_id}[k] return r def _vel_row( room_bucket: str = "2", deals_window: float = 48.0, avg_area: float = 55.0, avg_price_rub: float | None = 120000.0, obj_ids: list[int] | None = None, window_start: dt.date | None = None, window_end: dt.date | None = None, ) -> MagicMock: """Строка из _INLINE_VELOCITY_SQL. deals_window — реальные сделки за честное окно (не 24 мес). """ oids = obj_ids if obj_ids is not None else [1] ws = window_start or _TODAY - dt.timedelta(days=90) we = window_end or _TODAY r = MagicMock() r.__getitem__ = lambda self, k: { "room_bucket": room_bucket, "deals_window": deals_window, "avg_area_m2": avg_area, "avg_price_per_m2_rub": avg_price_rub, "competitor_obj_ids": oids, "competitor_count": len(oids), "window_start": ws, "window_end": we, }[k] return r def _supply_row(rb: str, ab: str, units: int) -> MagicMock: r = MagicMock() r.__getitem__ = lambda self, k: {"rb": rb, "ab": ab, "units": units}[k] return r def _make_db( coord: MagicMock | None = None, id_rows: list[MagicMock] | None = None, vel_rows: list[MagicMock] | None = None, supply_rows: list[MagicMock] | None = None, latest_snap: dt.date | None = None, ) -> MagicMock: """Сконструировать mock Session. Порядок db.execute(): 1. centroid → .mappings().first() 2. competitors → .mappings().all() 3. velocity → .mappings().all() 4. supply → .mappings().all() (только если latest_snap is not None) db.scalar() → latest_snap (MAX snapshot_date). """ db = MagicMock() db.scalar.return_value = latest_snap if latest_snap is not None else _TODAY r0 = MagicMock() r0.mappings.return_value.first.return_value = coord r1 = MagicMock() r1.mappings.return_value.all.return_value = id_rows or [] r2 = MagicMock() r2.mappings.return_value.all.return_value = vel_rows or [] r3 = MagicMock() r3.mappings.return_value.all.return_value = supply_rows or [] db.execute.side_effect = [r0, r1, r2, r3] return db def _request(**kwargs) -> BestLayoutsRequest: defaults: dict = { "radius_km": 1.0, "time_window": "last_quarter", "min_velocity_per_month": 0.0, } defaults.update(kwargs) return BestLayoutsRequest(**defaults) # ── Тесты TIME_WINDOW_PARAMS ────────────────────────────────────────────────── def test_time_window_params_keys() -> None: """Все три time_window определены, months_in_window > 0.""" for key in ("last_month", "last_quarter", "last_year"): assert key in _TIME_WINDOW_PARAMS interval_str, months = _TIME_WINDOW_PARAMS[key] assert isinstance(interval_str, str) and len(interval_str) > 0 assert months > 0 # ── Тест SF-01: разный deals_window → разный velocity ──────────────────────── def test_last_month_velocity_divisor_1() -> None: """time_window=last_month: velocity = deals_window / 1.0.""" deals = 30.0 db = _make_db( coord=_coord_row(), id_rows=[_obj_id_row(1)], vel_rows=[_vel_row("1", deals_window=deals, obj_ids=[1])], ) req = _request(time_window="last_month") resp = get_best_layouts(db, CAD_NUM, req) assert len(resp.top_layouts) == 1 assert resp.top_layouts[0].velocity_per_month == pytest.approx(30.0, rel=1e-3) def test_last_quarter_velocity_divisor_3() -> None: """time_window=last_quarter: velocity = deals_window / 3.0.""" deals = 30.0 db = _make_db( coord=_coord_row(), id_rows=[_obj_id_row(1)], vel_rows=[_vel_row("1", deals_window=deals, obj_ids=[1])], ) req = _request(time_window="last_quarter") resp = get_best_layouts(db, CAD_NUM, req) assert len(resp.top_layouts) == 1 assert resp.top_layouts[0].velocity_per_month == pytest.approx(10.0, rel=1e-3) def test_last_year_velocity_divisor_12() -> None: """time_window=last_year: velocity = deals_window / 12.0.""" deals = 60.0 db = _make_db( coord=_coord_row(), id_rows=[_obj_id_row(1)], vel_rows=[_vel_row("1", deals_window=deals, obj_ids=[1])], ) req = _request(time_window="last_year") resp = get_best_layouts(db, CAD_NUM, req) assert len(resp.top_layouts) == 1 assert resp.top_layouts[0].velocity_per_month == pytest.approx(5.0, rel=1e-3) def test_different_time_windows_produce_different_velocity() -> None: """Одни и те же deals_window → разная velocity_per_month для разных time_window. Главный acceptance-тест SF-01: time_window влияет на velocity, не только на масштаб. При одном и том же deals_window=30: last_month → 30.0 last_quarter → 10.0 last_year → 2.5 """ deals = 30.0 velocities: dict[str, float] = {} for tw in ("last_month", "last_quarter", "last_year"): db = _make_db( coord=_coord_row(), id_rows=[_obj_id_row(1)], vel_rows=[_vel_row("2", deals_window=deals, obj_ids=[1])], ) req = _request(time_window=tw) resp = get_best_layouts(db, CAD_NUM, req) assert len(resp.top_layouts) == 1, f"No layouts for {tw}" velocities[tw] = resp.top_layouts[0].velocity_per_month # Все три значения различаются vals = list(velocities.values()) assert vals[0] != vals[1] != vals[2], f"Velocities must differ: {velocities}" # last_month > last_quarter > last_year (одинаковые deals, разный знаменатель) assert velocities["last_month"] > velocities["last_quarter"] > velocities["last_year"] # ── Тест: ranking по velocity и sum pct = 100 ──────────────────────────────── def test_ranking_and_pct_sum_100() -> None: """3 room_buckets → ranking по velocity, sum pct = 100.""" id_rows = [_obj_id_row(1), _obj_id_row(2), _obj_id_row(3)] vel_rows = [ _vel_row("studio", deals_window=9.0, avg_area=26.0, obj_ids=[1]), # 9/3=3.0 _vel_row("1", deals_window=24.0, avg_area=40.0, obj_ids=[2]), # 24/3=8.0 _vel_row("2", deals_window=48.0, avg_area=55.0, obj_ids=[3]), # 48/3=16.0 ] supply_rows = [ _supply_row("studio", "<25", 20), _supply_row("1", "40-60", 60), _supply_row("2", "40-60", 80), ] db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows, supply_rows=supply_rows) req = _request(time_window="last_quarter") resp = get_best_layouts(db, CAD_NUM, req) top = resp.top_layouts assert len(top) == 3 # rank 1 = "2" (наибольший velocity 16.0) assert top[0].room_bucket == "2" assert top[0].rank == 1 assert top[0].velocity_per_month == pytest.approx(16.0, rel=1e-3) # rank 2 = "1" (8.0) assert top[1].room_bucket == "1" assert top[1].velocity_per_month == pytest.approx(8.0, rel=1e-3) # ранги уникальны assert sorted(t.rank for t in top) == [1, 2, 3] # sum pct = 100 mix = resp.recommendation_for_tz.mix assert sum(m.pct for m in mix) == 100 # ── Тест: пустые конкуренты ─────────────────────────────────────────────────── def test_no_competitors_returns_empty_response() -> None: """Нет конкурентов в радиусе → пустые top_layouts + confidence=low.""" db = _make_db(coord=_coord_row(), id_rows=[], vel_rows=[]) req = _request() resp = get_best_layouts(db, CAD_NUM, req) assert resp.top_layouts == [] assert resp.data_quality.confidence == "low" assert resp.recommendation_for_tz.based_on_obj_count == 0 # ── Тест: centroid не найден ────────────────────────────────────────────────── def test_centroid_not_found_raises_value_error() -> None: """Геометрия участка не найдена → ValueError.""" db = _make_db(coord=None) req = _request() with pytest.raises(ValueError, match="не найдена"): get_best_layouts(db, "99:99:9999999:999", req) # ── Тест: min_velocity фильтрует строки ────────────────────────────────────── def test_min_velocity_filters_low_rows() -> None: """min_velocity_per_month=5 → строки с velocity<5 не попадают в top_layouts. last_quarter (3 мес): studio: 9 / 3 = 3.0 < 5.0 → отфильтрован 1: 24 / 3 = 8.0 > 5.0 → остаётся """ id_rows = [_obj_id_row(1), _obj_id_row(2)] vel_rows = [ _vel_row("studio", deals_window=9.0, obj_ids=[1]), _vel_row("1", deals_window=24.0, obj_ids=[2]), ] db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows) req = _request(time_window="last_quarter", min_velocity_per_month=5.0) resp = get_best_layouts(db, CAD_NUM, req) top = resp.top_layouts assert len(top) == 1 assert top[0].room_bucket == "1" assert top[0].velocity_per_month == pytest.approx(8.0, rel=1e-3) # ── Тест: exclude_competitor_obj_ids ───────────────────────────────────────── def test_exclude_competitor_obj_ids() -> None: """exclude_competitor_obj_ids=[20] при единственном конкуренте → пустой ответ.""" id_rows = [_obj_id_row(20)] db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=[]) req = _request(exclude_competitor_obj_ids=[20]) resp = get_best_layouts(db, CAD_NUM, req) assert resp.top_layouts == [] assert resp.data_quality.objects_total_in_radius == 1 # ── Тест: total_sold_in_window совпадает с deals_window ────────────────────── def test_total_sold_in_window_matches_deals_window() -> None: """total_sold_in_window в TopLayoutRow = deals_window (целое).""" deals = 37.0 db = _make_db( coord=_coord_row(), id_rows=[_obj_id_row(5)], vel_rows=[_vel_row("3", deals_window=deals, obj_ids=[5])], ) req = _request(time_window="last_quarter") resp = get_best_layouts(db, CAD_NUM, req) assert len(resp.top_layouts) == 1 assert resp.top_layouts[0].total_sold_in_window == int(deals)