lekss361
50f0674977
All checks were successful
Deploy Trade-In / changes (push) Successful in 11s
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 4m9s
Deploy Trade-In / build-backend (push) Successful in 1m54s
Deploy Trade-In / deploy (push) Successful in 1m51s
Deploy Trade-In / deploy-status (push) Successful in 1s
Deploy Trade-In / perimeter-smoke (push) Successful in 11s
88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
"""Тесты радиусных агрегатов ДТП (app/services/dtp_index.py, мигр. 294, #3410).
|
|
|
|
Coverage:
|
|
- _bbox_from_point: геометрически разумный bbox (south<north, west<east, точка внутри).
|
|
- compute_dtp_stats: graceful fallback при пустой dtp_incidents (status="unavailable",
|
|
БЕЗ сфабрикованных нулей-как-"ok") — MagicMock db.
|
|
- compute_dtp_stats: "ok" путь считает агрегаты из mapping-строки (MagicMock db).
|
|
- Статические asserts по _DTP_STATS_SQL: ST_DWithin, geom::geography, CAST(...),
|
|
make_interval (не bare :years::interval), нет bare :x:: кастов параметров.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from unittest.mock import MagicMock
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
|
|
|
from app.services import dtp_index as di
|
|
|
|
|
|
def test_bbox_from_point_contains_center() -> None:
|
|
lat, lon, radius_m = 56.8389, 60.6057, 1000
|
|
south, north, west, east = di._bbox_from_point(lat, lon, radius_m)
|
|
assert south < lat < north
|
|
assert west < lon < east
|
|
# Порядок величины: ~1000м ~ 0.009° по широте — не на порядок больше/меньше.
|
|
assert 0.001 < (north - south) < 0.05
|
|
assert 0.001 < (east - west) < 0.1
|
|
|
|
|
|
def test_compute_dtp_stats_unavailable_on_empty_table() -> None:
|
|
db = MagicMock()
|
|
db.execute.return_value.scalar.return_value = 0
|
|
|
|
result = di.compute_dtp_stats(db, 56.8389, 60.6057)
|
|
|
|
assert result.status == "unavailable"
|
|
assert result.incidents_count == 0
|
|
assert result.dead == 0
|
|
assert result.injured == 0
|
|
# Только один execute — count(*) проверка; агрегатный запрос не выполняется.
|
|
assert db.execute.call_count == 1
|
|
|
|
|
|
def test_compute_dtp_stats_ok_path() -> None:
|
|
db = MagicMock()
|
|
count_result = MagicMock()
|
|
count_result.scalar.return_value = 31253
|
|
|
|
stats_row = {
|
|
"incidents_count": 12,
|
|
"severe_count": 3,
|
|
"dead_total": 1,
|
|
"injured_total": 9,
|
|
}
|
|
stats_result = MagicMock()
|
|
stats_result.mappings.return_value.first.return_value = stats_row
|
|
|
|
db.execute.side_effect = [count_result, stats_result]
|
|
|
|
result = di.compute_dtp_stats(db, 56.8389, 60.6057, radius_m=800, years=3)
|
|
|
|
assert result.status == "ok"
|
|
assert result.radius_m == 800
|
|
assert result.years == 3
|
|
assert result.incidents_count == 12
|
|
assert result.severe_count == 3
|
|
assert result.dead == 1
|
|
assert result.injured == 9
|
|
|
|
stats_call = db.execute.call_args_list[1]
|
|
assert stats_call.args[0] is di._DTP_STATS_SQL
|
|
params = stats_call.args[1]
|
|
assert params["radius_m"] == 800
|
|
assert params["years"] == 3
|
|
|
|
|
|
def test_dtp_stats_sql_static_shape() -> None:
|
|
sql = re.sub(r"\s+", " ", str(di._DTP_STATS_SQL.text))
|
|
assert "ST_DWithin(" in sql
|
|
assert "geom::geography" in sql
|
|
assert "CAST(:radius_m AS double precision)" in sql
|
|
assert "CAST(:bbox_south AS double precision)" in sql
|
|
assert "make_interval(years => CAST(:years AS integer))" in sql
|
|
# psycopg v3 trap: bind-параметр не должен иметь :name::type сразу за собой.
|
|
assert not re.search(r":[a-zA-Z_]+::[a-zA-Z]", sql)
|