fix(site-finder): optimize zone-regulation resolve in /analyze (#1850) #1901

Merged
bot-backend merged 1 commit from fix/analyze-zone-resolve-perf into main 2026-06-25 09:12:42 +00:00
6 changed files with 398 additions and 12 deletions

View file

@ -438,10 +438,10 @@ _GEOM_LABEL_MICRO_HA = 0.05 # ниже → label "микро" (комбинир
_GEOM_LABEL_GOOD = 0.7 _GEOM_LABEL_GOOD = 0.7
_GEOM_LABEL_MEDIUM = 0.4 _GEOM_LABEL_MEDIUM = 0.4
# ПЗЗ-градрегламент в analyze (#1067): bounded timeout (сек) для live-geoportal вызова # ПЗЗ-градрегламент в analyze (#1067/#1850): bounded timeout для live-geoportal вызова на
# на cache-miss. Короткий, чтобы медленный геопортал не вешал hot-path analyze — # cache-miss теперь единый — settings.geoportal_timeout_s (раньше локальная константа 3s
# cache-hit (норма для ЕКБ) сетевых вызовов не делает вообще. # расходилась с ird_analyze 4s для того же резолвера). cache-hit (норма для ЕКБ) +
_ZONE_REGULATION_TIMEOUT_S = 3 # мемоизация coord→zone_index сетевых вызовов не делают вообще.
def _polygon_suitability(geom_wkt: str) -> dict[str, Any]: def _polygon_suitability(geom_wkt: str) -> dict[str, Any]:
@ -2078,7 +2078,7 @@ def analyze_parcel(
# Резолвим числовые предельные параметры (КСИТ/max_far, высота, этажность, # Резолвим числовые предельные параметры (КСИТ/max_far, высота, этажность,
# %застройки, min площадь ЗУ) по centroid через get_or_fetch_zone_regulation: # %застройки, min площадь ЗУ) по centroid через get_or_fetch_zone_regulation:
# cache-first (кэш прогрет для ЕКБ), live-geoportal только на miss и закапан # cache-first (кэш прогрет для ЕКБ), live-geoportal только на miss и закапан
# коротким timeout'ом (_ZONE_REGULATION_TIMEOUT_S), чтобы медленный геопортал не # единым timeout'ом (settings.geoportal_timeout_s), чтобы медленный геопортал не
# вешал analyze. Hot-path-safe: любой сбой/таймаут → поля остаются None, analyze # вешал analyze. Hot-path-safe: любой сбой/таймаут → поля остаются None, analyze
# не падает. # не падает.
# #
@ -2094,6 +2094,12 @@ def analyze_parcel(
# raw_props.subcategory, по которым gate детектит жильё), только ДОБАВЛЯЕМ # raw_props.subcategory, по которым gate детектит жильё), только ДОБАВЛЯЕМ
# regulation-поля. Расширение dump-резидентности — OUT OF SCOPE (PR-B). # regulation-поля. Расширение dump-резидентности — OUT OF SCOPE (PR-B).
_nspd_zoning = nspd_dump_data.get("nspd_zoning") _nspd_zoning = nspd_dump_data.get("nspd_zoning")
# Регламент, резолвленный ЗДЕСЬ (zoning-merge), переиспользуется ИРД-блоком (#1850 Fix 2):
# _zone_regulation_resolved=True означает «этот блок отработал и _regulation валиден
# (dict|None)» → build_ird_analyze_block получит готовое значение, без повторного резолва.
# Когда блок НЕ отработал (флаг off / не-ЕКБ / нет centroid) — ИРД-блок резолвит сам.
_regulation: dict[str, Any] | None = None
_zone_regulation_resolved = False
# Region-guard: EKB-геопортал покрывает ТОЛЬКО Екатеринбург (КН 66:41). Для # Region-guard: EKB-геопортал покрывает ТОЛЬКО Екатеринбург (КН 66:41). Для
# участков вне ЕКБ zone_index_at всё равно вернул бы None — экономим WFS-round-trip # участков вне ЕКБ zone_index_at всё равно вернул бы None — экономим WFS-round-trip
# (резолвер теперь запускается на каждом analyze, а не только при dump-зоне). # (резолвер теперь запускается на каждом analyze, а не только при dump-зоне).
@ -2103,8 +2109,9 @@ def analyze_parcel(
and centroid_lat is not None and centroid_lat is not None
and centroid_lon is not None and centroid_lon is not None
): ):
_zone_regulation_resolved = True
try: try:
_client = EKBGeoportalClient(timeout=_ZONE_REGULATION_TIMEOUT_S) _client = EKBGeoportalClient(timeout=settings.geoportal_timeout_s)
_regulation = get_or_fetch_zone_regulation( _regulation = get_or_fetch_zone_regulation(
db, centroid_lon, centroid_lat, client=_client db, centroid_lon, centroid_lat, client=_client
) )
@ -3282,8 +3289,15 @@ def analyze_parcel(
# сбой не меняет успех/форму остального ответа. Additive: extra="allow" в AnalyzeResponse. # сбой не меняет успех/форму остального ответа. Additive: extra="allow" в AnalyzeResponse.
if settings.enable_ird_analyze: if settings.enable_ird_analyze:
try: try:
# #1850 Fix 2: если zoning-merge блок выше уже резолвил регламент по тому же
# centroid — передаём его готовым (_regulation, dict|None), ИРД-блок не делает
# повторный резолв/DB-read. Если блок НЕ отработал (флаг off / не-ЕКБ / нет
# centroid) — НЕ передаём kwarg, ИРД-блок резолвит сам (Fix 1 снимает network-цену).
_ird_kwargs: dict[str, Any] = {}
if _zone_regulation_resolved:
_ird_kwargs["resolved_zone_regulation"] = _regulation
result_payload["ird"] = build_ird_analyze_block( result_payload["ird"] = build_ird_analyze_block(
db, geom_wkt, centroid_lon, centroid_lat, cad_num db, geom_wkt, centroid_lon, centroid_lat, cad_num, **_ird_kwargs
) )
except Exception: except Exception:
logger.warning( logger.warning(

View file

@ -217,6 +217,14 @@ class Settings(BaseSettings):
# полностью пропускает резолв — поведение analyze без изменений. # полностью пропускает резолв — поведение analyze без изменений.
enable_zoning_regulation_in_analyze: bool = True enable_zoning_regulation_in_analyze: bool = True
# Единый bounded-timeout (сек) для live geoportal-вызовов резолвера ПЗЗ-регламента в
# hot-пути analyze (#1850). Раньше было два расходящихся значения: parcels.py=3s,
# ird_analyze.py=4s для ОДНОГО и того же резолвера. Унифицировано в 4s (более безопасный
# запас). Оба колл-сайта работают cache-first (+ мемоизация coord→zone_index), поэтому
# живые вызовы редки. Дефолт клиента EKBGeoportalClient (20s) слишком долог для sync-хэндлера.
# Override через env GEOPORTAL_TIMEOUT_S.
geoportal_timeout_s: int = 4
# Area-gate для blocker'а «инженерная/утилитарная охранная зона» в gate_verdict. # Area-gate для blocker'а «инженерная/утилитарная охранная зона» в gate_verdict.
# ЗОУИТ охранной зоны инж.сети (ЛЭП/газ/трубопровод/тепло/электро) в РФ ограничивает # ЗОУИТ охранной зоны инж.сети (ЛЭП/газ/трубопровод/тепло/электро) в РФ ограничивает
# застройку ВНУТРИ полосы (отступы, запрет капстроя над линией), а НЕ стерилизует весь # застройку ВНУТРИ полосы (отступы, запрет капстроя над линией), а НЕ стерилизует весь

View file

@ -30,6 +30,7 @@ from typing import Any
from sqlalchemy import text from sqlalchemy import text
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.core.config import settings
from app.services.scrapers.ekb_geoportal_client import EKBGeoportalClient from app.services.scrapers.ekb_geoportal_client import EKBGeoportalClient
from app.services.site_finder.functional_zone_lookup import parcel_functional_zones from app.services.site_finder.functional_zone_lookup import parcel_functional_zones
from app.services.site_finder.granddoc_lookup import parcel_granddoc from app.services.site_finder.granddoc_lookup import parcel_granddoc
@ -44,8 +45,10 @@ from app.services.site_finder.zone_regulation import get_or_fetch_zone_regulatio
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Короткий timeout для geoportal-вызовов в hot-пути analyze (default клиента — 20s, слишком долго). # Sentinel: «регламент НЕ передан извне» — отличает «caller не резолвил» от «резолв дал None»
_GEOPORTAL_TIMEOUT_S = 4 # (None — валидный результат: в точке нет терзоны). См. resolved_zone_regulation в
# build_ird_analyze_block (#1850 Fix 2).
_UNSET: Any = object()
# SQL: читаем КРТ-геометрии из БД (м.140 ekb_krt_geometry) по точке centroid участка. # SQL: читаем КРТ-геометрии из БД (м.140 ekb_krt_geometry) по точке centroid участка.
# ST_Intersects с GIST → дёшево (~1ms), без сетевой задержки. # ST_Intersects с GIST → дёшево (~1ms), без сетевой задержки.
@ -112,6 +115,7 @@ def build_ird_analyze_block(
cad_num: str | None = None, cad_num: str | None = None,
*, *,
client: EKBGeoportalClient | None = None, client: EKBGeoportalClient | None = None,
resolved_zone_regulation: Any = _UNSET,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Собрать ИРД-блок участка для analyze (#1067 D9b). """Собрать ИРД-блок участка для analyze (#1067 D9b).
@ -123,6 +127,13 @@ def build_ird_analyze_block(
None reservations=[]. None reservations=[].
client: переиспользуемый EKBGeoportalClient (для тестов/одного TLS-пула). По умолчанию client: переиспользуемый EKBGeoportalClient (для тестов/одного TLS-пула). По умолчанию
создаётся с укороченным timeout. создаётся с укороченным timeout.
resolved_zone_regulation: УЖЕ резолвленный ПЗЗ-регламент (dict|None) из parcels.py
zoning-merge блока (#1850 Fix 2). Если передан (включая None — «зоны нет») —
``_zone_regulation_at`` НЕ вызывается, используется переданное значение (исключает
повторный DB-read кэша на тот же centroid). Если НЕ передан (``_UNSET``, дефолт)
блок резолвит сам (путь «enable_zoning_regulation_in_analyze off, enable_ird_analyze
on»). Network-стоимость второго резолва уже снята мемоизацией coordzone_index
(Fix 1); этот параметр убирает оставшийся лишний DB-read.
Returns: Returns:
dict: ``ird_overlaps``/``ird_by_kind`` (DB-overlay, incl opportunity_*), dict: ``ird_overlaps``/``ird_by_kind`` (DB-overlay, incl opportunity_*),
@ -190,6 +201,12 @@ def build_ird_analyze_block(
block["functional_zones"] = fz block["functional_zones"] = fz
block["functional_zone"] = fz[0] if fz else None block["functional_zone"] = fz[0] if fz else None
# #1850 Fix 2: если регламент УЖЕ резолвлен в parcels.py (zoning-merge блок), берём его
# без повторного резолва — исключает лишний DB-read кэша на тот же centroid. None — тоже
# валидное переданное значение («зоны нет»), поэтому отличаем «передан» через _UNSET.
if resolved_zone_regulation is not _UNSET:
block["zone_regulation"] = resolved_zone_regulation
# guard активен только в unit-тестах; в prod centroid_lon/lat всегда float # guard активен только в unit-тестах; в prod centroid_lon/lat всегда float
# (EKB-fallback в parcels.py). # (EKB-fallback в parcels.py).
if lon is not None and lat is not None: if lon is not None and lat is not None:
@ -198,9 +215,11 @@ def build_ird_analyze_block(
block["krt"] = _krt_at(db, lon, lat) block["krt"] = _krt_at(db, lon, lat)
# zone_regulation остаётся единственным живым geoportal-вызовом в hot-пути. # zone_regulation остаётся единственным живым geoportal-вызовом в hot-пути.
# Резолвим САМИ только если caller не передал готовый регламент (Fix 2).
# Выполняем в основном потоке (Session не потокобезопасна). # Выполняем в основном потоке (Session не потокобезопасна).
client = client or EKBGeoportalClient(timeout=_GEOPORTAL_TIMEOUT_S) if resolved_zone_regulation is _UNSET:
block["zone_regulation"] = _zone_regulation_at(db, client, lon, lat) client = client or EKBGeoportalClient(timeout=settings.geoportal_timeout_s)
block["zone_regulation"] = _zone_regulation_at(db, client, lon, lat)
return block return block

View file

@ -28,7 +28,9 @@ from __future__ import annotations
import json import json
import logging import logging
import re import re
import threading
import time import time
from collections import OrderedDict
from typing import Any from typing import Any
from shapely.geometry import shape from shapely.geometry import shape
@ -44,6 +46,61 @@ from app.services.scrapers.ekb_geoportal_client import (
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# ── Мемоизация coord → zone_index (WFS-lookup) (#1850) ───────────────────────
# Маппинг «координата → индекс терзоны» СТАТИЧЕН: границы зон не двигаются. Поэтому
# единственный живой WFS-вызов резолвера (``client.zone_index_at``) кэшируется
# процесс-локально. Важно: кэшируется ТОЛЬКО индекс зоны — СОДЕРЖИМОЕ регламента всё
# равно читается свежим из ``zone_regulation_cache`` на каждый вызов, поэтому корректность
# обновления кэша регламента сохраняется (refresh-воркер обновляет БД, не этот memo).
#
# Ключ — координата, округлённая до 5 знаков (≈1 м): безопасно мельче границ зон,
# но достаточно грубо чтобы соседние /analyze одного участка (один centroid) попадали
# в один ключ. Округлять КРУПНЕЕ нельзя — рискуем перепрыгнуть границу зоны.
#
# Bounded LRU (OrderedDict, FIFO-эвикция самого старого) — клиент не хешируемый, поэтому
# ``functools.lru_cache`` неприменим; держим собственный OrderedDict-LRU.
# THREAD-SAFETY (#1850): /analyze — sync def → исполняется в Starlette threadpool, т.е.
# несколько потоков мутируют этот module-global OrderedDict ОДНОВРЕМЕННО. Свой LRU (в
# отличие от functools.lru_cache с внутренним локом) НЕ атомарен: check-then-act между
# ``key in memo`` и ``move_to_end``/``[]`` против чужого ``popitem`` даёт KeyError. Поэтому
# мутации под Lock'ом; WFS-вызов — ВНЕ лока (иначе медленный геопортал сериализует все потоки).
_ZONE_INDEX_MEMO_MAXSIZE = 4096
_zone_index_memo: OrderedDict[tuple[float, float], str | None] = OrderedDict()
_zone_index_memo_lock = threading.Lock()
# Sentinel «координата вне терзоны» — отличает закэшированный негативный результат
# (None-зона, не пере-дёргаем WFS для участков вне ЕКБ) от «ключа нет в memo».
_NO_ZONE = "\x00__no_zone__\x00"
def _memoized_zone_index(client: EKBGeoportalClient, lon: float, lat: float) -> str | None:
"""Индекс терзоны в точке с процесс-локальной мемоизацией (#1850).
Кэширует И положительный (индекс зоны), И отрицательный (None участок вне терзоны/ЕКБ)
результат, чтобы повтор по тем же/близким координатам не делал WFS round-trip.
"""
key = (round(lon, 5), round(lat, 5))
with _zone_index_memo_lock:
if key in _zone_index_memo:
_zone_index_memo.move_to_end(key) # LRU: освежаем недавно использованный ключ
cached = _zone_index_memo[key]
return None if cached == _NO_ZONE else cached
# WFS-вызов ВНЕ лока: сетевой round-trip не должен сериализовать тёплые потоки.
# Дубль-вызов на одном холодном ключе из двух потоков безвреден (lookup идемпотентен).
zone_index = client.zone_index_at(lon, lat)
with _zone_index_memo_lock:
_zone_index_memo[key] = _NO_ZONE if zone_index is None else zone_index
_zone_index_memo.move_to_end(key)
if len(_zone_index_memo) > _ZONE_INDEX_MEMO_MAXSIZE:
_zone_index_memo.popitem(last=False) # эвикция самого старого
return zone_index
def _reset_zone_index_memo() -> None:
"""Очистить memo coord→zone_index (для тестов; prod — процесс-локальный кэш)."""
_zone_index_memo.clear()
# bbox агломерации ЕКБ (minlon, minlat, maxlon, maxlat) EPSG:4326 — границы для WFS-перечисления # bbox агломерации ЕКБ (minlon, minlat, maxlon, maxlat) EPSG:4326 — границы для WFS-перечисления
# терзон. Слой territorial_zone геопортала — ЕКБ-only, поэтому bbox это просто bounds-обёртка # терзон. Слой territorial_zone геопортала — ЕКБ-only, поэтому bbox это просто bounds-обёртка
# (а не фильтр города). Шире чем нужно — лучше захватить лишнее, дедуп по zone_index снимет повтор. # (а не фильтр города). Шире чем нужно — лучше захватить лишнее, дедуп по zone_index снимет повтор.
@ -263,7 +320,9 @@ def get_or_fetch_zone_regulation(
""" """
client = client or EKBGeoportalClient() client = client or EKBGeoportalClient()
try: try:
zone_index = client.zone_index_at(lon, lat) # Мемоизированный coord→zone_index: повтор (тот же centroid в рамках запроса или
# между запросами) не делает WFS round-trip (#1850).
zone_index = _memoized_zone_index(client, lon, lat)
except Exception as exc: except Exception as exc:
logger.warning("get_or_fetch_zone_regulation: zone_index_at failed: %s", exc) logger.warning("get_or_fetch_zone_regulation: zone_index_at failed: %s", exc)
return None return None
@ -279,6 +338,15 @@ def get_or_fetch_zone_regulation(
if reg is None or not reg.zone_index: if reg is None or not reg.zone_index:
return None return None
upsert_zone_regulation(db, reg, city=city) upsert_zone_regulation(db, reg, city=city)
# SMELL (#1850, item 4): db.commit() здесь коммитит ВНЕШНЮЮ /analyze-транзакцию
# mid-request (она в норме идёт на SAVEPOINT'ах). upsert_zone_regulation пишет внутри
# begin_nested() (SAVEPOINT), но релиз SAVEPOINT не персистит durable — нужен commit
# внешней tx. Изолировать апсёрт в собственную короткую транзакцию здесь нельзя без
# смены семантики резолвера (он получает SHARED Session от хэндлера, не владеет её
# границами); отдельная сессия/коннект рискует потерять только что записанный регламент.
# ОСТАВЛЕНО как есть: cache-miss редок (Fix 1 мемоизирует zone_index, кэш ЕКБ прогрет ~33
# зоны), поэтому фактический коммит почти не случается в hot-пути. Корректный фикс —
# отдельная write-сессия для апсёрта — отдельной задачей, чтобы не рисковать резолвером.
db.commit() db.commit()
return get_cached_zone_regulation(db, reg.zone_index, city=city) return get_cached_zone_regulation(db, reg.zone_index, city=city)
@ -412,6 +480,7 @@ def backfill_ekb_zone_regulations(
__all__ = [ __all__ = [
"_reset_zone_index_memo",
"backfill_ekb_zone_regulations", "backfill_ekb_zone_regulations",
"get_cached_zone_regulation", "get_cached_zone_regulation",
"get_or_fetch_zone_regulation", "get_or_fetch_zone_regulation",

View file

@ -562,3 +562,109 @@ def test_ppt_tep_empty_when_no_overlap(monkeypatch: Any) -> None:
assert "ppt_tep" in block assert "ppt_tep" in block
assert block["ppt_tep"] == [] assert block["ppt_tep"] == []
# ---------------------------------------------------------------------------
# Fix 2 (#1850): резолв регламента ОДИН раз — переиспользование в ИРД-блоке
# ---------------------------------------------------------------------------
def test_resolved_zone_regulation_skips_resolver(monkeypatch: Any) -> None:
"""resolved_zone_regulation=<dict> → _zone_regulation_at НЕ зовётся, значение взято as-is."""
_patch_overlay(monkeypatch, _OVERLAY)
resolver_calls: list[int] = []
def _spy(*a: Any, **kw: Any) -> Any:
resolver_calls.append(1)
return {"zone_index": "DIFFERENT"}
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", _spy)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(
object(),
"POINT(60 56)",
60.6,
56.8,
client=client,
resolved_zone_regulation={"zone_index": "Ц-1", "max_far": 4},
)
# Переданное значение использовано напрямую; резолвер ни разу не вызван.
assert block["zone_regulation"] == {"zone_index": "Ц-1", "max_far": 4}
assert resolver_calls == []
def test_resolved_zone_regulation_none_skips_resolver(monkeypatch: Any) -> None:
"""resolved_zone_regulation=None (валидное «зоны нет») → резолвер НЕ зовётся, итог None."""
_patch_overlay(monkeypatch, _OVERLAY)
resolver_calls: list[int] = []
def _spy(*a: Any, **kw: Any) -> Any:
resolver_calls.append(1)
return {"zone_index": "SHOULD_NOT_APPEAR"}
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", _spy)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(
object(),
"POINT(60 56)",
60.6,
56.8,
client=client,
resolved_zone_regulation=None,
)
assert block["zone_regulation"] is None
assert resolver_calls == [] # None — переданное значение, не повод резолвить
def test_unset_resolved_regulation_resolves_itself(monkeypatch: Any) -> None:
"""Параметр НЕ передан (дефолт _UNSET) → блок резолвит сам (path: zoning-flag off, ird on)."""
_patch_overlay(monkeypatch, _OVERLAY)
resolver_calls: list[int] = []
def _spy(db: Any, lon: float, lat: float, client: Any = None) -> dict[str, Any]:
resolver_calls.append(1)
return {"zone_index": "Ж-2"}
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", _spy)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
client = _FakeClient(fz=[])
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
assert block["zone_regulation"] == {"zone_index": "Ж-2"}
assert resolver_calls == [1] # сам резолвил, ровно один раз
# ---------------------------------------------------------------------------
# Fix 3 (#1850): единый timeout из settings.geoportal_timeout_s
# ---------------------------------------------------------------------------
def test_geoportal_client_uses_settings_timeout(monkeypatch: Any) -> None:
"""Когда клиент не передан, ИРД-блок конструирует EKBGeoportalClient с settings.timeout."""
_patch_overlay(monkeypatch, _OVERLAY)
monkeypatch.setattr(
ird_analyze, "get_or_fetch_zone_regulation", lambda db, lon, lat, client=None: None
)
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
monkeypatch.setattr(ird_analyze.settings, "geoportal_timeout_s", 7)
constructed: list[int] = []
class _SpyClient:
def __init__(self, timeout: int = 20) -> None:
constructed.append(timeout)
monkeypatch.setattr(ird_analyze, "EKBGeoportalClient", _SpyClient)
# client=None → блок сам создаёт EKBGeoportalClient(timeout=settings.geoportal_timeout_s).
ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8)
assert constructed == [7]

View file

@ -0,0 +1,170 @@
"""Тесты мемоизации coord→zone_index в get_or_fetch_zone_regulation (#1850 Fix 1).
Маппинг «координата индекс терзоны» статичен единственный живой WFS-вызов резолвера
(``client.zone_index_at``) кэшируется процесс-локально. Проверяем:
- повтор по тем же координатам zone_index_at вызван ОДИН раз (memo hit),
- негативное кэширование: координата без зоны (None) тоже кэшируется (вне ЕКБ не дёргаем WFS),
- округление до 5 знаков: близкие координаты (<1e-5) один ключ,
- _reset_zone_index_memo() очищает кэш между тестами.
Сеть/БД не дёргаются: фейковый клиент считает вызовы zone_index_at; кэш регламента
(get_cached_zone_regulation) замокан так, чтобы зона считалась уже закэшированной
(short-circuit до urbanCard).
"""
from __future__ import annotations
from typing import Any
import pytest
from app.services.site_finder import zone_regulation as zr
class _CountingClient:
"""Фейковый WFS-клиент: считает вызовы zone_index_at, отдаёт заданный индекс (или None)."""
def __init__(self, zone_index: str | None) -> None:
self._zone_index = zone_index
self.index_calls: list[tuple[float, float]] = []
self.regulation_calls: int = 0
def zone_index_at(self, lon: float, lat: float) -> str | None:
self.index_calls.append((lon, lat))
return self._zone_index
def zone_regulation_at(self, lon: float, lat: float) -> Any:
self.regulation_calls += 1
return None
@pytest.fixture(autouse=True)
def _reset_memo() -> Any:
"""Чистый memo до и после каждого теста (процесс-локальный кэш течёт между тестами)."""
zr._reset_zone_index_memo()
yield
zr._reset_zone_index_memo()
def test_zone_index_memoized_across_calls(monkeypatch: Any) -> None:
"""Два вызова с теми же координатами → zone_index_at вызван РОВНО один раз (memo hit)."""
# Зона "Ц-1" уже в кэше регламента → short-circuit до urbanCard, изолируем memo.
monkeypatch.setattr(
zr,
"get_cached_zone_regulation",
lambda db, zone_index, **kw: {"zone_index": zone_index},
)
client = _CountingClient("Ц-1")
r1 = zr.get_or_fetch_zone_regulation(object(), 60.6, 56.8, client=client)
r2 = zr.get_or_fetch_zone_regulation(object(), 60.6, 56.8, client=client)
assert r1 == {"zone_index": "Ц-1"}
assert r2 == {"zone_index": "Ц-1"}
assert len(client.index_calls) == 1 # второй вызов — memo hit, без WFS
def test_negative_result_is_memoized(monkeypatch: Any) -> None:
"""Координата без терзоны (None) кэшируется → второй вызов не дёргает WFS (вне ЕКБ)."""
# get_cached не должен влиять: zone_index=None → ветка кэша не выполняется,
# zone_regulation_at вернёт None → результат None.
monkeypatch.setattr(zr, "get_cached_zone_regulation", lambda *a, **kw: None)
client = _CountingClient(None)
r1 = zr.get_or_fetch_zone_regulation(object(), 1.0, 1.0, client=client)
r2 = zr.get_or_fetch_zone_regulation(object(), 1.0, 1.0, client=client)
assert r1 is None
assert r2 is None
assert len(client.index_calls) == 1 # негативный результат закэширован
def test_nearby_coords_share_memo_key(monkeypatch: Any) -> None:
"""Координаты, совпадающие после round(,5), делят ключ → один WFS-вызов."""
monkeypatch.setattr(
zr,
"get_cached_zone_regulation",
lambda db, zone_index, **kw: {"zone_index": zone_index},
)
client = _CountingClient("Ж-2")
# Различие в 6-м знаке → одинаковый ключ после round(lon, 5)/round(lat, 5).
zr.get_or_fetch_zone_regulation(object(), 60.600001, 56.800002, client=client)
zr.get_or_fetch_zone_regulation(object(), 60.600002, 56.800001, client=client)
assert len(client.index_calls) == 1
def test_distinct_coords_each_hit_wfs(monkeypatch: Any) -> None:
"""Координаты, различимые на уровне 5 знаков, бьют WFS отдельно (round не слишком грубый)."""
monkeypatch.setattr(
zr,
"get_cached_zone_regulation",
lambda db, zone_index, **kw: {"zone_index": zone_index},
)
client = _CountingClient("Ц-2")
zr.get_or_fetch_zone_regulation(object(), 60.60000, 56.80000, client=client)
zr.get_or_fetch_zone_regulation(object(), 60.60002, 56.80000, client=client) # ~1.2 м восточнее
assert len(client.index_calls) == 2
def test_reset_memo_clears_cache(monkeypatch: Any) -> None:
"""_reset_zone_index_memo() сбрасывает кэш → следующий вызов снова бьёт WFS."""
monkeypatch.setattr(
zr,
"get_cached_zone_regulation",
lambda db, zone_index, **kw: {"zone_index": zone_index},
)
client = _CountingClient("Ц-1")
zr.get_or_fetch_zone_regulation(object(), 60.6, 56.8, client=client)
assert len(client.index_calls) == 1
zr._reset_zone_index_memo()
zr.get_or_fetch_zone_regulation(object(), 60.6, 56.8, client=client)
assert len(client.index_calls) == 2 # после reset — снова WFS
def test_memo_evicts_oldest_over_maxsize(monkeypatch: Any) -> None:
"""LRU-эвикция: при превышении maxsize выбрасывается самый старый ключ."""
monkeypatch.setattr(zr, "_ZONE_INDEX_MEMO_MAXSIZE", 3)
monkeypatch.setattr(zr, "get_cached_zone_regulation", lambda *a, **kw: None)
client = _CountingClient(None)
# 4 различных ключа при maxsize=3 → первый (самый старый) эвиктится.
for i in range(4):
zr.get_or_fetch_zone_regulation(object(), 60.0 + i * 0.001, 56.0, client=client)
assert len(client.index_calls) == 4
# Повтор самого старого (60.000) → cache miss (эвиктнут) → ещё один WFS-вызов.
zr.get_or_fetch_zone_regulation(object(), 60.0, 56.0, client=client)
assert len(client.index_calls) == 5
def test_memo_concurrent_access_no_error(monkeypatch: Any) -> None:
"""#1850 thread-safety: N потоков сквозь _memoized_zone_index при tiny maxsize →
мутации под Lock'ом не дают KeyError/гонок (memo делится между threadpool-потоками)."""
import threading
monkeypatch.setattr(zr, "_ZONE_INDEX_MEMO_MAXSIZE", 2) # форсим частую эвикцию
client = _CountingClient("Ц-1")
errors: list[BaseException] = []
barrier = threading.Barrier(8)
def worker(n: int) -> None:
try:
barrier.wait() # стартуем одновременно → максимум контеншена
for i in range(200):
zr._memoized_zone_index(client, 60.0 + (i % 5) * 0.001, 56.0)
except BaseException as exc: # фиксируем любую гонку (KeyError/RuntimeError)
errors.append(exc)
threads = [threading.Thread(target=worker, args=(n,)) for n in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
assert not errors, f"concurrent memo access raised: {errors[:3]}"