"""Tests de la pyramide de tuiles XYZ (schéma OpenStreetMap)."""

import math
from pathlib import Path


# ---------------------------------------------------------------------------
# Fixtures : dalles factices aux conventions du pipeline
# ---------------------------------------------------------------------------

def _basename(col, row):
    return f"LHD_FXX_{col:04d}_{row:04d}_PTS_LAMB93_IGN69"


def _make_dalle(output_dir, col, row, viz_keys, color=(200, 30, 30), px=64,
                thumbs=True):
    """Crée une dalle (image + vignettes) comme le ferait le pipeline."""
    from PIL import Image
    base = _basename(col, row)
    vis = Path(output_dir) / "visualisations" / base
    vis.mkdir(parents=True, exist_ok=True)
    thumb_dir = Path(output_dir) / "index_thumbs"
    thumb_dir.mkdir(parents=True, exist_ok=True)
    for key in viz_keys:
        Image.new("RGB", (px, px), color).save(
            str(vis / f"{base}_{key}.webp"), format="WEBP", lossless=True)
        if thumbs:
            Image.new("RGB", (64, 64), color).save(
                str(thumb_dir / f"{base}_{key}.jpg"), format="JPEG", quality=90)
            Image.new("RGB", (64, 64), color).save(
                str(thumb_dir / f"{base}_{key}_mid.jpg"), format="JPEG", quality=90)
    return base


def _tile_of_cell(col, row, z):
    """Indices (x, y) de la tuile du niveau z contenant le centre d'une dalle."""
    from lidar_pipeline.tiles import _transformer
    lon, lat = _transformer("EPSG:2154", "EPSG:4326").transform(
        col * 1000 + 500, (row - 1) * 1000 + 500)
    n = 2 ** z
    x = int((lon + 180.0) / 360.0 * n)
    rad = math.radians(lat)
    y = int((1.0 - math.log(math.tan(rad) + 1 / math.cos(rad)) / math.pi) / 2.0 * n)
    return x, y


# ---------------------------------------------------------------------------
# Géométrie de la grille
# ---------------------------------------------------------------------------

def test_tile_bounds_3857_known_values():
    """z0 = le monde entier ; z1/x1/y0 = quadrant nord-est."""
    from lidar_pipeline.tiles import ORIGIN, tile_bounds_3857
    w, s, e, n = tile_bounds_3857(0, 0, 0)
    assert (round(w), round(s), round(e), round(n)) == (
        round(-ORIGIN), round(-ORIGIN), round(ORIGIN), round(ORIGIN))
    w, s, e, n = tile_bounds_3857(1, 1, 0)
    assert abs(w) < 1e-6 and abs(s) < 1e-6
    assert abs(e - ORIGIN) < 1e-6 and abs(n - ORIGIN) < 1e-6


def test_tile_latitude_and_resolution():
    """Latitude du centre et résolution terrain (0,2 m/px ≈ z19 en France)."""
    from lidar_pipeline.tiles import (TILE_MAX_NATIVE_Z, target_resolution,
                                      tile_latitude)
    assert abs(tile_latitude(0, 0)) < 1e-9
    assert abs(tile_latitude(1, 0) - 66.51) < 0.05
    # Tuile du niveau natif à la latitude de la France métropolitaine
    z = TILE_MAX_NATIVE_Z
    y = int((1.0 - math.log(math.tan(math.radians(47)) + 1 / math.cos(math.radians(47)))
             / math.pi) / 2.0 * 2 ** z)
    res = target_resolution(z, y)
    assert 0.15 < res < 0.25, res
    # @2x : deux fois plus fin pour le même (z, x, y)
    assert abs(target_resolution(z, y, scale=2) - res / 2) < 1e-9


def test_tile_bounds_l93_covers_cell():
    """L'emprise L93 d'une tuile contient bien la dalle qu'elle recouvre."""
    from lidar_pipeline.tiles import tile_bounds_l93
    col, row, z = 1054, 6882, 14
    x, y = _tile_of_cell(col, row, z)
    min_x, min_y, max_x, max_y = tile_bounds_l93(z, x, y)
    assert min_x < col * 1000 + 500 < max_x
    assert min_y < (row - 1) * 1000 + 500 < max_y


