388 lines
15 KiB
Python
388 lines
15 KiB
Python
"""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, has_mapping) через LEFT JOIN
|
||
all_competitors + mapped (OBJ-2: ALL competitors included, unmapped has_mapping=False).
|
||
Третий вызов execute — bucket_rows (obj_id, room_bucket, units_sold, sqm_sold).
|
||
"""
|
||
|
||
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",
|
||
has_mapping: bool = True,
|
||
) -> 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,
|
||
# OBJ-2: LEFT JOIN all_competitors — все конкуренты включены,
|
||
# has_mapping=True если есть маппинг в objective_complex_mapping.
|
||
"has_mapping": has_mapping,
|
||
}[k]
|
||
return r
|
||
|
||
|
||
def _bucket_row(obj_id: int, room_bucket: str, units_sold: int, sqm_sold: float) -> MagicMock:
|
||
r = MagicMock()
|
||
r.__getitem__ = lambda self, k: {
|
||
"obj_id": obj_id,
|
||
"room_bucket": room_bucket,
|
||
"units_sold": units_sold,
|
||
"sqm_sold": sqm_sold,
|
||
}[k]
|
||
return r
|
||
|
||
|
||
def _make_db(
|
||
comp_rows: list,
|
||
sales_rows: list,
|
||
bucket_rows: list | None = None,
|
||
) -> MagicMock:
|
||
"""Сконструировать mock Session с тремя последовательными вызовами execute.
|
||
|
||
Порядок: comp_rows → sales_rows → bucket_rows.
|
||
bucket_rows=None → пустой список (bucket query gracefully degraded).
|
||
"""
|
||
db = MagicMock()
|
||
execute_results = []
|
||
for rows in [comp_rows, sales_rows, bucket_rows if bucket_rows is not None else []]:
|
||
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 → VelocityResult с velocity_data_available=False, score=0.
|
||
|
||
OBJ-2: функция больше не возвращает None при нулевых продажах —
|
||
возвращает «пустое» состояние (velocity_data_available=False, source='none').
|
||
Rosreestr-fallback пропускается т.к. cad_quarter не передан.
|
||
"""
|
||
comp_rows = [_comp_row(1)]
|
||
# total_sqm=0 — нет продаж → velocity_data_available=False
|
||
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 not None
|
||
assert result.velocity_data_available is False
|
||
assert result.velocity_score == pytest.approx(0.0)
|
||
assert result.velocity_source == "none"
|
||
|
||
|
||
def test_as_dict_structure():
|
||
"""as_dict() содержит все ожидаемые ключи, включая by_room_bucket."""
|
||
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=[],
|
||
by_room_bucket={"1": {"units": 10, "sqm": 450.0, "complexes_count": 2}},
|
||
velocity_data_available=True,
|
||
velocity_source="objective",
|
||
)
|
||
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)
|
||
assert "by_room_bucket" in d
|
||
assert d["by_room_bucket"]["1"]["units"] == 10
|
||
# OBJ-2: новые поля velocity_data_available и velocity_source
|
||
assert d["velocity_data_available"] is True
|
||
assert d["velocity_source"] == "objective"
|
||
|
||
|
||
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)
|
||
|
||
|
||
def test_by_room_bucket_aggregation():
|
||
"""by_room_bucket агрегирует units/sqm поверх всех конкурентов корректно."""
|
||
comp_rows = [_comp_row(1), _comp_row(2)]
|
||
sales_rows = [
|
||
_sales_row(1, total_sqm=3000.0, months=3),
|
||
_sales_row(2, total_sqm=2000.0, months=3),
|
||
]
|
||
bucket_rows = [
|
||
_bucket_row(1, "студия", units_sold=38, sqm_sold=1520.0),
|
||
_bucket_row(1, "1", units_sold=22, sqm_sold=990.0),
|
||
_bucket_row(2, "студия", units_sold=18, sqm_sold=720.0),
|
||
_bucket_row(2, "1", units_sold=13, sqm_sold=585.0),
|
||
]
|
||
db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows, bucket_rows=bucket_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 "студия" in result.by_room_bucket
|
||
assert "1" in result.by_room_bucket
|
||
# студия: 38+18=56 units, complexes from obj 1 and 2
|
||
assert result.by_room_bucket["студия"]["units"] == 56
|
||
assert result.by_room_bucket["студия"]["complexes_count"] == 2
|
||
# 1-к: 22+13=35 units
|
||
assert result.by_room_bucket["1"]["units"] == 35
|
||
# sqm rounded
|
||
assert result.by_room_bucket["студия"]["sqm"] == pytest.approx(2240.0)
|
||
|
||
|
||
def test_by_room_bucket_empty_when_no_bucket_data():
|
||
"""Если bucket query вернул пустой список — by_room_bucket пустой dict."""
|
||
comp_rows = [_comp_row(1)]
|
||
sales_rows = [_sales_row(1, total_sqm=5000.0, months=5)]
|
||
db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows, bucket_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.by_room_bucket == {}
|
||
|
||
|
||
def test_mapping_confidence_gate_in_sales_query():
|
||
"""OBJ-2 (#307): sales/bucket queries фильтруют objective_complex_mapping
|
||
по confidence — unreviewed low-score auto-matches исключены.
|
||
|
||
Проверяем, что SQL, переданный в db.execute для маппинг-CTE, содержит
|
||
предикат gate (is_reviewed / manual / match_score >= 0.85). Это гарантирует,
|
||
что ~115 fuzzy-trgm строк с is_reviewed=false и score<0.85 не попадают в velocity.
|
||
"""
|
||
comp_rows = [_comp_row(1), _comp_row(2)]
|
||
sales_rows = [
|
||
_sales_row(1, total_sqm=4000.0, months=4),
|
||
_sales_row(2, total_sqm=3000.0, months=4),
|
||
]
|
||
bucket_rows = [_bucket_row(1, "1", units_sold=20, sqm_sold=900.0)]
|
||
db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows, bucket_rows=bucket_rows)
|
||
|
||
with patch(
|
||
"app.services.site_finder.velocity._get_ekb_median",
|
||
return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
|
||
):
|
||
compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
|
||
|
||
# db.execute вызывается 3 раза: comp / sales / bucket. SQL берём из call_args.
|
||
executed_sql = [str(call.args[0]) for call in db.execute.call_args_list]
|
||
# comp-query НЕ трогает mapping; sales (idx 1) и bucket (idx 2) — должны иметь gate.
|
||
mapping_queries = [s for s in executed_sql if "objective_complex_mapping" in s]
|
||
assert len(mapping_queries) >= 2, "ожидаются sales + bucket запросы с mapping"
|
||
for sql in mapping_queries:
|
||
assert "cm.is_reviewed = TRUE" in sql
|
||
assert "cm.match_method = 'manual'" in sql
|
||
assert "cm.match_score >= 0.85" in sql
|
||
|
||
|
||
def test_sample_competitors_include_by_room_bucket():
|
||
"""sample_competitors каждого элемента содержит by_room_bucket."""
|
||
comp_rows = [_comp_row(1), _comp_row(2)]
|
||
sales_rows = [
|
||
_sales_row(1, total_sqm=6000.0, months=4),
|
||
_sales_row(2, total_sqm=4000.0, months=4),
|
||
]
|
||
bucket_rows = [
|
||
_bucket_row(1, "2", units_sold=30, sqm_sold=1800.0),
|
||
_bucket_row(2, "2", units_sold=20, sqm_sold=1200.0),
|
||
]
|
||
db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows, bucket_rows=bucket_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
|
||
for comp in result.sample_competitors:
|
||
assert "by_room_bucket" in comp
|
||
# obj_id=1 had bucket data
|
||
top = result.sample_competitors[0]
|
||
assert top["obj_id"] == 1
|
||
assert top["by_room_bucket"].get("2") == 30
|