lekss361
f3e10a04ad
All checks were successful
Deploy Trade-In / changes (push) Successful in 12s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 1m39s
Deploy Trade-In / build-backend (push) Successful in 8m5s
Deploy Trade-In / deploy (push) Successful in 52s
249 lines
8.9 KiB
Python
249 lines
8.9 KiB
Python
"""Тесты API «доля квартир дома в продаже» (мигр. 143).
|
||
|
||
Two layers:
|
||
1. SQL-билдеры (чистые функции) — shape, CAST(:x AS type), whitelist-сортировка,
|
||
clamp лимита, фильтры. БД не нужна.
|
||
2. Contract — минимальный app с замоканным get_db: 200 + правильная форма ответа
|
||
(RBAC middleware не монтируем, mirror trade_in_app из test_estimate_idor).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from unittest.mock import MagicMock
|
||
|
||
# psycopg v3 driver требует DATABASE_URL до импорта app.core.db.
|
||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.services.buildings_query import (
|
||
ALLOWED_SORTS,
|
||
HISTOGRAM_BUCKETS,
|
||
build_listings_query,
|
||
build_sale_share_query,
|
||
build_summary_histogram_query,
|
||
build_summary_scalars_query,
|
||
)
|
||
|
||
# ── SQL-билдеры ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_sale_share_minimal_defaults():
|
||
sql, args = build_sale_share_query()
|
||
assert "FROM v_building_sale_share" in sql
|
||
assert "sale_share_pct IS NOT NULL" in sql
|
||
assert "sale_share_pct >= CAST(:min_pct AS numeric)" in sql
|
||
assert "(sale_share_pct > 100) AS over_100" in sql
|
||
assert "COALESCE(short_address, full_address, address) AS address" in sql
|
||
assert args["min_pct"] == 5.0
|
||
assert args["limit"] == 200
|
||
# default sort = share_desc
|
||
assert "ORDER BY sale_share_pct DESC NULLS LAST" in sql
|
||
|
||
|
||
def test_sale_share_all_filters():
|
||
sql, args = build_sale_share_query(
|
||
min_pct=10,
|
||
max_pct=80,
|
||
city="Екатеринбург",
|
||
price_min=3_000_000,
|
||
price_max=12_000_000,
|
||
year_min=1990,
|
||
year_max=2015,
|
||
house_type="panel",
|
||
sort="price_asc",
|
||
)
|
||
assert "sale_share_pct <= CAST(:max_pct AS numeric)" in sql
|
||
assert "ILIKE CAST(:city_like AS text)" in sql
|
||
assert args["city_like"] == "%Екатеринбург%"
|
||
assert "median_price_rub >= CAST(:price_min AS bigint)" in sql
|
||
assert "median_price_rub <= CAST(:price_max AS bigint)" in sql
|
||
assert "year_built >= CAST(:year_min AS integer)" in sql
|
||
assert "year_built <= CAST(:year_max AS integer)" in sql
|
||
assert "house_type = CAST(:house_type AS text)" in sql
|
||
assert "ORDER BY median_price_rub ASC NULLS LAST" in sql
|
||
|
||
|
||
def test_sale_share_no_double_colon_cast():
|
||
"""psycopg v3: ни одного `:name::type` — только CAST(...)."""
|
||
sql, _ = build_sale_share_query(city="x", price_min=1, year_min=2000, house_type="brick")
|
||
import re
|
||
|
||
assert re.search(r":[a-z_]+::[a-z]", sql) is None
|
||
|
||
|
||
def test_sale_share_limit_clamped():
|
||
_, args_hi = build_sale_share_query(limit=99999)
|
||
assert args_hi["limit"] == 1000
|
||
_, args_lo = build_sale_share_query(limit=0)
|
||
assert args_lo["limit"] == 1
|
||
_, args_neg = build_sale_share_query(limit=-5)
|
||
assert args_neg["limit"] == 1
|
||
|
||
|
||
@pytest.mark.parametrize("sort", sorted(ALLOWED_SORTS))
|
||
def test_sale_share_sort_whitelisted(sort: str):
|
||
sql, _ = build_sale_share_query(sort=sort)
|
||
assert "ORDER BY" in sql
|
||
# tiebreaker house_id присутствует во всех сортировках
|
||
assert "house_id" in sql.split("ORDER BY", 1)[1]
|
||
|
||
|
||
def test_sale_share_unknown_sort_falls_back():
|
||
"""Несуществующий sort не интерполируется — fallback на share_desc."""
|
||
sql, _ = build_sale_share_query(sort="DROP TABLE houses")
|
||
assert "DROP TABLE" not in sql
|
||
assert "ORDER BY sale_share_pct DESC NULLS LAST" in sql
|
||
|
||
|
||
def test_listings_query_shape():
|
||
sql, args = build_listings_query(42)
|
||
assert "FROM listings" in sql
|
||
assert "house_id_fk = CAST(:house_id AS bigint)" in sql
|
||
assert "listing_segment = 'vtorichka'" in sql
|
||
assert "is_active" in sql
|
||
assert "ORDER BY price_rub" in sql
|
||
assert "id AS listing_id" in sql
|
||
assert args["house_id"] == 42
|
||
|
||
|
||
def test_summary_scalars_query_shape():
|
||
sql = build_summary_scalars_query()
|
||
assert "total_secondary_buildings" in sql
|
||
assert "FILTER (WHERE sale_share_pct IS NOT NULL)" in sql
|
||
assert "NULLIF(count(*), 0)" in sql
|
||
assert "percentile_cont(0.95)" in sql
|
||
|
||
|
||
def test_summary_histogram_buckets_cover_spec():
|
||
sql = build_summary_histogram_query()
|
||
for boundary in ("'0-5'", "'5-10'", "'10-20'", "'20-30'", "'30-50'", "'50-100'", "'100+'"):
|
||
assert boundary in sql
|
||
assert HISTOGRAM_BUCKETS == ("0-5", "5-10", "10-20", "20-30", "30-50", "50-100", "100+")
|
||
|
||
|
||
# ── Contract (минимальный app, замоканный get_db) ─────────────────────────────
|
||
|
||
|
||
@pytest.fixture()
|
||
def buildings_app() -> FastAPI:
|
||
"""FastAPI mounting только buildings router (без RBAC middleware)."""
|
||
from app.api.v1 import buildings as buildings_module
|
||
|
||
application = FastAPI()
|
||
application.include_router(buildings_module.router, prefix="/api/v1/buildings")
|
||
return application
|
||
|
||
|
||
def _client_with_rows(app: FastAPI, *execute_results: object) -> TestClient:
|
||
"""Override get_db → mock, где db.execute даёт execute_results по порядку."""
|
||
from app.core.db import get_db
|
||
|
||
db = MagicMock()
|
||
if len(execute_results) == 1:
|
||
db.execute.return_value = execute_results[0]
|
||
else:
|
||
db.execute.side_effect = list(execute_results)
|
||
|
||
def _override_db():
|
||
yield db
|
||
|
||
app.dependency_overrides[get_db] = _override_db
|
||
return TestClient(app)
|
||
|
||
|
||
def _mapped(rows: object) -> MagicMock:
|
||
m = MagicMock()
|
||
m.mappings.return_value.all.return_value = rows
|
||
m.mappings.return_value.one.return_value = rows
|
||
return m
|
||
|
||
|
||
def test_get_sale_share_list_200(buildings_app: FastAPI):
|
||
row = {
|
||
"house_id": 7,
|
||
"address": "ул. Ленина, 1",
|
||
"lat": 56.8,
|
||
"lon": 60.6,
|
||
"sale_share_pct": 12.5,
|
||
"over_100": False,
|
||
"active_secondary": 3,
|
||
"flat_count_effective": 24,
|
||
"gar_match_method": None,
|
||
"median_price_rub": 5_000_000,
|
||
"median_price_per_m2": 95_000,
|
||
"avg_days_on_market": 41.0,
|
||
"year_built": 1999,
|
||
"house_type": "panel",
|
||
"total_floors": 9,
|
||
"series_name": None,
|
||
"is_emergency": False,
|
||
}
|
||
client = _client_with_rows(buildings_app, _mapped([row]))
|
||
resp = client.get("/api/v1/buildings/sale-share?min_pct=10")
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert len(body) == 1
|
||
assert body[0]["house_id"] == 7
|
||
assert body[0]["over_100"] is False
|
||
assert body[0]["sale_share_pct"] == 12.5
|
||
|
||
|
||
def test_get_sale_share_empty_when_no_gar(buildings_app: FastAPI):
|
||
"""ГАР не загружен → пустой список, но 200 (форма, не падение)."""
|
||
client = _client_with_rows(buildings_app, _mapped([]))
|
||
resp = client.get("/api/v1/buildings/sale-share")
|
||
assert resp.status_code == 200
|
||
assert resp.json() == []
|
||
|
||
|
||
def test_get_building_listings_200(buildings_app: FastAPI):
|
||
row = {
|
||
"listing_id": 101,
|
||
"source": "avito",
|
||
"source_url": "https://avito.ru/101",
|
||
"price_rub": 4_900_000,
|
||
"price_per_m2": 92_000,
|
||
"rooms": 2,
|
||
"area_m2": 53.2,
|
||
"floor": 4,
|
||
"total_floors": 9,
|
||
"days_on_market": 30,
|
||
"listing_date": "2026-05-01",
|
||
"address": "ул. Ленина, 1",
|
||
}
|
||
client = _client_with_rows(buildings_app, _mapped([row]))
|
||
resp = client.get("/api/v1/buildings/7/listings")
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body[0]["listing_id"] == 101
|
||
assert body[0]["source"] == "avito"
|
||
|
||
|
||
def test_get_summary_zero_fills_histogram(buildings_app: FastAPI):
|
||
scalars = _mapped(
|
||
{
|
||
"total_secondary_buildings": 10,
|
||
"buildings_with_denominator": 4,
|
||
"coverage_pct": 40.0,
|
||
"max_pct": 250.0,
|
||
"p95_pct": 33.3,
|
||
}
|
||
)
|
||
hist = _mapped([{"bucket": "0-5", "count": 2}, {"bucket": "100+", "count": 1}])
|
||
client = _client_with_rows(buildings_app, scalars, hist)
|
||
resp = client.get("/api/v1/buildings/sale-share/summary")
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["total_secondary_buildings"] == 10
|
||
assert body["buildings_with_denominator"] == 4
|
||
assert body["coverage_pct"] == 40.0
|
||
# все 7 корзин присутствуют, недостающие — count 0, порядок фиксирован
|
||
assert [b["bucket"] for b in body["histogram"]] == list(HISTOGRAM_BUCKETS)
|
||
counts = {b["bucket"]: b["count"] for b in body["histogram"]}
|
||
assert counts["0-5"] == 2
|
||
assert counts["100+"] == 1
|
||
assert counts["20-30"] == 0
|