def test_perspective_coeffs_identity_and_scale():
    """Identité → coefficients neutres ; homothétie → facteur exact."""
    from lidar_pipeline.tiles import perspective_coeffs
    quad = [(0, 0), (256, 0), (256, 256), (0, 256)]
    c = perspective_coeffs(quad, quad)
    assert [round(v, 9) for v in c] == [1, 0, 0, 0, 1, 0, 0, 0]
    # Sortie deux fois plus grande que la source : Pillow échantillonne à x/2
    c = perspective_coeffs([(0, 0), (512, 0), (512, 512), (0, 512)], quad)
    assert abs(c[0] - 0.5) < 1e-9 and abs(c[4] - 0.5) < 1e-9


def test_perspective_coeffs_degenerate():
    """Quadrilatère dégénéré (dalle réduite à un point) → None, pas d'exception."""
    from lidar_pipeline.tiles import perspective_coeffs
    flat = [(0, 0), (0, 0), (0, 0), (0, 0)]
    assert perspective_coeffs(flat, [(0, 0), (1, 0), (1, 1), (0, 1)]) is None


def test_zoom_supported_bounds():
    """Plage de zooms servie ; @2x s'arrête un cran plus tôt (512 px)."""
    from lidar_pipeline.tiles import (TILE_MAX_NATIVE_Z, TILE_MIN_Z,
                                      zoom_supported)
    assert not zoom_supported(TILE_MIN_Z - 1)
    assert zoom_supported(TILE_MIN_Z)
    assert zoom_supported(TILE_MAX_NATIVE_Z)
    assert not zoom_supported(TILE_MAX_NATIVE_Z + 1)
    assert not zoom_supported(TILE_MAX_NATIVE_Z, scale=2)
    assert zoom_supported(TILE_MAX_NATIVE_Z - 1, scale=2)


# ---------------------------------------------------------------------------
# Index des sources
# ---------------------------------------------------------------------------

def test_source_index_and_layers(tmp_path, monkeypatch):
    """L'index liste les couches présentes et leurs paliers (grossier → fin)."""
    from lidar_pipeline import index, tiles
    monkeypatch.setattr(index, "PANEL_VIZ", None)
    _make_dalle(tmp_path, 1054, 6882, ["aspect", "slope"])
    layers = tiles.source_index(tmp_path, force=True)
    assert set(layers) == {"aspect", "slope"}
    tiers = layers["aspect"][(1054, 6882)]
    # vignette (3,9 m/px) → intermédiaire (1,56) → dalle (0,5)
    assert [round(t[0].res, 2) for t in tiers] == [3.91, 1.56, 0.5]
    assert tiles.available_layers(tmp_path) == ["slope", "aspect"] or \
        set(tiles.available_layers(tmp_path)) == {"slope", "aspect"}


def test_available_layers_follow_panel(tmp_path, monkeypatch):
    """Couches sur disque hors PANEL_VIZ : ni affichées ni servies."""
    from lidar_pipeline import index, tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect", "relief_oriente"])
    tiles.source_index(tmp_path, force=True)
    monkeypatch.setattr(index, "PANEL_VIZ", ("relief_oriente",))
    assert tiles.available_layers(tmp_path) == ["relief_oriente"]


def test_grid_bounds(tmp_path):
    """Emprise L93 et WGS84 de la grille disponible."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    _make_dalle(tmp_path, 1055, 6883, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    assert tiles.grid_bounds_l93(tmp_path) == (1054000.0, 6881000.0,
                                               1056000.0, 6883000.0)
    w, s, e, n = tiles.grid_bounds_wgs84(tmp_path)
    assert 7.0 < w < 8.5 and 47.5 < s < 49.5 and e > w and n > s


def test_pick_tier_by_resolution(tmp_path):
    """Palier retenu : le plus grossier dont la résolution suffit à la tuile."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiers = tiles.source_index(tmp_path, force=True)["aspect"][(1054, 6882)]
    assert tiles._pick_tier(tiers, 10.0)[0].res == tiers[0][0].res      # vignette
    assert tiles._pick_tier(tiers, 2.0)[0].res == tiers[1][0].res       # intermédiaire
    assert tiles._pick_tier(tiers, 0.2)[0].res == tiers[-1][0].res      # dalle
    # Cible plus fine que tout ce qui existe : on garde le palier le plus fin
    assert tiles._pick_tier(tiers, 0.01)[0].res == tiers[-1][0].res


