393 lines
15 KiB
Python
393 lines
15 KiB
Python
"""Тесты для GET /{cad_num}/connection-capacity (Forgejo #2119 Phase A).
|
||
|
||
FastAPI TestClient + mock сервиса get_connection_capacity — без Postgres.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
from unittest.mock import patch
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.main import app
|
||
|
||
_VALID_CAD = "66:41:0204016:10"
|
||
|
||
|
||
def _power_point(
|
||
name: str = "ПС 110/10 Уктус",
|
||
reserve: float | None = 15.0,
|
||
load_index: str | None = "open",
|
||
distance_m: float = 850.0,
|
||
reserve_unit: str | None = "МВА",
|
||
district: str | None = None,
|
||
reserve_note: str | None = None,
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"name": name,
|
||
"dzo_name": "Россети Урал",
|
||
"voltage_class": "110/10",
|
||
"load_index": load_index,
|
||
"installed_capacity_mva": 40.0,
|
||
"current_load_mva": 25.0,
|
||
"reserve_mva": reserve,
|
||
"reserve_unit": reserve_unit,
|
||
"district": district,
|
||
"reserve_asof": "2026-06-30",
|
||
"reserve_note": reserve_note,
|
||
"distance_m": distance_m,
|
||
"lat": 56.78,
|
||
"lon": 60.61,
|
||
}
|
||
|
||
|
||
def _heat_block(
|
||
systems: list[dict[str, Any]] | None = None,
|
||
total: float | None = 3.26,
|
||
) -> dict[str, Any]:
|
||
if systems is None:
|
||
systems = [
|
||
{
|
||
"org": "ЕТК",
|
||
"system_name": "СТ №54",
|
||
"reserve_gcal_h": 3.26,
|
||
"period": "2026-Q2",
|
||
}
|
||
]
|
||
return {"systems": systems, "total_reserve_gcal_h": total}
|
||
|
||
|
||
def _gas_block(
|
||
outlets_total: int = 1037,
|
||
outlets_deficit: int = 218,
|
||
outlets_needs_calc: int = 144,
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"city_grs": [
|
||
{
|
||
"grs_name": "ГРС Свердловская",
|
||
"design_capacity_th_m3_h": 210.0,
|
||
"free_capacity_th_m3_h": 40.5,
|
||
"free_capacity_pct": 20.2,
|
||
"upgrade_due": "2028",
|
||
"outputs_count": 3,
|
||
}
|
||
],
|
||
"total_free_th_m3_h": 40.5,
|
||
"outlets_total": outlets_total,
|
||
"outlets_deficit": outlets_deficit,
|
||
"outlets_needs_calc": outlets_needs_calc,
|
||
}
|
||
|
||
|
||
def _gas_outlet_point(
|
||
outlet_name: str = "Котельная (Складской проезд,4а)",
|
||
free: float | None = 0.512,
|
||
needs_calc: bool = False,
|
||
distance_m: float = 145.3,
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"outlet_name": outlet_name,
|
||
"consumer_type": "котельная",
|
||
"free_capacity_mln_m3": free,
|
||
"needs_calc": needs_calc,
|
||
"distance_m": distance_m,
|
||
"lat": 56.812345,
|
||
"lon": 60.611111,
|
||
}
|
||
|
||
|
||
def _make_response(
|
||
power_points: list[dict[str, Any]] | None = None,
|
||
water: list[dict[str, Any]] | None = None,
|
||
heat: dict[str, Any] | None = None,
|
||
gas: dict[str, Any] | None = None,
|
||
gas_outlet_points: list[dict[str, Any]] | None = None,
|
||
) -> dict[str, Any]:
|
||
if power_points is None:
|
||
power_points = [_power_point()]
|
||
if water is None:
|
||
water = [
|
||
{
|
||
"system_kind": "water",
|
||
"system_name": "ЗФС, ГСВ",
|
||
"reserve_thousand_m3_day": -168.5,
|
||
"note": "дефицит",
|
||
"period": "2026-Q1",
|
||
}
|
||
]
|
||
by_load_index: dict[str, int] = {}
|
||
for p in power_points:
|
||
li = p["load_index"] or "unknown"
|
||
by_load_index[li] = by_load_index.get(li, 0) + 1
|
||
nearest = next((p for p in power_points if p["reserve_mva"] and p["reserve_mva"] > 0), None)
|
||
return {
|
||
"power_points": power_points,
|
||
"power_summary": {
|
||
"total_power_points": len(power_points),
|
||
"by_load_index": by_load_index,
|
||
"nearest_with_reserve": nearest,
|
||
},
|
||
"water": water,
|
||
"gas": gas if gas is not None else _gas_block(),
|
||
"gas_outlet_points": (
|
||
gas_outlet_points if gas_outlet_points is not None else [_gas_outlet_point()]
|
||
),
|
||
"heat": heat if heat is not None else _heat_block(),
|
||
}
|
||
|
||
|
||
def test_connection_capacity_returns_power_and_water() -> None:
|
||
full = _make_response()
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
return_value=full,
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity")
|
||
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
assert len(body["power_points"]) == 1
|
||
assert body["power_points"][0]["voltage_class"] == "110/10"
|
||
assert body["power_summary"]["total_power_points"] == 1
|
||
assert body["power_summary"]["by_load_index"]["open"] == 1
|
||
assert body["power_summary"]["nearest_with_reserve"]["reserve_mva"] == pytest.approx(15.0)
|
||
# water: отрицательный резерв (дефицит) сохраняет знак.
|
||
assert body["water"][0]["reserve_thousand_m3_day"] == pytest.approx(-168.5)
|
||
assert body["water"][0]["system_name"] == "ЗФС, ГСВ"
|
||
|
||
|
||
def test_connection_capacity_empty() -> None:
|
||
empty = _make_response(power_points=[], water=[])
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
return_value=empty,
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity")
|
||
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
assert body["power_points"] == []
|
||
assert body["power_summary"]["total_power_points"] == 0
|
||
assert body["power_summary"]["nearest_with_reserve"] is None
|
||
assert body["water"] == []
|
||
|
||
|
||
def test_connection_capacity_no_reserve_nearest_null() -> None:
|
||
"""Все ЦП без положительного резерва → nearest_with_reserve=null."""
|
||
pts = [_power_point(reserve=None, load_index="closed")]
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
return_value=_make_response(power_points=pts),
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity")
|
||
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
assert body["power_summary"]["nearest_with_reserve"] is None
|
||
assert body["power_summary"]["by_load_index"]["closed"] == 1
|
||
|
||
|
||
def test_connection_capacity_parcel_not_found_404() -> None:
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
side_effect=ValueError("Участок '66:41:9999999:1' не найден в БД"),
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get("/api/v1/parcels/66:41:9999999:1/connection-capacity")
|
||
|
||
assert r.status_code == 404
|
||
assert "не найден" in r.json()["detail"]
|
||
|
||
|
||
def test_connection_capacity_passes_radius() -> None:
|
||
with patch("app.api.v1.parcels.get_connection_capacity") as mock_svc:
|
||
mock_svc.return_value = _make_response()
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity?radius_m=5000")
|
||
assert r.status_code == 200
|
||
call = mock_svc.call_args
|
||
assert call[0][2] == 5000 or call[1].get("radius_m") == 5000
|
||
|
||
|
||
def test_connection_capacity_radius_too_small_422() -> None:
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity?radius_m=50")
|
||
assert r.status_code == 422
|
||
|
||
|
||
def test_connection_capacity_radius_too_large_422() -> None:
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity?radius_m=20000")
|
||
assert r.status_code == 422
|
||
|
||
|
||
# ── Phase B2: heat-блок + reserve_unit/district + reserve_note ────────────────
|
||
|
||
|
||
def test_connection_capacity_heat_block_per_org() -> None:
|
||
"""Heat-блок: строки за последний период per-org + суммарный резерв. #2119 B2."""
|
||
heat = _heat_block(
|
||
systems=[
|
||
{"org": "ЕТК", "system_name": "СТ №54", "reserve_gcal_h": 3.26, "period": "2026-Q2"},
|
||
# Другая организация может публиковать свежий квартал раньше/позже —
|
||
# per-org MAX(period) сервиса это учитывает (тут просто передаём готовое).
|
||
{
|
||
"org": "Екатеринбургэнерго",
|
||
"system_name": "ТЭЦ Зона-1",
|
||
"reserve_gcal_h": -5.0, # дефицит — знак сохраняется
|
||
"period": "2026-Q1",
|
||
},
|
||
],
|
||
total=-1.74,
|
||
)
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
return_value=_make_response(heat=heat),
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity")
|
||
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
assert len(body["heat"]["systems"]) == 2
|
||
assert body["heat"]["systems"][0]["org"] == "ЕТК"
|
||
assert body["heat"]["systems"][0]["reserve_gcal_h"] == pytest.approx(3.26)
|
||
# дефицит (отрицательный резерв) сохраняет знак.
|
||
assert body["heat"]["systems"][1]["reserve_gcal_h"] == pytest.approx(-5.0)
|
||
assert body["heat"]["systems"][1]["period"] == "2026-Q1"
|
||
assert body["heat"]["total_reserve_gcal_h"] == pytest.approx(-1.74)
|
||
|
||
|
||
def test_connection_capacity_heat_graceful_absent() -> None:
|
||
"""Ключ 'heat' отсутствует (таблица не наполнена / миграция не применена) → null."""
|
||
resp = _make_response()
|
||
del resp["heat"]
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
return_value=resp,
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity")
|
||
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["heat"] is None
|
||
|
||
|
||
def test_connection_capacity_heat_empty_systems() -> None:
|
||
"""Таблица есть, но систем нет → пустой блок, total=null (graceful degrade)."""
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
return_value=_make_response(heat=_heat_block(systems=[], total=None)),
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity")
|
||
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
assert body["heat"]["systems"] == []
|
||
assert body["heat"]["total_reserve_gcal_h"] is None
|
||
|
||
|
||
def test_connection_capacity_power_reserve_unit_and_district() -> None:
|
||
"""power_points пробрасывает reserve_unit ('МВт' у ЕЭСК) + district. #2119 B2."""
|
||
pt = _power_point(
|
||
name="ПС 110/10 Свердловск",
|
||
reserve_unit="МВт",
|
||
district="Октябрьский",
|
||
)
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
return_value=_make_response(power_points=[pt]),
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity")
|
||
|
||
assert r.status_code == 200, r.text
|
||
p0 = r.json()["power_points"][0]
|
||
assert p0["reserve_unit"] == "МВт"
|
||
assert p0["district"] == "Октябрьский"
|
||
|
||
|
||
def test_connection_capacity_reserve_note_null_when_reserve_filled() -> None:
|
||
"""ЕЭСК теперь с числами → reserve_note NULL когда reserve_mva заполнен. #2119 B2."""
|
||
# Заполненный резерв → сервис отдаёт reserve_note=None (даже для ЕЭСК-строк).
|
||
pt = _power_point(reserve=12.5, reserve_note=None)
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
return_value=_make_response(power_points=[pt]),
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity")
|
||
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["power_points"][0]["reserve_note"] is None
|
||
|
||
|
||
# ── Phase B2 PR-4: газовые точки выхода (счётчики в gas + радиус-слой) ─────────
|
||
|
||
|
||
def test_connection_capacity_gas_outlet_counters() -> None:
|
||
"""gas-блок несёт счётчики точек выхода последнего среза (city-level). #2119 B2 PR-4."""
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
return_value=_make_response(),
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity")
|
||
|
||
assert r.status_code == 200, r.text
|
||
gas = r.json()["gas"]
|
||
assert gas["outlets_total"] == 1037
|
||
assert gas["outlets_deficit"] == 218
|
||
assert gas["outlets_needs_calc"] == 144
|
||
# ГРС-часть B1 не сломана.
|
||
assert gas["total_free_th_m3_h"] == pytest.approx(40.5)
|
||
|
||
|
||
def test_connection_capacity_gas_outlet_points_layer() -> None:
|
||
"""gas_outlet_points — гео-слой точек выхода: дистанция/needs_calc/дефицит. #2119 B2 PR-4."""
|
||
pts = [
|
||
_gas_outlet_point(distance_m=145.3, free=0.512, needs_calc=False),
|
||
# дефицит со знаком.
|
||
_gas_outlet_point(
|
||
outlet_name="Промплощадка №2", distance_m=980.4, free=-0.084, needs_calc=False
|
||
),
|
||
# «0.000*» → NULL мощность + needs_calc=True.
|
||
_gas_outlet_point(outlet_name="ТЭЦ узел", distance_m=1500.0, free=None, needs_calc=True),
|
||
]
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
return_value=_make_response(gas_outlet_points=pts),
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity")
|
||
|
||
assert r.status_code == 200, r.text
|
||
outlets = r.json()["gas_outlet_points"]
|
||
assert len(outlets) == 3
|
||
assert outlets[0]["distance_m"] == pytest.approx(145.3)
|
||
assert outlets[0]["free_capacity_mln_m3"] == pytest.approx(0.512)
|
||
assert outlets[1]["free_capacity_mln_m3"] == pytest.approx(-0.084) # дефицит-знак
|
||
assert outlets[2]["free_capacity_mln_m3"] is None
|
||
assert outlets[2]["needs_calc"] is True
|
||
|
||
|
||
def test_connection_capacity_gas_outlet_points_empty_default() -> None:
|
||
"""gas_outlet_points пуст (нет geom в радиусе) → []; ключ может отсутствовать → []."""
|
||
resp = _make_response(gas_outlet_points=[])
|
||
del resp["gas_outlet_points"] # сервис может не собрать ключ (миграция 184 не применена)
|
||
with patch(
|
||
"app.api.v1.parcels.get_connection_capacity",
|
||
return_value=resp,
|
||
):
|
||
client = TestClient(app)
|
||
r = client.get(f"/api/v1/parcels/{_VALID_CAD}/connection-capacity")
|
||
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["gas_outlet_points"] == []
|