"""Tests for velocity-score service (#34 D2). Mock-based — не требуют живой БД. Источник данных — objective_corpus_room_month (мигрировано с domrf_kn_sale_graph). Mock shape совместим: sales query возвращает те же aliases (obj_id, total_sqm, months_with_data, period_start, period_end) через GROUP BY domrf_obj_id. """ from __future__ import annotations import datetime from unittest.mock import MagicMock, patch import pytest from app.services.site_finder.velocity import ( _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH, VelocityResult, compute_velocity, ) # Тестовый WKT — небольшой квадрат в центре ЕКБ. _PARCEL_WKT = "POINT(60.605 56.838)" # ── Вспомогательные фабрики mock-строк ──────────────────────────────────────── def _comp_row(obj_id: int, distance_m: float = 500.0) -> MagicMock: r = MagicMock() r.__getitem__ = lambda self, k: { "obj_id": obj_id, "comm_name": f"ЖК-{obj_id}", "dev_name": "TestDev", "obj_class": "комфорт", "district_name": "Ленинский", "distance_m": distance_m, }[k] return r def _sales_row( obj_id: int, total_sqm: float, months: int, start: str = "2024-11-01", end: str = "2025-04-01", ) -> MagicMock: r = MagicMock() start_d = datetime.date.fromisoformat(start) end_d = datetime.date.fromisoformat(end) r.__getitem__ = lambda self, k: { "obj_id": obj_id, "total_sqm": total_sqm, "months_with_data": months, "period_start": start_d, "period_end": end_d, }[k] return r def _make_db(comp_rows: list, sales_rows: list) -> MagicMock: """Сконструировать mock Session с двумя последовательными вызовами execute.""" db = MagicMock() execute_results = [] for rows in [comp_rows, sales_rows]: result = MagicMock() result.mappings.return_value.all.return_value = rows execute_results.append(result) db.execute.side_effect = execute_results return db # ── Тесты ───────────────────────────────────────────────────────────────────── def test_no_competitors_returns_none(): """Нет ЖК в радиусе → None.""" db = _make_db(comp_rows=[], sales_rows=[]) result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT) assert result is None def test_no_sales_data_returns_none(): """ЖК есть, но нет данных objective_corpus_room_month → None.""" comp_rows = [_comp_row(1), _comp_row(2)] db = _make_db(comp_rows=comp_rows, sales_rows=[]) result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT) assert result is None def test_healthy_sales_returns_result(): """12 конкурентов с нормальными продажами → score в (0,1), confidence='high'.""" n = 12 comp_rows = [_comp_row(i, distance_m=300.0 + i * 100) for i in range(1, n + 1)] # Каждый ЖК продаёт 4500 м² за 6 мес → 750 м²/мес. Суммарно: 4500*12 = 54000 за 6 мес. sales_rows = [_sales_row(i, total_sqm=4500.0, months=6) for i in range(1, n + 1)] db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows) with patch( "app.services.site_finder.velocity._get_ekb_median", return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH, ): result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT) assert result is not None assert result.competitors_count == n assert 0.0 < result.velocity_score <= 1.0 assert result.confidence == "high" assert result.months_observed == 6 def test_few_competitors_low_confidence(): """2 конкурента → confidence='low'.""" comp_rows = [_comp_row(1), _comp_row(2)] sales_rows = [_sales_row(1, total_sqm=3000.0, months=2)] db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows) with patch( "app.services.site_finder.velocity._get_ekb_median", return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH, ): result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT) assert result is not None assert result.confidence == "low" def test_medium_confidence(): """7 конкурентов, 4 месяца → confidence='medium'.""" n = 7 comp_rows = [_comp_row(i) for i in range(1, n + 1)] sales_rows = [_sales_row(i, total_sqm=4000.0, months=4) for i in range(1, n + 1)] db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows) with patch( "app.services.site_finder.velocity._get_ekb_median", return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH, ): result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT) assert result is not None assert result.confidence == "medium" def test_ekb_median_fallback_used_when_none(): """Если _get_ekb_median вернул None — используется fallback-константа.""" comp_rows = [_comp_row(1)] sales_rows = [_sales_row(1, total_sqm=9000.0, months=6)] db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows) with patch("app.services.site_finder.velocity._get_ekb_median", return_value=None): result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT) assert result is not None assert result.ekb_median_sqm == _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH def test_score_capped_at_1(): """Огромный объём → score не превышает 1.0.""" comp_rows = [_comp_row(1)] # 1 000 000 м² за месяц — абсурдно много sales_rows = [_sales_row(1, total_sqm=6_000_000.0, months=6)] db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows) with patch( "app.services.site_finder.velocity._get_ekb_median", return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH, ): result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT) assert result is not None assert result.velocity_score == pytest.approx(1.0) def test_score_zero_when_no_sales_sqm(): """total_sqm=0 → None (нет данных, не score=0).""" comp_rows = [_comp_row(1)] # total_sqm=0 — нет продаж → должен вернуть None sales_rows = [_sales_row(1, total_sqm=0.0, months=5)] db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows) with patch( "app.services.site_finder.velocity._get_ekb_median", return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH, ): result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT) assert result is None def test_as_dict_structure(): """as_dict() содержит все ожидаемые ключи.""" vr = VelocityResult( competitors_count=5, monthly_velocity_sqm=3000.0, ekb_median_sqm=4500.0, velocity_score=0.333, confidence="medium", months_observed=4, period_start="2024-11", period_end="2025-02", sample_competitors=[], ) d = vr.as_dict() assert "competitors_count" in d assert "velocity_score" in d assert "confidence" in d assert "period" in d assert d["period"]["start"] == "2024-11" assert d["period"]["end"] == "2025-02" assert d["velocity_score"] == pytest.approx(0.333, abs=1e-3) def test_sample_competitors_top5(): """sample_competitors содержит не более 5 элементов, отсортированных по убыванию.""" n = 8 comp_rows = [_comp_row(i) for i in range(1, n + 1)] sales_rows = [_sales_row(i, total_sqm=float(i * 1000), months=5) for i in range(1, n + 1)] db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows) with patch( "app.services.site_finder.velocity._get_ekb_median", return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH, ): result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT) assert result is not None assert len(result.sample_competitors) <= 5 sqms = [c["total_sqm_period"] for c in result.sample_competitors] assert sqms == sorted(sqms, reverse=True)