def test_sources_in_bbox_picks_tier(tmp_path):
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"], px=64)
    tiles.source_index(tmp_path, force=True)
    bbox = (1054000.0, 6881000.0, 1055000.0, 6882000.0)
    fine = tiles.sources_in_bbox(tmp_path, "aspect", bbox, 0.2)
    coarse = tiles.sources_in_bbox(tmp_path, "aspect", bbox, 50.0)
    assert fine and coarse and fine[0][0] == (1054, 6882)
    assert min(s.res for _c, s in fine) <= min(s.res for _c, s in coarse)
    assert tiles.sources_in_bbox(tmp_path, "aspect", (0, 0, 10, 10), 0.2) == []
    img = tiles.load_source(fine[0][1])
    assert img is not None and img.mode in ("RGB", "RGBA")


def test_subtiles_preferred_over_full_dalle(tmp_path):
    """Les quadrants index_subtiles servent de palier fin (4× moins à décoder)."""
    import pytest
    from PIL import Image
    from lidar_pipeline import tiles
    base = _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    sub = tmp_path / "index_subtiles"
    sub.mkdir(parents=True, exist_ok=True)
    try:
        for i in range(2):
            for j in range(2):
                Image.new("RGB", (32, 32), (10, 10, 10)).save(
                    str(sub / f"{base}_aspect_{i}_{j}.avif"), format="AVIF")
    except Exception:
        pytest.skip("encodeur AVIF indisponible")
    tiers = tiles.source_index(tmp_path, force=True)["aspect"][(1054, 6882)]
    quads = next(t for t in tiers if len(t) == 4)
    # À résolution égale, les quadrants passent AVANT la dalle entière
    assert tiers.index(quads) < tiers.index(next(t for t in tiers if len(t) == 1
                                                 and t[0].path.suffix == ".webp"))
    assert len(quads) == 4
    # Emprises des quadrants : quatre demi-kilomètres jointifs
    assert {q.bounds for q in quads} == {
        (1054000.0, 6881000.0, 1054500.0, 6881500.0),
        (1054500.0, 6881000.0, 1055000.0, 6881500.0),
        (1054000.0, 6881500.0, 1054500.0, 6882000.0),
        (1054500.0, 6881500.0, 1055000.0, 6882000.0)}


# ---------------------------------------------------------------------------
# Rendu et cache
# ---------------------------------------------------------------------------

def test_render_tile_paints_cell(tmp_path):
    """Une tuile au-dessus de la dalle est peinte ; ailleurs elle est vide."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"], color=(200, 30, 30))
    tiles.source_index(tmp_path, force=True)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    img = tiles.render_tile(tmp_path, "aspect", z, x, y)
    assert img is not None and img.size == (256, 256)
    r, g, b, a = img.getpixel((128, 128))
    assert a == 255 and r > 150 and g < 90 and b < 90
    # Tuile lointaine (autre continent) : aucune source
    assert tiles.render_tile(tmp_path, "aspect", z, 1, 1) is None


def test_render_tile_scale2(tmp_path):
    """`scale=2` rend la même emprise en 512 px (convention @2x)."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    img = tiles.render_tile(tmp_path, "aspect", z, x, y, scale=2)
    assert img is not None and img.size == (512, 512)


