enduro-trails/tests/api/test_gps_mvt_zoom_tiers.py

"""Unit-тесты zoom-tier в build_gps_mvt (ET-012, ADR-016).

Покрытие из 04-test-plan.yaml / 02-trz.md REQ-F-09:
  UT-Z5-01 — треки < 10 км отфильтровываются на z=5.
  UT-Z5-02 — limit=1500 на z=5.
  UT-Z6-01 — треки < 5 км отфильтровываются на z=6.
  UT-Z6-02 — limit=2000 на z=6.
  UT-Z7-01 — регрессия: min_length=2000, limit=3000 на z=7.
  UT-Z8-01 — регрессия: нет min_length, limit=8000 на z=8.
  UT-Z12-01 — регрессия: нет min_length, limit=25000 на z=12.

Все тесты используют mock-rows с фиксированной геометрией, помещённой
в тестовый тайл (см. ``_tile_for``). После build_gps_mvt MVT декодируется
mapbox_vector_tile.decode и подсчитывается число features в layer
``gps_tracks``.
"""
import json
import math

import mapbox_vector_tile
from shapely import wkb
from shapely.geometry import LineString

from src.api.gps_tracks.mvt import build_gps_mvt


# ─── Helpers ────────────────────────────────────────────────────────────────

def _tile_for(z: int, lon: float, lat: float) -> tuple[int, int]:
    """Возвращает (x, y) Web-Mercator-тайла для координат на зуме z.

    Использует ту же формулу, что и обратное преобразование в mvt._tile_to_bbox,
    но в прямую сторону: lon/lat → x/y.
    """
    n = 2 ** z
    x = int((lon + 180.0) / 360.0 * n)
    y_rad = math.radians(lat)
    y = int((1 - math.asinh(math.tan(y_rad)) / math.pi) / 2 * n)
    # На границе мира clamp
    return max(0, min(n - 1, x)), max(0, min(n - 1, y))


def _make_row(
    track_id: int,
    length_m: float,
    *,
    lon_center: float = 37.0,
    lat_center: float = 55.0,
    span: float = 0.005,
    activity_type: str = "enduro",
    source_id: str = "osm",
):
    """Создаёт mock sqlite3.Row-словарь с маленьким треком вокруг центра.

    span задан в градусах. По умолчанию ~500 м — линия безопасно лежит
    внутри тайлов z>=5 над выбранной точкой.
    """
    coords = [
        (lon_center - span, lat_center - span / 2),
        (lon_center, lat_center),
        (lon_center + span, lat_center + span / 2),
    ]
    geom = wkb.dumps(LineString(coords))

    class _Row(dict):
        def __getitem__(self, key):
            return super().__getitem__(key)

    return _Row({
        "id": track_id,
        "activity_type": activity_type,
        "sources_json": json.dumps([source_id]),
        "external_urls_json": json.dumps([]),
        "length_m": length_m,
        "name": f"Track {track_id}",
        "geom": geom,
    })


def _decode_features(mvt_bytes: bytes) -> list:
    """Декодирует MVT и возвращает список features в layer gps_tracks.

    Если тайл пуст (b"") — возвращает [].
    """
    if not mvt_bytes:
        return []
    decoded = mapbox_vector_tile.decode(mvt_bytes)
    layer = decoded.get("gps_tracks")
    if not layer:
        return []
    return layer.get("features", [])


# ─── UT-Z5-01: треки < 10 км отфильтровываются ──────────────────────────────

def test_ut_z5_01_short_tracks_filtered():
    """REQ-F-09 / UT-Z5-01: на z=5 проходят только треки длиной >= 10 км.

    Из 10 треков [500..120000] должны попасть в MVT ровно 6
    (длины 12000, 15000, 25000, 50000, 80000, 120000).
    """
    lengths = [500, 2000, 3000, 8000, 12000, 15000, 25000, 50000, 80000, 120000]
    lon, lat = 37.0, 55.0
    x, y = _tile_for(5, lon, lat)

    rows = [
        _make_row(i, length_m=length, lon_center=lon, lat_center=lat)
        for i, length in enumerate(lengths, start=1)
    ]

    mvt = build_gps_mvt(rows, z=5, x=x, y=y)
    features = _decode_features(mvt)

    # Ожидаем 6 features — треки длиной >= 10000 м.
    assert len(features) == 6
    # Все попавшие — с length_km >= 10.0
    for feat in features:
        assert feat["properties"]["length_km"] >= 10.0


# ─── UT-Z5-02: limit=1500 на z=5 ────────────────────────────────────────────

