gendesign/backend/tests/api/v1/test_locations.py
Light1YT 8da1c00138
All checks were successful
CI / changes (push) Successful in 6s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 6s
Deploy / changes (push) Successful in 6s
CI / frontend-tests (pull_request) Has been skipped
Deploy / build-backend (push) Successful in 1m29s
Deploy / build-frontend (push) Has been skipped
Deploy / build-worker (push) Successful in 2m42s
Deploy / deploy (push) Successful in 1m11s
CI / backend-tests (push) Successful in 6m27s
CI / backend-tests (pull_request) Successful in 6m22s
feat(location): district-level Location entity + indices (#948 part B)
Promote district to a first-class `location` entity (ТЗ §8.2), ADDITIVE — no
district->FK refactor. New `location` table keyed by district_name (joinable by
string), carrying 4 normalized [0,1] indices (NULL when no data, never 0):
- infra: _district_poi_score / _city_avg_poi_score (per-district POI aggregate)
- competition: market_metrics.overstock_index (available/stuck competing supply
  — orthogonal to demand; NOT sell-through, which is market-heat correlated w/ demand)
- demand: market_metrics.unit_velocity (saturating /50)
- future_supply: future_supply_pressure.index (passthrough, already 0..1)

- data/sql/146_location.sql: idempotent table + UNIQUE(district_name) + range
  CHECK + centroid GIST
- services/site_finder/locations.py: compute_location_indices (reuses forecast
  per-district fns) + refresh_locations (SAVEPOINT per-row, CAST, ON CONFLICT)
- workers/tasks/location_refresh.py + beat (Mon 07:00 MSK, after supply-layers)
- api/v1/locations.py: read-only GET list + GET by name (analyst+admin via rbac,
  frontend pilot-gated)
- tests: 34 (normalization 0..1/null, competition⊥demand orthogonality, idempotent
  upsert, read API list/by-name/404)

Part of #948 (Part A insight shipped #1164).
2026-06-08 13:28:19 +05:00

162 lines
6.1 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Тесты read-only API для location (#948 Part B, ТЗ §8.2).
Покрывает:
1. GET /api/v1/locations → LocationList (total + rows с индексами)
2. GET /api/v1/locations/{district_name} → 200 (одна локация)
3. GET /api/v1/locations/{district_name} → 404 если нет
4. Индексы-None (нет данных) корректно сериализуются как null
5. Нет write-методов: POST/PUT/DELETE → 405 (read-only сущность)
Стратегия mock: сервисные функции (list_locations / get_location) патчим через
unittest.mock.patch; get_db переопределяем заглушкой (сервис запатчен — БД не
трогается). settings.testing=True (conftest) отключает rbac_guard.
"""
from __future__ import annotations
from datetime import UTC, datetime
from unittest.mock import MagicMock, patch
from fastapi.testclient import TestClient
from app.main import app
from app.schemas.location import LocationList, LocationOut
_TS = datetime(2026, 6, 7, 10, 0, 0, tzinfo=UTC)
_DISTRICT = "Кировский"
def _make_out(
district_name: str = _DISTRICT,
infra: float | None = 0.82,
competition: float | None = 0.6,
demand: float | None = 0.5,
future_supply: float | None = 0.3,
lon: float | None = 60.6,
lat: float | None = 56.84,
) -> LocationOut:
return LocationOut(
id=1,
district_name=district_name,
region=None,
lon=lon,
lat=lat,
infra_index=infra,
competition_index=competition,
demand_index=demand,
future_supply_index=future_supply,
indices_computed_at=_TS,
created_at=_TS,
updated_at=_TS,
)
def _override_db():
"""Заглушка get_db — сервис запатчен, реальная сессия не нужна."""
def _get_db_override():
yield MagicMock()
return _get_db_override
# ── 1. LIST ─────────────────────────────────────────────────────────────────────
def test_list_locations_returns_list_with_indices() -> None:
from app.core.db import get_db
expected = LocationList(
total=2,
rows=[_make_out("Кировский"), _make_out("Ленинский", infra=None)],
)
app.dependency_overrides[get_db] = _override_db()
with patch("app.api.v1.locations.list_locations", return_value=expected) as mock_list:
client = TestClient(app)
resp = client.get("/api/v1/locations")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["total"] == 2
assert len(body["rows"]) == 2
assert body["rows"][0]["district_name"] == "Кировский"
assert body["rows"][0]["competition_index"] == 0.6
# None-индекс сериализуется как null (нет данных, не 0).
assert body["rows"][1]["infra_index"] is None
mock_list.assert_called_once()
def test_list_locations_empty() -> None:
from app.core.db import get_db
app.dependency_overrides[get_db] = _override_db()
with patch("app.api.v1.locations.list_locations", return_value=LocationList(total=0, rows=[])):
client = TestClient(app)
resp = client.get("/api/v1/locations")
assert resp.status_code == 200, resp.text
assert resp.json() == {"total": 0, "rows": []}
# ── 2 + 3. GET BY NAME ──────────────────────────────────────────────────────────
def test_get_location_by_name_returns_200() -> None:
from app.core.db import get_db
app.dependency_overrides[get_db] = _override_db()
with patch("app.api.v1.locations.get_location", return_value=_make_out()) as mock_get:
client = TestClient(app)
resp = client.get(f"/api/v1/locations/{_DISTRICT}")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["district_name"] == _DISTRICT
assert body["demand_index"] == 0.5
assert body["future_supply_index"] == 0.3
assert body["lon"] == 60.6
# district_name пробрасывается из path в сервис.
assert mock_get.call_args[0][1] == _DISTRICT
def test_get_location_unknown_returns_404() -> None:
from app.core.db import get_db
app.dependency_overrides[get_db] = _override_db()
with patch("app.api.v1.locations.get_location", return_value=None):
client = TestClient(app)
resp = client.get("/api/v1/locations/НесуществующийРайон")
assert resp.status_code == 404, resp.text
assert resp.json()["detail"] == "Location not found"
def test_get_location_null_indices_serialize_as_null() -> None:
from app.core.db import get_db
app.dependency_overrides[get_db] = _override_db()
out = _make_out(
infra=None, competition=None, demand=None, future_supply=None, lon=None, lat=None
)
with patch("app.api.v1.locations.get_location", return_value=out):
client = TestClient(app)
resp = client.get(f"/api/v1/locations/{_DISTRICT}")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["infra_index"] is None
assert body["competition_index"] is None
assert body["demand_index"] is None
assert body["future_supply_index"] is None
assert body["lon"] is None
# ── 5. READ-ONLY: no write methods ──────────────────────────────────────────────
def test_post_location_not_allowed() -> None:
"""Локации вычисляются, не вводятся → POST не зарегистрирован (405)."""
client = TestClient(app)
resp = client.post("/api/v1/locations", json={"district_name": "Кировский"})
assert resp.status_code == 405
def test_delete_location_not_allowed() -> None:
client = TestClient(app)
resp = client.delete(f"/api/v1/locations/{_DISTRICT}")
assert resp.status_code == 405