def test_tile_edges_transparent_outside_data(tmp_path):
    """Hors emprise des dalles, la tuile reste transparente (superposable)."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    # Zoom où une tuile est bien plus grande que la dalle : les bords sont vides
    z = 11
    x, y = _tile_of_cell(1054, 6882, z)
    img = tiles.render_tile(tmp_path, "aspect", z, x, y)
    assert img is not None
    assert img.getpixel((0, 0))[3] == 0
    assert img.getpixel((255, 255))[3] == 0


def test_get_tile_cache_and_staleness(tmp_path, monkeypatch):
    """Le cache disque est réutilisé, puis invalidé par une dalle régénérée."""
    from lidar_pipeline import tiles as _t
    monkeypatch.setattr(_t, "TILE_EVEN_LEVELS", False)   # règle historique : tout est stocké
    monkeypatch.setattr(_t, "TILE_CACHE_MAX_Z", 99)
    import os
    import time
    from lidar_pipeline import tiles
    base = _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    data = tiles.get_tile(tmp_path, "aspect", z, x, y)
    assert data and data[:8] == b"\x89PNG\r\n\x1a\n"
    cache = tiles.tile_cache_path(tmp_path, "aspect", z, x, y)
    assert cache.is_file()
    first = cache.stat().st_mtime

    # Sans changement : la tuile en cache est resservie telle quelle
    time.sleep(0.02)
    assert tiles.get_tile(tmp_path, "aspect", z, x, y) == data
    assert cache.stat().st_mtime == first

    # Dalle régénérée (mtime plus récente) : la tuile est recalculée
    src = tmp_path / "visualisations" / base / f"{base}_aspect.webp"
    newer = first + 10
    os.utime(src, (newer, newer))
    for f in (tmp_path / "index_thumbs").iterdir():
        os.utime(f, (newer, newer))
    tiles.source_index(tmp_path, force=True)
    tiles.get_tile(tmp_path, "aspect", z, x, y)
    assert cache.stat().st_mtime > first


def test_get_tile_empty_marker(tmp_path, monkeypatch):
    """Tuile sans donnée : None + marqueur .empty mémorisé."""
    from lidar_pipeline import tiles as _t
    monkeypatch.setattr(_t, "TILE_EVEN_LEVELS", False)   # règle historique : tout est stocké
    monkeypatch.setattr(_t, "TILE_CACHE_MAX_Z", 99)
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    assert tiles.get_tile(tmp_path, "aspect", 15, 1, 1) is None
    assert tiles.empty_marker_exists(tmp_path, "aspect", 15, 1, 1)


def test_get_tile_webp_scale2(tmp_path):
    """Palier @2x en WebP : 512 px, chemin de cache distinct du 256 px."""
    from PIL import Image
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    data = tiles.get_tile(tmp_path, "aspect", z, x, y, scale=2, fmt="webp")
    assert data and data[:4] == b"RIFF"
    path = tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "webp")
    assert path.name.endswith("@2x.webp") and path.is_file()
    import io
    assert Image.open(io.BytesIO(data)).size == (512, 512)


def test_transparent_tile_is_fully_transparent():
    """La tuile de repli (zone vide) est entièrement transparente."""
    import io
    from PIL import Image
    from lidar_pipeline import tiles
    img = Image.open(io.BytesIO(tiles.transparent_tile()))
    assert img.size == (256, 256)
    assert img.convert("RGBA").getextrema()[3] == (0, 0)


def test_tiles_in_bounds_and_warm(tmp_path):
    """Pré-chauffage : les tuiles de l'emprise sont calculées et mises en cache."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    bounds = tiles.grid_bounds_wgs84(tmp_path)
    assert len(tiles.tiles_in_bounds(bounds, 14)) >= 1
    report = tiles.warm(tmp_path, ["aspect"], 12, 13)
    assert report["rendues"] >= 1 and report["limite"] is False
    assert any((tmp_path / tiles.TILE_DIRNAME / "aspect").rglob("*.png"))


def test_tiles_stamp_follows_sources(tmp_path):
    """La version globale suit la mtime la plus récente des dalles."""
    import os
    from lidar_pipeline import tiles
    base = _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    first = tiles.tiles_stamp(tmp_path)
    src = tmp_path / "visualisations" / base / f"{base}_aspect.webp"
    os.utime(src, (first / 1000 + 60, first / 1000 + 60))
    tiles.source_index(tmp_path, force=True)
    assert tiles.tiles_stamp(tmp_path) > first


def test_source_cache_respects_memory_budget(tmp_path, monkeypatch):
    """Le cache d'images sources évince selon un budget en octets, pas un compte.

    Une dalle 5000² pèse ~75 Mo décodée : un cache « N entrées » ferait
    déborder la mémoire d'une petite machine (Raspberry Pi).
    """
    from PIL import Image
    from lidar_pipeline import tiles
    tiles.clear_source_cache()
    # Budget volontairement minuscule : une seule image tient à la fois
    monkeypatch.setattr(tiles, "SOURCE_CACHE_BYTES", 40 * 40 * 3 * 2 - 1)
    paths = []
    for i in range(3):
        f = tmp_path / f"src{i}.png"
        Image.new("RGB", (40, 40), (i * 40, 0, 0)).save(str(f))
        paths.append(f)
    for f in paths:
        tiles._open_source(str(f), f.stat().st_mtime)
    assert len(tiles._source_cache) == 1
    # La dernière source utilisée est celle qui reste
    assert str(paths[-1]) == list(tiles._source_cache)[0][0]
    tiles.clear_source_cache()
    assert tiles._source_cache == {}


