gendesign/backend/tests/test_poi_loader.py

"""Unit tests для poi_loader — региональный bbox + адаптивное дробление Overpass-тайлов.

Mock-based / pure — БЕЗ живых походов в Overpass и без БД (правило: реальные запросы
к Overpass в тестах недопустимы). Покрывает:
- REGION_BBOX / DEFAULT_REGION — дефолт остаётся "ekb", не меняется молча.
- _bbox_tiles — для ЕКБ ровно один тайл, идентичный EKB_BBOX; для Москвы — несколько
  тайлов, покрывающих исходный bbox без дыр/нахлёста (по площади).
- _build_overpass_query — bbox теперь параметр, а не глобальная константа.
- _fetch_category — retry (было и раньше) + НОВОЕ: адаптивное дробление тайла на 4
  четверти при устойчивом провале (вместо тихой потери тайла), с остановкой на
  RECURSIVE_SPLIT_MAX_DEPTH (без бесконечной рекурсии).
"""

from __future__ import annotations

from types import SimpleNamespace

import pytest

from app.services.site_finder.poi_loader import (
    DEFAULT_REGION,
    EKB_BBOX,
    RECURSIVE_SPLIT_MAX_DEPTH,
    REGION_BBOX,
    _bbox_tiles,
    _build_overpass_query,
    _fetch_category,
    _split_bbox_quadrants,
)


class _FakeResponse:
    def __init__(self, ok: bool, elements: list[dict] | None = None) -> None:
        self._ok = ok
        self._elements = elements or []

    def raise_for_status(self) -> None:
        if not self._ok:
            raise RuntimeError("simulated Overpass failure")

    def json(self) -> dict:
        return {"elements": self._elements}


@pytest.fixture
def instant_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
    """Подменяет asyncio.sleep внутри poi_loader на no-op — тесты дробления тайлов иначе
    реально спали бы минуты (retry-пауза 3с + 1с между каждой из 4 четвертей на каждом
    уровне рекурсии)."""

    async def _instant(_seconds: float) -> None:
        return None

    monkeypatch.setattr("app.services.site_finder.poi_loader.asyncio.sleep", _instant)


# ── REGION_BBOX / дефолт ──────────────────────────────────────────────────────


def test_default_region_is_ekb_unchanged() -> None:
    assert DEFAULT_REGION == "ekb"
    assert REGION_BBOX["ekb"] == EKB_BBOX


def test_region_bbox_has_msk_product_core() -> None:
    assert "msk" in REGION_BBOX
    south, west, north, east = REGION_BBOX["msk"]
    assert south < north
    assert west < east


# ── _bbox_tiles ────────────────────────────────────────────────────────────────


def test_bbox_tiles_ekb_is_single_tile_identical_to_ekb_bbox() -> None:
    """Дефолтный регион не должен молча поменять поведение — один тайл, байт-в-байт EKB_BBOX."""
    tiles = _bbox_tiles(EKB_BBOX)
    assert tiles == [EKB_BBOX]


def test_bbox_tiles_msk_splits_into_multiple_tiles_without_gaps() -> None:
    bbox = REGION_BBOX["msk"]
    tiles = _bbox_tiles(bbox)
    assert len(tiles) > 1
    south, west, north, east = bbox
    total_area = (north - south) * (east - west)
    tiles_area = sum((t[2] - t[0]) * (t[3] - t[1]) for t in tiles)
    assert tiles_area == pytest.approx(total_area, rel=1e-9)


# ── _split_bbox_quadrants ────────────────────────────────────────────────────────


def test_split_bbox_quadrants_covers_original_area() -> None:
    bbox = (55.55, 37.30, 55.95, 37.90)
    quads = _split_bbox_quadrants(bbox)
    assert len(quads) == 4
    south, west, north, east = bbox
    total_area = (north - south) * (east - west)
    quads_area = sum((q[2] - q[0]) * (q[3] - q[1]) for q in quads)
    assert quads_area == pytest.approx(total_area, rel=1e-9)


# ── _build_overpass_query ─────────────────────────────────────────────────────


def test_build_overpass_query_uses_given_bbox_not_global_constant() -> None:
    q = _build_overpass_query((("amenity", "pharmacy"),), (1.0, 2.0, 3.0, 4.0))
    assert "(1.0,2.0,3.0,4.0)" in q
    assert '["amenity"="pharmacy"]' in q


# ── _fetch_category: retry (существующее поведение) ──────────────────────────────


async def test_fetch_category_retries_then_succeeds(instant_sleep: None) -> None:
    calls = {"n": 0}

    async def fake_post(_url: str, data: dict) -> _FakeResponse:
        calls["n"] += 1
        if calls["n"] == 1:
            return _FakeResponse(ok=False)
        return _FakeResponse(ok=True, elements=[{"type": "node", "id": 1, "lat": 1, "lon": 2}])

    client = SimpleNamespace(post=fake_post)
    result = await _fetch_category(client, (("amenity", "pharmacy"),), "pharmacy", (0, 0, 1, 1))
    assert calls["n"] == 2
    assert len(result) == 1
    assert result[0]["_gd_category"] == "pharmacy"


# ── _fetch_category: адаптивное дробление (НОВОЕ) ─────────────────────────────────


async def test_fetch_category_splits_into_quadrants_on_persistent_failure(
    instant_sleep: None,
) -> None:
    """Тайл, где оба attempt проваливаются, дробится на 4 четверти вместо потери данных."""
    calls = {"n": 0}

    async def fake_post(_url: str, data: dict) -> _FakeResponse:
        calls["n"] += 1
        query = data["data"]
        if "(0.0,0.0,1.0,1.0)" in query:  # верхнеуровневый тайл всегда 504
            return _FakeResponse(ok=False)
        return _FakeResponse(
            ok=True, elements=[{"type": "node", "id": calls["n"], "lat": 0.1, "lon": 0.1}]
        )

    client = SimpleNamespace(post=fake_post)
    result = await _fetch_category(
        client, (("amenity", "pharmacy"),), "pharmacy", (0.0, 0.0, 1.0, 1.0)
    )
    # верхний тайл: 2 неудачных attempt, затем 4 успешных запроса по четвертям
    assert calls["n"] == 2 + 4
    assert len(result) == 4


async def test_fetch_category_gives_up_at_max_depth_without_infinite_recursion(
    instant_sleep: None,
) -> None:
    """Тайл, падающий на ЛЮБОМ размере, останавливает дробление на RECURSIVE_SPLIT_MAX_DEPTH
    и возвращает пустой список — не зацикливается и не падает."""
    calls = {"n": 0}

    async def fake_post(_url: str, data: dict) -> _FakeResponse:
        calls["n"] += 1
        assert data  # параметр используется — сигнатура должна совпадать с client.post
        return _FakeResponse(ok=False)

    client = SimpleNamespace(post=fake_post)
    result = await _fetch_category(
        client, (("amenity", "pharmacy"),), "pharmacy", (0.0, 0.0, 1.0, 1.0)
    )
    assert result == []
    # sum_{d=0}^{max_depth} 4^d узлов, каждый по 2 attempt — рекурсия конечна
    expected_nodes = sum(4**d for d in range(RECURSIVE_SPLIT_MAX_DEPTH + 1))
    assert calls["n"] == expected_nodes * 2