Sadmin
3360f7a8ac
- services/osm_local.py: слой запросов к planet_osm_* (osm2pgsql, SRID 3857): roads/density (ST_Length merc × cos(lat) — проверено на merc/geo=1/cos), forest (честная площадь ST_Intersection), water (полигон/водоток), settlements (village/town/city). Сектор-«пирог» 45° строится в WGS84. - geo_service.get_zone_features: PostGIS-путь, Overpass — fallback (osm_available() по наличию planet_osm_line). - scripts/b16-import.sh: osm2pgsql --slim -C 800, Беларусь 333МБ pbf (Geofabrik, md5 проверен) → 4.5 мин на CT108; PostGIS 3.6 установлен в живой vector-postgres (apt, без смены образа). - Валидация: Минск/Полоцк/Браслав — правдоподобные плотности (7.9/2.9/1.5 км/км²), зоны различаются по направлениям; 32 сектора за 1.9 c (без сети). - Тесты: +8 (osm_local контракт, интеграция, отсутствие Overpass-вызовов), вне БД — skip. Всего 190 passed, 5 skipped. Критерий B16: build_search_zones без внешнего интернета — ВЫПОЛНЕН (единственный Overpass остаётся fallback-веткой при пустых таблицах).
103 lines
5.1 KiB
Python
103 lines
5.1 KiB
Python
"""B16: тесты локального OSM-слоя (PostGIS вместо Overpass).
|
|
|
|
Тесты работают и без БД: если DATABASE_URL недоступен или таблицы не
|
|
залиты, osm_available() возвращает False и fallback-ветка geo_service
|
|
ведёт себя по-старому (проверяется моком).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from services import geo_service
|
|
from services import osm_local
|
|
|
|
|
|
class TestOsmLocal:
|
|
def test_sector_wkt_is_valid_pie(self):
|
|
"""WKT сектора — пирог: центр + дуга, замкнутый."""
|
|
wkt = osm_local._sector_wkt_3857(53.9, 27.56, 1000, 'N')
|
|
assert wkt.startswith('POLYGON((')
|
|
assert wkt.endswith('))')
|
|
# Центр входит дважды (начало и конец)
|
|
assert wkt.count('27.5600000') >= 2 or '27.56 ' in wkt
|
|
|
|
def test_sector_geometry_valid(self):
|
|
"""PostGIS-валидность: сектор не самопересекается (если БД доступна)."""
|
|
if not osm_local.osm_available():
|
|
pytest.skip('OSM PostGIS недоступен')
|
|
import sqlalchemy
|
|
from sqlalchemy import text
|
|
e = sqlalchemy.create_engine(osm_local._PG_URL)
|
|
wkt = osm_local._sector_wkt_3857(53.9, 27.56, 1000, 'NE')
|
|
with e.connect() as c:
|
|
valid = c.execute(text(
|
|
"SELECT ST_IsValid(ST_GeomFromText(:w, 4326))"), {'w': wkt}
|
|
).scalar()
|
|
assert valid
|
|
|
|
def test_features_contract(self):
|
|
"""Контракт Zone-фич: все ключи, типы, диапазоны."""
|
|
if not osm_local.osm_available():
|
|
pytest.skip('OSM PostGIS недоступен')
|
|
f = osm_local.get_zone_features_postgis(53.9, 27.56, 'N', 1000)
|
|
assert set(f) == {'roads_km', 'road_density', 'water_distance_km',
|
|
'settlement_distance_km', 'forest_pct'}
|
|
assert f['roads_km'] >= 0
|
|
assert 0 <= f['forest_pct'] <= 100
|
|
assert f['water_distance_km'] is None or f['water_distance_km'] > 0
|
|
|
|
def test_zones_differ_between_directions(self):
|
|
"""Критерий B16: соседние направления дают РАЗНЫЕ характеристики."""
|
|
if not osm_local.osm_available():
|
|
pytest.skip('OSM PostGIS недоступен')
|
|
n = osm_local.get_zone_features_postgis(53.9, 27.56, 'N', 1000)
|
|
ne = osm_local.get_zone_features_postgis(53.9, 27.56, 'NE', 1000)
|
|
assert (n['roads_km'], n['forest_pct']) != (ne['roads_km'], ne['forest_pct'])
|
|
|
|
def test_wild_point_outside_data(self):
|
|
"""Точка вне Беларуси → нули (не падает)."""
|
|
if not osm_local.osm_available():
|
|
pytest.skip('OSM PostGIS недоступен')
|
|
f = osm_local.get_zone_features_postgis(40.0, 0.0, 'N', 1000)
|
|
assert f['roads_km'] == 0.0
|
|
assert f['forest_pct'] == 0.0
|
|
|
|
|
|
class TestGeoServiceIntegration:
|
|
async def test_build_search_zones_uses_postgis_when_available(self):
|
|
"""osm_available()=True → Overpass НЕ вызывается."""
|
|
with patch.object(geo_service, 'osm_available', return_value=True), \
|
|
patch.object(geo_service, 'get_zone_features_postgis') as mock_pg, \
|
|
patch.object(geo_service, 'get_zone_features', wraps=geo_service.get_zone_features):
|
|
mock_pg.side_effect = lambda lat, lon, d, r: {
|
|
'roads_km': 1.0, 'road_density': 0.8, 'water_distance_km': 0.5,
|
|
'settlement_distance_km': 0.3, 'forest_pct': 10.0,
|
|
}
|
|
zones = await geo_service.build_search_zones(53.9, 27.56, {}, 1.08)
|
|
assert len(zones) == 32
|
|
assert mock_pg.call_count == 32
|
|
assert zones[0].forest_pct == 10.0
|
|
|
|
async def test_build_search_zones_falls_back_to_overpass(self):
|
|
"""osm_available()=False → идёт старый Overpass-путь (мокаем get_zone_features)."""
|
|
with patch.object(geo_service, 'osm_available', return_value=False), \
|
|
patch.object(geo_service, 'get_zone_features') as mock_get:
|
|
mock_get.side_effect = lambda lat, lon, d, r: {
|
|
'roads_km': 2.0, 'road_density': 1.6, 'water_distance_km': 0.7,
|
|
'settlement_distance_km': 0.4, 'forest_pct': 5.0,
|
|
}
|
|
zones = await geo_service.build_search_zones(53.9, 27.56, {}, 1.08)
|
|
assert len(zones) == 32
|
|
assert mock_get.call_count == 32
|
|
assert zones[0].forest_pct == 5.0
|
|
|
|
def test_no_overpass_calls_with_local_data(self):
|
|
"""Критерий B16: с локальными данными query_overpass не дёргается."""
|
|
if not osm_local.osm_available():
|
|
pytest.skip('OSM PostGIS недоступен')
|
|
with patch.object(geo_service, 'query_overpass') as mock_q:
|
|
import asyncio
|
|
zones = asyncio.run(geo_service.build_search_zones(53.9, 27.56, {}, 1.08))
|
|
mock_q.assert_not_called() |