def test_source_cache_holds_plain_decoded_images(tmp_path):
    """Le cache ne retient que les pixels : ni fichier ouvert ni décodeur.

    Une image AVIF ouverte garde son décodeur (tampons libavif/dav1d) :
    ~43 Mo retenus par quadrant 2500² au lieu de 25 — le conteneur du Pi
    (1 Go) était tué par l'OOM killer en navigation à fort zoom. Le budget
    compte 4 octets/pixel : PIL stocke le RGB sur 32 bits.
    """
    from PIL import Image
    from lidar_pipeline import tiles
    tiles.clear_source_cache()
    f = tmp_path / "src.png"
    Image.new("RGB", (40, 30), (10, 20, 30)).save(str(f))
    img = tiles._open_source(str(f), f.stat().st_mtime)
    assert type(img) is Image.Image
    assert getattr(img, "fp", None) is None
    assert img.info["_bytes"] == 40 * 30 * 4
    assert img.getpixel((0, 0))[:3] == (10, 20, 30)
    tiles.clear_source_cache()


def test_source_cache_keyed_by_mtime(tmp_path):
    """Une source réécrite n'est pas resservie depuis le cache."""
    import os
    from PIL import Image
    from lidar_pipeline import tiles
    tiles.clear_source_cache()
    f = tmp_path / "src.png"
    Image.new("RGB", (8, 8), (10, 10, 10)).save(str(f))
    first = tiles._open_source(str(f), f.stat().st_mtime)
    assert first.getpixel((0, 0))[:3] == (10, 10, 10)
    Image.new("RGB", (8, 8), (200, 200, 200)).save(str(f))
    os.utime(f, (f.stat().st_mtime + 5, f.stat().st_mtime + 5))
    second = tiles._open_source(str(f), f.stat().st_mtime)
    assert second.getpixel((0, 0))[:3] == (200, 200, 200)
    tiles.clear_source_cache()


def test_png_palette_option_shrinks_tiles(tmp_path, monkeypatch):
    """`LIDAR_TILE_PNG_PALETTE=1` allège le PNG canonique (palette + alpha).

    Mesuré sur un rendu réaliste (rampe de couleur bruitée) : une dalle unie
    compresse déjà mieux en RGBA qu'en palette, elle ne prouverait rien.
    """
    import io
    import random
    from PIL import Image
    from lidar_pipeline import tiles

    base = _basename(1054, 6882)
    vis = tmp_path / "visualisations" / base
    vis.mkdir(parents=True, exist_ok=True)
    rng = random.Random(7)
    img = Image.new("RGB", (256, 256))
    px = img.load()
    for j in range(256):
        for i in range(256):
            px[i, j] = (min(255, i + rng.randint(0, 12)),
                        min(255, j + rng.randint(0, 12)),
                        rng.randint(40, 90))
    img.save(str(vis / f"{base}_aspect.webp"), format="WEBP", lossless=True)
    tiles.source_index(tmp_path, force=True)

    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    rendered = tiles.render_tile(tmp_path, "aspect", z, x, y)
    lossless = tiles._encode(rendered, "png")
    monkeypatch.setattr(tiles, "PNG_PALETTE", True)
    palette = tiles._encode(rendered, "png")
    assert len(palette) < len(lossless)
    # Le PNG palettisé reste un PNG lisible, à la bonne taille, avec alpha
    out = Image.open(io.BytesIO(palette))
    assert out.size == (256, 256)
    assert out.convert("RGBA").getextrema()[3][1] == 255


# ---------------------------------------------------------------------------
# Serveur de dalles amont (LIDAR_SOURCE_URL)
# ---------------------------------------------------------------------------

def _remote_payload():
    """Charge utile /api/tiles telle que la sert mapserve (serveur de dalles)."""
    base = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_r0p2"
    tiles = []
    for i in range(2):
        for j in range(2):
            tiles.append({
                "col": 1054, "row": 6882, "resolution": 0.2,
                "dir_name": base, "sub_i": i, "sub_j": j, "sub_k": 2,
                "viz": {"aspect": {
                    "thumb": f"index_subtiles/{base}_aspect_{i}_{j}_thumb160.webp?v=1700000000000",
                    "mid": f"index_subtiles/{base}_aspect_{i}_{j}_mid.webp?v=1700000000000",
                    "full": f"index_subtiles/{base}_aspect_{i}_{j}.avif?v=1700000000000",
                }},
            })
    return {"tiles": tiles, "viz_meta": {"aspect": {"label": "Aspect"}}}