def test_ut_z5_02_limit_1500():
    """REQ-F-09 / UT-Z5-02: на z=5 cap=1500 при большом числе длинных треков.

    Все 2000 треков проходят min_length=10000 (15 км), но в MVT уходит
    только первые 1500 (build_gps_mvt брейкает цикл по len(features) >= limit).
    """
    lon, lat = 37.0, 55.0
    x, y = _tile_for(5, lon, lat)

    rows = [
        _make_row(i, length_m=15000, lon_center=lon, lat_center=lat)
        for i in range(1, 2001)
    ]

    mvt = build_gps_mvt(rows, z=5, x=x, y=y)
    features = _decode_features(mvt)

    assert len(features) == 1500


# ─── UT-Z6-01: треки < 5 км отфильтровываются ───────────────────────────────

def test_ut_z6_01_short_tracks_filtered():
    """REQ-F-09 / UT-Z6-01: на z=6 проходят только треки длиной >= 5 км.

    Из [1000, 3000, 5000, 7000, 10000] должны попасть 3 (5000, 7000, 10000).
    """
    lengths = [1000, 3000, 5000, 7000, 10000]
    lon, lat = 37.0, 55.0
    x, y = _tile_for(6, lon, lat)

    rows = [
        _make_row(i, length_m=length, lon_center=lon, lat_center=lat)
        for i, length in enumerate(lengths, start=1)
    ]

    mvt = build_gps_mvt(rows, z=6, x=x, y=y)
    features = _decode_features(mvt)

    assert len(features) == 3
    for feat in features:
        assert feat["properties"]["length_km"] >= 5.0


# ─── UT-Z6-02: limit=2000 на z=6 ────────────────────────────────────────────

def test_ut_z6_02_limit_2000():
    """REQ-F-09 / UT-Z6-02: на z=6 cap=2000 при 2500 треках >= 5 км."""
    lon, lat = 37.0, 55.0
    x, y = _tile_for(6, lon, lat)

    rows = [
        _make_row(i, length_m=6000, lon_center=lon, lat_center=lat)
        for i in range(1, 2501)
    ]

    mvt = build_gps_mvt(rows, z=6, x=x, y=y)
    features = _decode_features(mvt)

    assert len(features) == 2000


# ─── UT-Z7-01: регрессия — min_length=2000, limit=3000 ──────────────────────

def test_ut_z7_01_regression():
    """REQ-F-09 / UT-Z7-01: поведение z=7 не изменилось (min_length=2000)."""
    lengths = [1000, 2000, 3000, 5000]
    lon, lat = 37.0, 55.0
    x, y = _tile_for(7, lon, lat)

    rows = [
        _make_row(i, length_m=length, lon_center=lon, lat_center=lat)
        for i, length in enumerate(lengths, start=1)
    ]

    mvt = build_gps_mvt(rows, z=7, x=x, y=y)
    features = _decode_features(mvt)

    # 1000 < 2000 → отфильтрован; 2000, 3000, 5000 — попадают.
    assert len(features) == 3
    for feat in features:
        assert feat["properties"]["length_km"] >= 2.0


def test_ut_z7_01_limit_3000():
    """REQ-F-09 / UT-Z7-01 (доп.): cap=3000 на z=7."""
    lon, lat = 37.0, 55.0
    x, y = _tile_for(7, lon, lat)

    rows = [
        _make_row(i, length_m=4000, lon_center=lon, lat_center=lat)
        for i in range(1, 3500)
    ]

    mvt = build_gps_mvt(rows, z=7, x=x, y=y)
    features = _decode_features(mvt)

    assert len(features) == 3000


# ─── UT-Z8-01: регрессия — нет min_length, limit=8000 ───────────────────────

def test_ut_z8_01_regression_no_min_length():
    """REQ-F-09 / UT-Z8-01: на z=8 любые треки проходят."""
    lengths = [500, 1000, 2000, 5000]
    lon, lat = 37.0, 55.0
    x, y = _tile_for(8, lon, lat)

    rows = [
        _make_row(i, length_m=length, lon_center=lon, lat_center=lat)
        for i, length in enumerate(lengths, start=1)
    ]

    mvt = build_gps_mvt(rows, z=8, x=x, y=y)
    features = _decode_features(mvt)

    assert len(features) == 4


# ─── UT-Z12-01: регрессия — limit=25000, без min_length ─────────────────────

def test_ut_z12_01_regression_no_filtering():
    """REQ-F-09 / UT-Z12-01: на z=12 любая длина проходит, малое число фич."""
    lon, lat = 37.0, 55.0
    x, y = _tile_for(12, lon, lat)

    rows = [
        _make_row(i, length_m=100 * i, lon_center=lon, lat_center=lat)
        for i in range(1, 101)
    ]

    mvt = build_gps_mvt(rows, z=12, x=x, y=y)
    features = _decode_features(mvt)

    assert len(features) == 100