def test_remote_index_lists_tiles_without_local_data(tmp_path, monkeypatch):
    """Sans aucune donnée locale, l'index vient de l'amont (quadrants placés)."""
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
    monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: _remote_payload())
    layers = tiles.source_index(tmp_path, force=True)
    assert set(layers) == {"aspect"}
    tiers = layers["aspect"][(1054, 6882)]
    # Trois paliers (vignette, intermédiaire, quadrant) × 4 quadrants chacun
    assert [len(t) for t in tiers] == [4, 4, 4]
    assert [round(t[0].res, 2) for t in tiers] == [3.12, 0.78, 0.2]
    fine = tiers[-1][0]
    assert fine.url.startswith("http://amont:8973/index_subtiles/")
    assert fine.path == tmp_path / fine.url.split("8973/")[1].split("?")[0]
    # La date de référence vient du ?v= annoncé, sans rien télécharger
    assert fine.mtime() == 1700000000.0
    assert tiles.grid_bounds_l93(tmp_path) == (1054000.0, 6881000.0,
                                               1055000.0, 6882000.0)


def test_remote_source_downloaded_on_demand(tmp_path, monkeypatch):
    """Une source distante n'est rapatriée qu'au premier rendu qui en a besoin."""
    from PIL import Image
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
    monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: _remote_payload())
    tiles.clear_source_cache()

    import io
    buf = io.BytesIO()
    Image.new("RGB", (64, 64), (12, 200, 90)).save(buf, format="WEBP", lossless=True)
    body = buf.getvalue()
    fetched = []

    def fake_fetch(url, dest):
        fetched.append(url)
        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_bytes(body)
        return True

    monkeypatch.setattr(tiles, "_fetch_source", fake_fetch)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    img = tiles.render_tile(tmp_path, "aspect", z, x, y)
    assert img is not None
    assert fetched, "aucune source rapatriée"
    r, g, b, a = img.getpixel((128, 128))
    assert a == 255 and g > 150 and r < 80
    # Les fichiers rapatriés atterrissent dans le cache local, au même chemin
    assert list(tmp_path.rglob("*.webp")) or list(tmp_path.rglob("*.avif"))
    # Second rendu : plus aucun téléchargement (cache local)
    before = len(fetched)
    tiles.render_tile(tmp_path, "aspect", z, x, y)
    assert len(fetched) == before
    tiles.clear_source_cache()


def test_remote_index_failure_keeps_local(tmp_path, monkeypatch):
    """Amont injoignable : l'index local reste servi, sans exception."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["slope"])
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
    monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: None)
    layers = tiles.source_index(tmp_path, force=True)
    assert set(layers) == {"slope"}


def test_cached_tile_states(tmp_path):
    """`cached_tile` : lecture cache seule — fresh / pending / empty, sans rendu."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["slope"])
    x, y = _tile_of_cell(1054, 6882, 14)
    data, state = tiles.cached_tile(tmp_path, "slope", 14, x, y)
    assert state == "pending" and data is None
    assert not tiles.tile_cache_path(tmp_path, "slope", 14, x, y).exists()
    tiles.get_tile(tmp_path, "slope", 14, x, y)  # rendu (maintenance, warm…)
    data, state = tiles.cached_tile(tmp_path, "slope", 14, x, y)
    assert state == "fresh" and data
    # Hors données : état empty + marqueur posé, toujours sans rendu
    data, state = tiles.cached_tile(tmp_path, "slope", 14, 0, 0)
    assert state == "empty" and data is None
    assert tiles.empty_marker_exists(tmp_path, "slope", 14, 0, 0)


def test_storage_policy_even_levels_up_to_max(monkeypatch):
    """Stockage réduit : niveaux standard pairs ≤ LIDAR_TILE_CACHE_MAX_Z (une
    tuile @2x de niveau z vaut une 256 px de z+1)."""
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", True)
    monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 16)
    assert tiles.zoom_cached(16, 1) and not tiles.zoom_cached(15, 1)
    assert tiles.zoom_cached(15, 2) and not tiles.zoom_cached(16, 2)      # @2x z15 = z16 standard
    assert not tiles.zoom_cached(18, 1) and not tiles.zoom_cached(17, 2)  # niveau fin : à la volée


def test_unstored_level_rendered_on_the_fly_without_disk(tmp_path, monkeypatch):
    """Niveau non stocké : tuile rendue, rien d'écrit sur disque, seconde
    demande servie par le cache mémoire."""
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", True)
    monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 16)
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    z = 17
    x, y = _tile_of_cell(1054, 6882, z)
    calls = []
    real = tiles.render_tile
    monkeypatch.setattr(tiles, "render_tile", lambda *a, **k: calls.append(1) or real(*a, **k))
    data = tiles.get_tile(tmp_path, "aspect", z, x, y, 1, "png")
    assert data and data[:4] == b"\x89PNG"
    assert not tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 1, "png").exists()
    assert tiles.get_tile(tmp_path, "aspect", z, x, y, 1, "png") == data and len(calls) == 1


def test_remote_payload_failure_not_retried_each_call(monkeypatch):
    """Amont injoignable sans inventaire connu : l'échec est mémorisé (TTL).

    Sinon chaque appel (plusieurs par /api/map/meta) repaie le délai de
    connexion : l'interface ne se chargeait plus quand le worker était éteint.
    """
    import urllib.request
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
    monkeypatch.setattr(tiles, "_remote_cache",
                        {"payload": None, "at": 0.0, "index": None, "root": None})
    calls = []

    def boom(*a, **k):
        calls.append(1)
        raise OSError("hôte injoignable")

    monkeypatch.setattr(urllib.request, "urlopen", boom)
    for _ in range(3):
        assert tiles._remote_payload() is None
    assert len(calls) == 1


def test_fetch_source_offline_breaker(tmp_path, monkeypatch):
    """Source amont en échec : les suivantes échouent sans attendre le délai
    réseau pendant la suspension (la maintenance passe au rendu local)."""
    import urllib.request
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "_SOURCE_OFFLINE", {"until": 0.0})
    calls = []

    def boom(*a, **k):
        calls.append(1)
        raise OSError("hôte injoignable")

    monkeypatch.setattr(urllib.request, "urlopen", boom)
    assert tiles._fetch_source("http://amont/a", tmp_path / "a.avif") is False
    assert tiles._fetch_source("http://amont/b", tmp_path / "b.avif") is False
    assert len(calls) == 1


def test_fetched_source_dated_to_upstream_version(tmp_path, monkeypatch):
    """Une source rapatriée porte la date de version amont : quand l'amont
    s'éteint, l'index local retrouve les mêmes dates et les tuiles déjà
    faites restent fraîches (pas de pyramide entière à refaire)."""
    from lidar_pipeline import tiles

    def fake_fetch(url, dest):
        dest.write_bytes(b"x")
        return True

    monkeypatch.setattr(tiles, "_fetch_source", fake_fetch)
    src = tiles._Source(tmp_path / "q.avif", (0, 0, 1, 1), 0.2,
                        url="http://amont/q.avif", version=1700000000000)
    assert src.ensure() is True
    assert (tmp_path / "q.avif").stat().st_mtime == 1700000000.0


def test_tile_refreshed_when_older_dalle_appears(tmp_path):
    """Dalle entrée dans l'inventaire APRÈS le rendu d'une tuile qui la couvre,
    mais avec une date de version plus ancienne (écrite avant, inventoriée
    après) : la tuile doit être périmée — sinon trou permanent à ce niveau,
    sur le disque, en mémoire et dans le navigateur (stamp inchangé)."""
    import io
    import os
    from PIL import Image
    from lidar_pipeline import tiles
    tiles.clear_source_cache()
    tiles._mem_tiles.clear()
    z = 10
    assert _tile_of_cell(1054, 6882, z) == _tile_of_cell(1055, 6882, z)
    x, y = _tile_of_cell(1054, 6882, z)
    _make_dalle(tmp_path, 1054, 6882, ["slope"], color=(200, 30, 30))
    tiles.source_index(tmp_path, force=True)
    first = tiles.get_tile(tmp_path, "slope", z, x, y)
    assert first is not None
    stamp = tiles.tiles_stamp(tmp_path)

    _make_dalle(tmp_path, 1055, 6882, ["slope"], color=(30, 200, 30))
    for f in tmp_path.rglob("*1055_6882*"):
        os.utime(f, (1_000_000, 1_000_000))   # version antérieure à la tuile
    tiles.source_index(tmp_path, force=True)
    assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "pending"
    second = tiles.get_tile(tmp_path, "slope", z, x, y)
    img = Image.open(io.BytesIO(second)).convert("RGBA")
    assert any(g > 150 and r < 80 and a == 255
               for r, g, b, a in img.getdata()), "nouvelle dalle absente de la tuile"
    assert tiles.tiles_stamp(tmp_path) > stamp
    # Registre persistant : un redémarrage ne réinvalide rien
    tiles._index_cache.clear()
    tiles.source_index(tmp_path, force=True)
    assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "fresh"
    tiles.clear_source_cache()


def test_existing_cache_without_registry_is_refreshed_once(tmp_path):
    """Mise à jour : cache de tuiles présent mais pas de registre — les
    tuiles existantes (peut-être trouées) sont périmées une seule fois."""
    from lidar_pipeline import tiles
    tiles._seen_cache.clear()
    tiles._mem_tiles.clear()
    z = 10
    x, y = _tile_of_cell(1054, 6882, z)
    _make_dalle(tmp_path, 1054, 6882, ["slope"])
    tiles.source_index(tmp_path, force=True)
    assert tiles.get_tile(tmp_path, "slope", z, x, y) is not None
    (tmp_path / tiles.TILE_DIRNAME / tiles._SEEN_FILE).unlink()   # version précédente
    tiles._seen_cache.clear()
    tiles.source_index(tmp_path, force=True)
    assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "pending"
    tiles.get_tile(tmp_path, "slope", z, x, y)
    tiles._seen_cache.clear()
    tiles.source_index(tmp_path, force=True)
    assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "fresh"


def test_webp_subtiles_indexed_and_rendered_nearest(tmp_path):
    """Couche densité : quadrants .webp sans perte reconnus comme palier fin,
    et rendus au plus proche voisin (16 gris exacts même agrandis)."""
    from PIL import Image
    from lidar_pipeline import tiles
    base = _make_dalle(tmp_path, 1054, 6882, ["densite_sol"])
    sub = tmp_path / "index_subtiles"
    sub.mkdir(parents=True, exist_ok=True)
    for i in range(2):
        for j in range(2):
            im = Image.new("L", (8, 8), 0)
            im.paste(255, (0, 0, 4, 8))  # moitié blanche, moitié noire
            im.save(str(sub / f"{base}_densite_sol_{i}_{j}.webp"), format="WEBP", lossless=True)
    tiers = tiles.source_index(tmp_path, force=True)["densite_sol"][(1054, 6882)]
    quads = next(t for t in tiers if len(t) == 4)
    assert all(q.path.suffix == ".webp" for q in quads)
    assert "densite_sol" in tiles.NEAREST_LAYERS
    # z17 : résolution visée (~0,8 m) atteinte par les quadrants (0,5 m ici)
    x, y = _tile_of_cell(1054, 6882, 17)
    img = tiles.render_tile(tmp_path, "densite_sol", 17, x, y)
    assert img is not None
    grays = {p[0] for p in img.getdata() if p[3] == 255}
    assert grays <= {0, 255}  # aucun gris intermédiaire inventé


def test_remote_quality_persisted_locally(tmp_path, monkeypatch):
    """La table qualité amont est recopiée en sidecars locaux (Pi autonome)."""
    from lidar_pipeline import tiles
    from lidar_pipeline.quality import read_quality
    payload = _remote_payload()
    base = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
    payload["quality"] = {base: {"version": 1, "ground_density": 3.0},
                          "../evasion": {"version": 1}}
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont")
    monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: payload)
    monkeypatch.setattr(tiles, "_remote_cache", dict(tiles._remote_cache, index=None, built=None))
    tiles._remote_index(tmp_path, force=True)
    assert read_quality(tmp_path, base) == {"version": 1, "ground_density": 3.0}
    assert not (tmp_path / "evasion.json").exists()
    assert list((tmp_path / "quality").iterdir()) == [tmp_path / "quality" / f"{base}.json"]