"""Tests du serveur de tuiles XYZ et de sa découverte (TileJSON, WMTS, JOSM)."""

import asyncio
import json
import xml.etree.ElementTree as ET

from lidar_pipeline.tests.test_tiles import _make_dalle, _tile_of_cell


class _FakeRequest:
    """Requête minimale : seules l'origine et les en-têtes de proxy comptent."""

    def __init__(self, host="carte.local:8975", proto=None, scheme="http"):
        self.headers = {"host": host}
        if proto:
            self.headers["x-forwarded-proto"] = proto
        self.url = type("U", (), {"scheme": scheme, "hostname": "carte.local",
                                  "port": 8975})()


def _setup(tmp_path, monkeypatch, layers=("aspect", "slope"), panel=None):
    """Dalle factice ; panel=None lève la restriction PANEL_VIZ pour que les
    couches de test (aspect, slope) soient servies."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import index, tiles
    monkeypatch.setattr(index, "PANEL_VIZ", panel)
    _make_dalle(tmp_path, 1054, 6882, list(layers))
    monkeypatch.setattr(mapserve, "OUTPUT_DIR", tmp_path)
    tiles.source_index(tmp_path, force=True)
    return mapserve


def test_tile_png_is_256_rgba(tmp_path, monkeypatch):
    """La tuile canonique est un PNG RGBA de 256 px, avec en-têtes CORS."""
    import io
    from PIL import Image
    mapserve = _setup(tmp_path, monkeypatch)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    resp = asyncio.run(mapserve.tile("aspect", z, x, f"{y}.png"))
    assert resp.media_type == "image/png"
    assert resp.headers["Access-Control-Allow-Origin"] == "*"
    assert resp.headers["X-Tile-Empty"] == "0"
    img = Image.open(io.BytesIO(resp.body))
    assert img.size == (256, 256) and img.mode in ("RGBA", "P")


def test_tile_2x_is_512_webp(tmp_path, monkeypatch):
    """Le palier @2x sert du WebP 512 px (interface interne)."""
    import io
    from PIL import Image
    mapserve = _setup(tmp_path, monkeypatch)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    resp = asyncio.run(mapserve.tile("aspect", z, x, f"{y}@2x.webp"))
    assert resp.media_type == "image/webp"
    assert Image.open(io.BytesIO(resp.body)).size == (512, 512)


def test_tile_outside_data_is_transparent(tmp_path, monkeypatch):
    """Hors données : tuile transparente (et pas une erreur) pour JOSM/iD."""
    import io
    from PIL import Image
    mapserve = _setup(tmp_path, monkeypatch)
    resp = asyncio.run(mapserve.tile("aspect", 15, 1, "1.png"))
    assert resp.headers["X-Tile-Empty"] == "1"
    img = Image.open(io.BytesIO(resp.body)).convert("RGBA")
    assert img.getextrema()[3] == (0, 0)


def test_tile_cache_headers(tmp_path, monkeypatch):
    """URL versionnée → cache immuable ; sans version → revalidation courte."""
    mapserve = _setup(tmp_path, monkeypatch)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    plain = asyncio.run(mapserve.tile("aspect", z, x, f"{y}.png"))
    assert "must-revalidate" in plain.headers["Cache-Control"]
    versioned = asyncio.run(mapserve.tile("aspect", z, x, f"{y}.png", v="123"))
    assert "immutable" in versioned.headers["Cache-Control"]


def test_tile_rejects_bad_input(tmp_path, monkeypatch):
    """Nom de tuile, couche, zoom et indices sont validés (404)."""
    from fastapi import HTTPException
    import pytest
    mapserve = _setup(tmp_path, monkeypatch)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    for layer, zz, xx, name in (("aspect", z, x, "abc.png"),      # nom invalide
                                ("aspect", z, x, f"{y}.tif"),     # format refusé
                                ("inconnue", z, x, f"{y}.png"),   # couche absente
                                ("../etc", z, x, f"{y}.png"),     # traversée
                                ("aspect", 2, x, f"{y}.png"),     # zoom trop bas
                                ("aspect", 25, x, f"{y}.png"),    # zoom trop haut
                                ("aspect", z, 10 ** 9, f"{y}.png")):  # hors grille
        with pytest.raises(HTTPException) as e:
            asyncio.run(mapserve.tile(layer, zz, xx, name))
        assert e.value.status_code == 404


def test_tile_upstream_fallback(tmp_path, monkeypatch):
    """Tuile absente localement : rapatriée depuis LIDAR_MAPS_URL et mise en cache."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    _setup(tmp_path, monkeypatch)
    monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
    monkeypatch.setattr(mapserve, "_UPSTREAM", {"fails": 0, "until": 0.0})
    called = {}

    def fake_fetch(rel_path):
        called["path"] = rel_path
        return b"\x89PNG\r\n\x1a\nFAKE"

    monkeypatch.setattr(mapserve, "_fetch_upstream", fake_fetch)
    # Zone sans donnée locale → le rendu local échoue, l'amont prend le relais
    resp = asyncio.run(mapserve.tile("aspect", 15, 1, "1.png"))
    assert called["path"] == "tiles/aspect/15/1/1.png"
    assert resp.body == b"\x89PNG\r\n\x1a\nFAKE"
    assert tiles.tile_cache_path(tmp_path, "aspect", 15, 1, 1).is_file()


def test_upstream_breaker(monkeypatch):
    """3 échecs consécutifs suspendent les tentatives amont ; un succès réarme."""
    import lidar_pipeline.mapserve as mapserve
    monkeypatch.setattr(mapserve, "_UPSTREAM", {"fails": 0, "until": 0.0})
    for _ in range(2):
        mapserve._upstream_mark(False)
    assert mapserve._upstream_offline() is False
    mapserve._upstream_mark(False)
    assert mapserve._upstream_offline() is True
    mapserve._upstream_mark(True)
    assert mapserve._upstream_offline() is False


def test_tilejson_is_valid(tmp_path, monkeypatch):
    """TileJSON 3.0.0 : champs obligatoires, URL XYZ absolue, attribution."""
    mapserve = _setup(tmp_path, monkeypatch)
    data = json.loads(mapserve.tilejson("aspect", _FakeRequest()).body)
    assert data["tilejson"] == "3.0.0"
    assert data["scheme"] == "xyz"
    assert data["tiles"] == ["http://carte.local:8975/tiles/aspect/{z}/{x}/{y}.png"]
    assert data["minzoom"] < data["maxzoom"] <= 19
    assert len(data["bounds"]) == 4 and data["bounds"][0] < data["bounds"][2]
    assert "IGN" in data["attribution"]


def test_tilejson_unknown_layer(tmp_path, monkeypatch):
    """Couche absente du disque → 404."""
    from fastapi import HTTPException
    import pytest
    mapserve = _setup(tmp_path, monkeypatch)
    with pytest.raises(HTTPException) as e:
        mapserve.tilejson("wavelet", _FakeRequest())
    assert e.value.status_code == 404


def test_base_url_honours_proxy(tmp_path, monkeypatch):
    """Derrière un proxy TLS, les URL publiées sont en https."""
    mapserve = _setup(tmp_path, monkeypatch)
    data = json.loads(mapserve.tilejson(
        "aspect", _FakeRequest(host="lidar.example.fr", proto="https")).body)
    assert data["tiles"][0].startswith("https://lidar.example.fr/tiles/")


def test_josm_imagery_xml(tmp_path, monkeypatch):
    """Fichier d'imagerie JOSM : une entrée TMS par couche, URL en {zoom}/{x}/{y}."""
    mapserve = _setup(tmp_path, monkeypatch)
    xml = mapserve.josm_imagery(_FakeRequest()).body.decode("utf-8")
    root = ET.fromstring(xml)
    ns = {"j": "http://josm.openstreetmap.de/maps-1.0"}
    entries = root.findall("j:entry", ns)
    assert {e.find("j:id", ns).text for e in entries} == {
        "lidar-hd-aspect", "lidar-hd-slope"}
    for entry in entries:
        assert entry.find("j:type", ns).text == "tms"
        assert "{zoom}/{x}/{y}.png" in entry.find("j:url", ns).text
        assert entry.find("j:max-zoom", ns).text == "19"
        assert entry.find("j:bounds", ns) is not None


def test_wmts_capabilities(tmp_path, monkeypatch):
    """Capacités WMTS : couches, grille GoogleMapsCompatible, gabarit REST."""
    mapserve = _setup(tmp_path, monkeypatch)
    xml = mapserve.wmts_capabilities(_FakeRequest()).body.decode("utf-8")
    root = ET.fromstring(xml)
    ns = {"w": "http://www.opengis.net/wmts/1.0",
          "ows": "http://www.opengis.net/ows/1.1"}
    ids = [l.find("ows:Identifier", ns).text
           for l in root.findall(".//w:Layer", ns)]
    assert set(ids) == {"aspect", "slope"}
    # `.//w:TileMatrixSet` attraperait d'abord le renvoi de
    # TileMatrixSetLink (simple texte) : on vise la définition.
    tms = root.find("w:Contents/w:TileMatrixSet", ns)
    assert tms.find("ows:Identifier", ns).text == "GoogleMapsCompatible"
    levels = [m.find("ows:Identifier", ns).text
              for m in tms.findall("w:TileMatrix", ns)]
    assert levels[0] == "5" and levels[-1] == "19"
    template = root.find(".//w:ResourceURL", ns).get("template")
    assert template.endswith("/{TileMatrix}/{TileCol}/{TileRow}.png")


def test_map_meta(tmp_path, monkeypatch):
    """/api/map/meta décrit les couches et le contrat de tuilage de l'interface."""
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente", "aspect", "slope"))
    meta = mapserve.map_meta()
    assert {l["key"] for l in meta["layers"]} == {"relief_oriente", "aspect", "slope"}
    assert all(l["label"] for l in meta["layers"])
    assert meta["tile_url"] == "tiles/{layer}/{z}/{x}/{y}@2x.avif"
    assert meta["even_levels"] is False       # pyramide complète générée d'avance
    assert meta["tile_size"] == 512 and meta["zoom_offset"] == -1
    assert meta["max_native_zoom"] == 19
    assert meta["bounds"] and meta["stamp"] > 0
    assert meta["default_main"] == "relief_oriente"
    assert meta["precision_layer"] is None      # densité absente de la fixture
    assert meta["default_mode"] == "relief"


def test_only_panel_layers_are_served(tmp_path, monkeypatch):
    """Seules les couches de PANEL_VIZ sont affichées et servies ; la
    génération depuis la carte ne produit qu'elles."""
    from fastapi import HTTPException
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente", "aspect", "slope"),
                      panel=("relief_oriente",))
    meta = mapserve.map_meta()
    assert [l["key"] for l in meta["layers"]] == ["relief_oriente"]
    with pytest.raises(HTTPException):
        asyncio.run(mapserve.tile("aspect", 15, 1, "1.png"))
    assert mapserve._panel_viz_steps() == ["relief_oriente"]


def test_map_tile_info(tmp_path, monkeypatch):
    """/api/map/tile identifie la dalle sous un point et lit ses métadonnées."""
    mapserve = _setup(tmp_path, monkeypatch)
    dtm = tmp_path / "DTM"
    dtm.mkdir(parents=True, exist_ok=True)
    (dtm / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_dtm_method.txt").write_text(
        "ign", encoding="utf-8")
    from lidar_pipeline.tiles import _transformer
    lon, lat = _transformer("EPSG:2154", "EPSG:4326").transform(1054500, 6881500)
    data = mapserve.map_tile_info(lat=lat, lng=lon)
    assert data["found"] is True
    assert (data["col"], data["row"]) == (1054, 6882)
    assert data["method"] == "ign"
    assert data["layers"] == ["aspect", "slope"]
    # Point hors zone : réponse explicite, sans erreur
    assert mapserve.map_tile_info(lat=45.0, lng=1.0)["found"] is False


def test_healthz_and_assets(tmp_path, monkeypatch):
    """Sonde de santé et assets de l'interface (revalidés, jamais immuables)."""
    from fastapi import HTTPException
    import pytest
    mapserve = _setup(tmp_path, monkeypatch)
    assert mapserve.healthz()["layers"] == 2
    resp = mapserve.assets("app.js")
    assert resp.headers["Cache-Control"] == "no-cache, must-revalidate"
    with pytest.raises(HTTPException):
        mapserve.assets("introuvable.js")


def test_ui_main_layer_and_precision_modes():
    """Une couche d'affichage principal et la précision : relief seul,
    précision seule, ou précision en « produit » sur le relief.

    Plus de pile réordonnable ni de fusion par couche. Le produit est
    cantonné à la pile LiDAR par un conteneur isolé (sinon il assombrirait
    aussi le fond de carte). Bascule rapide au clavier (touche P).
    """
    from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
    for mode in ('data-mode="relief"', 'data-mode="precision"', 'data-mode="both"'):
        assert mode in _MAP_HTML
    assert 'id="precOpacity"' in _MAP_HTML and 'id="precLegend"' in _MAP_HTML
    assert 'id="mainSel"' in _MAP_HTML
    assert "blend = 'multiply'" in _MAP_JS
    assert "createPane('lidarStack')" in _MAP_JS and "isolation = 'isolate'" in _MAP_JS
    assert "e.key === 'p'" in _MAP_JS
    # Légende en 16 paliers exacts (échelle log de la densité)
    assert "const PREC_LEVELS = 16;" in _MAP_JS and "repeat(16, 1fr)" in _MAP_CSS
    # Pile et fusions supprimées
    for gone in ("function moveLayer(", "reorderRelative", "stackBlend", "dragstart", "blendOptions"):
        assert gone not in _MAP_JS, gone
    assert "Haut de pile" not in _MAP_HTML
    # Lien partagé : principale, mode:opacité, fond
    assert "'&M=' + STATE.main" in _MAP_JS
    assert "'&P=' + STATE.mode" in _MAP_JS


def test_map_ui_tile_load_indicator_and_selection_pane():
    """Chargement : compteur de tuiles restantes + contour de dalle visible.

    Le compteur lit l'état réel des tuiles (une tuile retirée en cours de
    chargement n'émet aucun événement de fin). Le contour de sélection vit
    dans un panneau au-dessus de la pile LiDAR (z-index 400), sinon les
    tuiles le masquent.
    """
    from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
    assert 'id="tileload"' in _MAP_HTML and 'id="tileloadText"' in _MAP_HTML
    assert "@keyframes tlspin" in _MAP_CSS
    assert "ZoomAnimTileLayer.addInitHook" in _MAP_JS
    assert "t.current && !t.loaded" in _MAP_JS
    assert "['tileSel', 680]" in _MAP_JS
    assert "pane: 'tileSel'" in _MAP_JS


def test_defaults_roundtrip(tmp_path, monkeypatch):
    """L'affichage réglé depuis la carte devient la configuration servie à tous."""
    import lidar_pipeline.mapserve as mapserve
    _setup(tmp_path, monkeypatch, layers=("relief_oriente", "aspect", "densite_sol"))
    monkeypatch.setattr(mapserve, "DEFAULTS_FILE", tmp_path / ".map-defaults.json")

    # Sans enregistrement : réglages du registre (index.py)
    meta = mapserve.map_meta()
    assert meta["defaults_saved"] is False
    assert meta["default_main"] == "relief_oriente"
    assert meta["precision_layer"] == "densite_sol"
    assert meta["default_mode"] == "relief"
    assert meta["default_precision_opacity"] == 0.6

    req = mapserve.DefaultsRequest(main="aspect", mode="both", precision_opacity=0.4,
                                   base={"on": True, "opacity": 0.4, "dark": False})
    assert mapserve.set_defaults(req)["enregistré"] is True

    meta = mapserve.map_meta()
    assert meta["defaults_saved"] is True
    assert meta["default_main"] == "aspect"
    assert meta["default_mode"] == "both"
    assert meta["default_precision_opacity"] == 0.4
    assert meta["default_base"] == {"on": True, "opacity": 0.4, "dark": False}
    # Persisté sur disque : survit au redémarrage du conteneur
    assert (tmp_path / ".map-defaults.json").is_file()
    assert mapserve.get_defaults()["defaults"]["mode"] == "both"

    # Retrait : retour aux réglages du registre
    assert mapserve.clear_defaults()["supprimé"] is True
    assert mapserve.map_meta()["defaults_saved"] is False


def test_defaults_legacy_stack_file_falls_back(tmp_path, monkeypatch):
    """Fichier de l'ancienne pile (order/on/blend) ou couche disparue :
    couche principale du registre, mode relief ; le fond est conservé."""
    import json
    import lidar_pipeline.mapserve as mapserve
    _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    defaults = tmp_path / ".map-defaults.json"
    monkeypatch.setattr(mapserve, "DEFAULTS_FILE", defaults)
    defaults.write_text(json.dumps({
        "order": ["aspect"], "on": ["aspect"], "opacity": {}, "blend": {},
        "base": {"on": True, "opacity": 0.5, "dark": True}, "stack_blend": "normal"}),
        encoding="utf-8")
    meta = mapserve.map_meta()
    assert meta["default_main"] == "relief_oriente"
    assert meta["default_mode"] == "relief"
    assert meta["default_base"]["opacity"] == 0.5
    defaults.write_text(json.dumps({"main": "aspect", "mode": "vaudou"}), encoding="utf-8")
    meta = mapserve.map_meta()
    assert meta["default_main"] == "relief_oriente" and meta["default_mode"] == "relief"


def test_defaults_sanitised(tmp_path, monkeypatch):
    """Couche inconnue ou précision refusée comme principale, mode validé, opacités bornées."""
    import lidar_pipeline.mapserve as mapserve
    _setup(tmp_path, monkeypatch, layers=("relief_oriente", "densite_sol"))
    monkeypatch.setattr(mapserve, "DEFAULTS_FILE", tmp_path / ".map-defaults.json")
    data = mapserve.set_defaults(mapserve.DefaultsRequest(
        main="inexistante", mode="vaudou", precision_opacity=5,
        base={"opacity": -3}))["defaults"]
    assert data["main"] is None
    assert data["mode"] == "relief"
    assert data["precision_opacity"] == 1.0          # borné à 1
    assert data["base"]["opacity"] == 0.0            # borné à 0
    data = mapserve.set_defaults(mapserve.DefaultsRequest(main="densite_sol"))["defaults"]
    assert data["main"] is None                      # la précision n'est pas une principale


def test_ui_applies_server_defaults():
    """L'interface part des défauts du serveur et sait les (re)définir."""
    from lidar_pipeline.mapui import _MAP_HTML, _MAP_JS
    assert 'id="btnDefault"' in _MAP_HTML and 'id="btnReset"' in _MAP_HTML
    assert "api/map/defaults" in _MAP_JS
    assert "meta.default_main" in _MAP_JS and "meta.default_mode" in _MAP_JS
    assert "meta.default_base" in _MAP_JS
    assert "meta.default_precision_opacity" in _MAP_JS
    # Réinitialiser oublie l'état local avant de reprendre celui du serveur
    assert "localStorage.removeItem(LS_KEY)" in _MAP_JS


def test_ui_uses_standard_tilelayer():
    """L'interface s'appuie sur le LOD natif de Leaflet, pas sur un palier maison."""
    from lidar_pipeline.mapui import _MAP_JS, render_html, ui_version
    assert "L.TileLayer.extend(" in _MAP_JS      # TileLayer Leaflet, LOD natif
    assert "tileSize: META.tile_size" in _MAP_JS
    assert "zoomOffset: META.zoom_offset" in _MAP_JS
    assert "maxNativeZoom: META.max_native_zoom" in _MAP_JS
    # Le niveau natif impair (19 = 0,2 m/px) est demandé tel quel, jamais
    # replié sur le pair inférieur (affichage plafonné à 0,4 m/px).
    assert "z % 2 && (maxN === undefined || z < maxN)" in _MAP_JS
    # Plus aucune mécanique de l'ancienne carte
    for banned in ("quadMatrix3d", "FULL_CAP", "ovTick", "layoutTiles"):
        assert banned not in _MAP_JS
    html = render_html()
    assert ui_version() in html and "__UI_VERSION__" not in html


def test_write_map_assets(tmp_path):
    """Les assets écrits portent la version et embarquent Leaflet vendorisé."""
    from lidar_pipeline.mapui import _MAP_JS, ui_version, write_map_assets
    write_map_assets(tmp_path / "ui")
    js = (tmp_path / "ui" / "app.js").read_text(encoding="utf-8")
    assert js == _MAP_JS.replace("__UI_VERSION__", ui_version())
    assert (tmp_path / "ui" / "app.css").read_text(encoding="utf-8").startswith("/*")
    assert (tmp_path / "ui" / "vendor" / "leaflet" / "leaflet.js").is_file()


def test_warm_endpoint(tmp_path, monkeypatch):
    """/api/map/warm lance le pré-calcul en tâche de fond puis rend compte."""
    import time
    mapserve = _setup(tmp_path, monkeypatch)
    monkeypatch.setattr(mapserve, "_warm", {"running": False, "done": None,
                                            "result": None})
    req = mapserve.WarmRequest(layers=["aspect"], z_min=12, z_max=12)
    assert mapserve.map_warm(req)["couches"] == ["aspect"]
    for _ in range(200):
        if not mapserve._warm["running"]:
            break
        time.sleep(0.05)
    assert mapserve.map_warm_status()["result"]["rendues"] >= 1


def test_tile_locks_are_refcounted(tmp_path, monkeypatch):
    """Verrou par tuile : partagé pendant le rendu, retiré une fois libéré.

    Sans compteur, le registre grossit sans fin (des millions de tuiles) ou un
    verrou disparaît alors qu'un autre fil l'attend — deux rendus concurrents
    de la même tuile.
    """
    mapserve = _setup(tmp_path, monkeypatch)
    key = ("aspect", 15, 1, 1, 1, "png")
    a = mapserve._acquire_lock(key)
    b = mapserve._acquire_lock(key)
    assert a is b and a[1] == 2
    mapserve._release_lock(key, a)
    assert key in mapserve._locks
    mapserve._release_lock(key, b)
    assert key not in mapserve._locks


def test_concurrent_requests_render_once(tmp_path, monkeypatch):
    """Deux requêtes simultanées sur la même tuile ne la rendent qu'une fois."""
    import threading
    import lidar_pipeline.mapserve as mapserve
    _setup(tmp_path, monkeypatch)
    calls = []
    real = mapserve.tiles_mod.get_tile

    def slow_get_tile(*args, **kwargs):
        calls.append(args)
        threading.Event().wait(0.05)
        return real(*args, **kwargs)

    monkeypatch.setattr(mapserve.tiles_mod, "get_tile", slow_get_tile)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    results = []
    threads = [threading.Thread(
        target=lambda: results.append(mapserve._tile_bytes("aspect", z, x, y, 1, "png")))
        for _ in range(4)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert all(r == results[0] for r in results)
    # Les appels suivants lisent le cache disque : un seul rendu effectif
    assert len(calls) == 4          # chaque requête interroge get_tile…
    assert mapserve._locks == {}    # …mais sérialisées, et les verrous libérés


def test_tile_bytes_survives_render_error(tmp_path, monkeypatch):
    """Un rendu qui échoue rend une tuile vide sans laisser de verrou fuité."""
    import lidar_pipeline.mapserve as mapserve
    _setup(tmp_path, monkeypatch)

    def boom(*a, **k):
        raise RuntimeError("source corrompue")

    monkeypatch.setattr(mapserve.tiles_mod, "get_tile", boom)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    assert mapserve._tile_bytes("aspect", z, x, y, 1, "png") is None
    assert mapserve._locks == {}


def test_render_concurrency_is_capped(tmp_path, monkeypatch):
    """Le plafond de rendus simultanés est tenu (petit serveur protégé).

    Sans lui, autant de rendus que de requêtes en vol : chacun décode une dalle
    de plusieurs dizaines de Mo — mémoire et CPU d'un Raspberry Pi saturés.
    """
    import threading
    import lidar_pipeline.mapserve as mapserve
    _setup(tmp_path, monkeypatch)

    # Plafond volontairement bas pour le test
    monkeypatch.setattr(mapserve, "_render_sem", threading.Semaphore(2))
    state = {"now": 0, "peak": 0}
    guard = threading.Lock()
    real = mapserve.tiles_mod.get_tile

    def slow(*args, **kwargs):
        with guard:
            state["now"] += 1
            state["peak"] = max(state["peak"], state["now"])
        try:
            threading.Event().wait(0.05)
            return real(*args, **kwargs)
        finally:
            with guard:
                state["now"] -= 1

    monkeypatch.setattr(mapserve.tiles_mod, "get_tile", slow)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    # Tuiles DIFFÉRENTES : le verrou par tuile ne masque pas le sémaphore
    threads = [threading.Thread(target=mapserve._tile_bytes,
                                args=("aspect", z, x + i, y, 1, "png"))
               for i in range(8)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert state["peak"] <= 2, state["peak"]
    assert mapserve._locks == {}


def test_worker_limits_are_configurable(monkeypatch):
    """Les plafonds viennent de l'environnement, avec un défaut modeste."""
    import importlib
    import lidar_pipeline.mapserve as mapserve
    import lidar_pipeline.tiles as tiles

    # Défauts : 2 rendus, 2 téléchargements — taillés pour un petit serveur
    assert mapserve.TILE_WORKERS == 2
    assert tiles.FETCH_WORKERS == 2

    monkeypatch.setenv("LIDAR_TILE_WORKERS", "6")
    monkeypatch.setenv("LIDAR_TILE_FETCH_WORKERS", "3")
    try:
        importlib.reload(tiles)
        importlib.reload(mapserve)
        assert mapserve.TILE_WORKERS == 6
        assert tiles.FETCH_WORKERS == 3
        # Valeur absurde : jamais zéro fil (blocage complet du service)
        monkeypatch.setenv("LIDAR_TILE_WORKERS", "0")
        importlib.reload(mapserve)
        assert mapserve.TILE_WORKERS == 1
    finally:
        monkeypatch.delenv("LIDAR_TILE_WORKERS", raising=False)
        monkeypatch.delenv("LIDAR_TILE_FETCH_WORKERS", raising=False)
        importlib.reload(tiles)
        importlib.reload(mapserve)


def _reset_bg(mapserve):
    mapserve._bg["queue"].clear()
    mapserve._bg["queued"].clear()
    mapserve._bg["snapshot"] = None
    mapserve._bg["incomplete"] = {}
    mapserve._bg["stats"].update(rendues=0, a_jour=0, vides=0, scans=0,
                                 dernier_scan=None, ajouts_dernier_scan=0)


def test_tile_cache_only_never_renders(tmp_path, monkeypatch):
    """Mode cache seule : tuile manquante = transparente non mémorisable."""
    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)
    mapserve = _setup(tmp_path, monkeypatch, layers=("slope",))
    from lidar_pipeline import tiles
    monkeypatch.setenv("LIDAR_TILE_CACHE_ONLY", "1")
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    resp = asyncio.run(mapserve.tile("slope", z, x, f"{y}.png"))
    assert resp.headers["X-Tile-Pending"] == "1"
    assert resp.headers["X-Tile-Empty"] == "1"
    assert resp.headers["Cache-Control"] == "no-store"
    assert not tiles.tile_cache_path(tmp_path, "slope", z, x, y).exists()
    # Une fois rendue par ailleurs : servie depuis le cache, mémorisable
    tiles.get_tile(tmp_path, "slope", z, x, y)
    resp = asyncio.run(mapserve.tile("slope", z, x, f"{y}.png", v="s"))
    assert resp.headers["X-Tile-Empty"] == "0"
    assert "X-Tile-Pending" not in resp.headers
    assert "immutable" in resp.headers["Cache-Control"]


def test_background_scan_detects_and_fills(tmp_path, monkeypatch):
    """La maintenance détecte les dalles, remplit la pyramide, puis se tait."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["slope"])
    monkeypatch.setattr(mapserve, "OUTPUT_DIR", tmp_path)
    monkeypatch.setenv("LIDAR_TILE_BACKGROUND_MAX_Z", "14")
    monkeypatch.setenv("LIDAR_TILE_BACKGROUND_SCALE", "1")
    monkeypatch.setenv("LIDAR_TILE_BACKGROUND_FMT", "png")
    monkeypatch.setenv("LIDAR_TILE_BACKGROUND_PAUSE", "0")
    _reset_bg(mapserve)
    try:
        # Premier scan : tout est nouveau, bas niveaux en tête de file
        assert mapserve.background_scan(tmp_path) > 0
        zs = [t[1] for t in mapserve._bg["queue"]]
        assert zs == sorted(zs) and max(zs) <= 14
        processed = 0
        while mapserve._bg["queue"] and processed < 1000:
            mapserve._bg_process_one()
            processed += 1
        assert not mapserve._bg["queue"]
        assert mapserve._bg["stats"]["rendues"] > 0
        x, y = _tile_of_cell(1054, 6882, 14)
        assert tiles.tile_cache_path(
            tmp_path, "slope", 14, x, y, 1, "png").exists()
        # Rien n'a changé : plus aucun ajout
        assert mapserve.background_scan(tmp_path) == 0
        # Une dalle nouvelle (worker) : sa pyramide repart seule en file
        _make_dalle(tmp_path, 1055, 6882, ["slope"])
        assert mapserve.background_scan(tmp_path) > 0
    finally:
        _reset_bg(mapserve)


# ---------------------------------------------------------------------------
# Génération de tuiles (portée de la webapp historique)
# ---------------------------------------------------------------------------

import types

from fastapi import HTTPException
import pytest


class _ClientRequest:
    """Requête minimale avec IP de connexion (restriction LIDAR_REGEN_CIDR)."""

    def __init__(self, ip, xff=None):
        self.client = types.SimpleNamespace(host=ip)
        self.headers = {"x-forwarded-for": xff} if xff else {}


def _gen_setup(tmp_path, monkeypatch):
    import lidar_pipeline.mapserve as mapserve
    _make_dalle(tmp_path, 1054, 6882, ["aspect", "slope"])
    monkeypatch.setattr(mapserve, "OUTPUT_DIR", tmp_path)
    monkeypatch.setattr(mapserve, "INPUT_DIR", tmp_path / "input")
    monkeypatch.setattr(mapserve, "GENERATION_URL", "")
    monkeypatch.setattr(mapserve, "LOG_FILE", tmp_path / ".generation.log")
    monkeypatch.setattr(mapserve, "JOB_FILE", tmp_path / ".generation.job.json")
    monkeypatch.setattr(mapserve, "QUEUE_FILE", tmp_path / ".generation.queue.json")
    mapserve._queue.clear()
    mapserve._pipeline_import_cache.clear()
    return mapserve


def test_bbox_to_cells_covers_1km():
    """Une bbox d'1 km² WGS84 sélectionne la cellule L93 correspondante."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline.tiles import _transformer
    x, y = 1054500, 6881500
    w, n = _transformer("EPSG:2154", "EPSG:4326").transform(x - 200, y + 200)
    e, s = _transformer("EPSG:2154", "EPSG:4326").transform(x + 200, y - 200)
    cells = mapserve.bbox_to_cells(w, s, e, n)
    assert (1054, 6882) in cells
    # Cellule (col, row) : X ∈ [col, col+1] km, Y ∈ [row-1, row] km
    assert all(1053 <= c <= 1055 and 6881 <= r <= 6883 for c, r in cells)


def test_preview_bbox_lists_cells_with_corners(tmp_path, monkeypatch):
    """Le preview retourne les cellules à traiter avec leurs coins GPS."""
    mapserve = _gen_setup(tmp_path, monkeypatch)
    from lidar_pipeline.tiles import _transformer
    x, y = 1054500, 6881500
    w, n = _transformer("EPSG:2154", "EPSG:4326").transform(x - 100, y + 100)
    e, s = _transformer("EPSG:2154", "EPSG:4326").transform(x + 100, y - 100)
    req = mapserve.PreviewRequest(bbox=[w, s, e, n], viz=["aspect"])
    d = mapserve.preview(req)
    assert d["capped"] is False
    cells = {(c["col"], c["row"]) for c in d["cells"]}
    assert (1054, 6882) in cells
    first = d["cells"][0]
    assert first["corners"] and len(first["corners"]) == 4


def test_preview_rejects_unknown_viz(tmp_path, monkeypatch):
    mapserve = _gen_setup(tmp_path, monkeypatch)
    req = mapserve.PreviewRequest(bbox=[0.0, 44.0, 0.01, 44.01],
                                  viz=["inconnue"])
    with pytest.raises(HTTPException) as e:
        mapserve.preview(req)
    assert e.value.status_code == 400


def test_preview_all_missing_without_laz(tmp_path, monkeypatch):
    """Passe globale sans dalle dans input/ : liste vide (rien à compléter).

    L'erreur 400 « aucune tuile à compléter » ne concerne que le lancement
    (/api/generate → _resolve_request) : le preview reste informatif.
    """
    mapserve = _gen_setup(tmp_path, monkeypatch)
    (tmp_path / "input").mkdir()
    req = mapserve.PreviewRequest(all_missing=True, viz=["aspect"])
    d = mapserve.preview(req)
    assert d["count"] == 0 and d["cells"] == []


def test_generate_validates_request(tmp_path, monkeypatch):
    mapserve = _gen_setup(tmp_path, monkeypatch)
    for kwargs in ({"tiles": []},
                   {"tiles": [[1054]]},
                   {"tiles": [[c, 6882] for c in range(500)]}):
        req = mapserve.GenerateRequest(**kwargs)
        with pytest.raises(HTTPException) as e:
            mapserve._resolve_request(req)
        assert e.value.status_code == 400
    # Tri nord → sud (row décroissante) puis col croissante
    req = mapserve.GenerateRequest(tiles=[[1055, 6882], [1054, 6883],
                                          [1054, 6882]])
    tiles, _viz = mapserve._resolve_request(req)
    assert tiles == [(1054, 6883), (1054, 6882), (1055, 6882)]


def test_generate_launches_and_queues(tmp_path, monkeypatch):
    mapserve = _gen_setup(tmp_path, monkeypatch)
    launched = []
    monkeypatch.setattr(mapserve, "_launch_job",
                        lambda tiles, viz, req, qid=None:
                        launched.append((tiles, qid)) or ["fake", "cmd"])

    req = mapserve.GenerateRequest(tiles=[[1054, 6882]], viz=["aspect"])
    d = mapserve.generate(req)
    assert d["demarré"] is True and d["tuiles"] == 1

    # Run en cours : la demande part en file (jamais de coupure du travail)
    import threading
    with mapserve._job_lock:
        mapserve._job["proc"] = types.SimpleNamespace(
            poll=lambda: None, pid=4242, terminate=lambda: None, wait=lambda: None)
    d = mapserve.generate(req)
    assert d["demarré"] is False and d["en_file"] == 1
    assert mapserve.generate(req)["en_file"] == 2
    assert len(mapserve._queue_summary()) == 2
    d = mapserve.queue_clear()
    assert d["retirées"] == 2
    with mapserve._job_lock:
        mapserve._job["proc"] = None
    mapserve._queue.clear()


def test_status_reports_rights_and_availability(tmp_path, monkeypatch):
    """/api/status annonce available (pipeline local) et regen_allowed (IP)."""
    mapserve = _gen_setup(tmp_path, monkeypatch)
    monkeypatch.setattr(mapserve, "_pipeline_import_cache", {"ok": True})
    d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
    assert d["available"] is True and d["regen_allowed"] is True
    assert d["running"] is False and d["tiles"] == []

    # Client hors réseau local : génération interdite (boutons masqués)
    d = mapserve.generation_status(_ClientRequest("203.0.113.7"))
    assert d["regen_allowed"] is False

    # X-Forwarded-For honoré seulement depuis un pair de confiance
    d = mapserve.generation_status(
        _ClientRequest("172.25.0.1", xff="203.0.113.7"))
    assert d["regen_allowed"] is False
    d = mapserve.generation_status(
        _ClientRequest("172.25.0.1", xff="192.168.3.10"))
    assert d["regen_allowed"] is True

    # Image légère sans pipeline ni LIDAR_GENERATION_URL : indisponible
    monkeypatch.setattr(mapserve, "_pipeline_import_cache", {"ok": False})
    d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
    assert d["available"] is False


def test_status_proxies_to_worker(tmp_path, monkeypatch):
    """Webapp légère : /api/status relaie l'état du worker, 503 hors ligne."""
    mapserve = _gen_setup(tmp_path, monkeypatch)
    monkeypatch.setattr(mapserve, "GENERATION_URL", "http://worker:8973")
    monkeypatch.setattr(mapserve, "_pipeline_import_cache", {"ok": False})

    def fake_proxy(method, path, payload=None, timeout=30):
        assert (method, path) == ("GET", "/api/status")
        return {"running": True, "tiles": []}

    monkeypatch.setattr(mapserve, "_proxy_generation", fake_proxy)
    d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
    assert d["distant"] is True and d["running"] is True
    assert d["available"] is True and d["regen_allowed"] is True

    def offline(method, path, payload=None, timeout=30):
        raise HTTPException(503, "machine de traitement injoignable")

    monkeypatch.setattr(mapserve, "_proxy_generation", offline)
    d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
    assert d["worker_offline"] is True and d["running"] is None


def test_generate_requires_lan(tmp_path, monkeypatch):
    """Un client hors LIDAR_REGEN_CIDR ne peut pas lancer de génération."""
    mapserve = _gen_setup(tmp_path, monkeypatch)
    with pytest.raises(HTTPException) as e:
        mapserve._require_lan_for_generation(_ClientRequest("203.0.113.7"))
    assert e.value.status_code == 403


def test_cell_at_point(tmp_path, monkeypatch):
    """/api/cell donne la dalle sous un point (sélection manuelle)."""
    mapserve = _gen_setup(tmp_path, monkeypatch)
    from lidar_pipeline.tiles import _transformer
    lon, lat = _transformer("EPSG:2154", "EPSG:4326").transform(1054500, 6881500)
    d = mapserve.cell_at_point(lat=lat, lng=lon)
    assert (d["col"], d["row"]) == (1054, 6882)
    assert d["corners"] and len(d["corners"]) == 4


def test_tiles_inventory_endpoint(tmp_path, monkeypatch):
    """/api/tiles sert index_tiles.json (machines légères LIDAR_SOURCE_URL)."""
    mapserve = _gen_setup(tmp_path, monkeypatch)
    payload = {"tiles": [{"col": 1054, "row": 6882,
                          "viz": {"aspect": {"full": "visualisations/x.avif?v=1"}}}],
               "viz_meta": {}, "stats": {"n_tiles": 1}}
    (tmp_path / "index_tiles.json").write_text(
        __import__("json").dumps(payload), encoding="utf-8")
    d = mapserve.tiles_inventory()
    assert d["tiles"] == payload["tiles"] and d["stamp"] > 0
    # ?stamp inchangé → réponse allégée
    light = mapserve.tiles_inventory(stamp=d["stamp"])
    assert light["tiles"] is None


def test_generation_always_ign_with_edge_buffer(tmp_path, monkeypatch):
    """Classification IGN (sol) et raccord des bords de 100 m imposés, quel
    que soit ce que le client envoie."""
    mapserve = _gen_setup(tmp_path, monkeypatch)
    cmd = mapserve._build_command([(1054, 6882)])
    i = cmd.index("--edge-buffer")
    assert cmd[i + 1] == "100"
    assert cmd[cmd.index("--ground-classification") + 1] == "ign"
    assert cmd[cmd.index("--ign-classes") + 1] == "sol"
    req = mapserve.GenerateRequest(tiles=[[1054, 6882]], ground_class="csf", edge_buffer=False)
    assert not hasattr(req, "ground_class") or getattr(req, "ground_class", None) != "csf"


def test_ui_offers_generation_controls():
    """L'interface embarque les contrôles de génération (＋ Zone, options, run)."""
    from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
    assert 'id="btnZone"' in _MAP_HTML and 'id="btnComplete"' in _MAP_HTML
    assert 'id="gencard"' in _MAP_HTML and 'id="genGo"' in _MAP_HTML
    assert 'id="genStop"' in _MAP_HTML and 'id="genQueueClear"' in _MAP_HTML
    assert 'id="genViz"' in _MAP_HTML
    # Classification et raccord des bords imposés : plus de réglage
    for gone in ('id="genClass"', 'id="genIgnClasses"', 'id="genReclass"', 'id="genEdge"'):
        assert gone not in _MAP_HTML
    # Boutons masqués par défaut : révélés seulement si /api/status autorise
    assert "GEN.available && GEN.allowed" in _MAP_JS
    # Dessin de zone + cadres de run + sondage d'état
    assert "genSetDrawing" in _MAP_JS and "pane: 'genDraw'" in _MAP_JS
    assert "gen-frame-running" in _MAP_CSS and "genDrawFrames" in _MAP_JS
    assert "api/status" in _MAP_JS and "api/preview" in _MAP_JS
    assert "api/generate" in _MAP_JS and "api/stop" in _MAP_JS
    assert "all_missing: true" in _MAP_JS
    # Fin de run : rafraîchissement des couches + relance de la maintenance
    assert "genAfterRun" in _MAP_JS and "api/map/background" in _MAP_JS


def test_cache_only_fetches_missing_tile_upstream(tmp_path, monkeypatch):
    """Cache seule + LIDAR_MAPS_URL : une tuile absente est rapatriee du serveur
    amont (aucun rendu local : les zooms au-dela de la maintenance s'affichent).
    """
    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 lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    _setup(tmp_path, monkeypatch)
    monkeypatch.setenv("LIDAR_TILE_CACHE_ONLY", "1")
    monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
    monkeypatch.setattr(mapserve, "_UPSTREAM", {"fails": 0, "until": 0.0})
    calls = []
    PNG = bytes([0x89, 0x50, 0x4E, 0x47, 13, 10, 0x1A, 10]) + b'TUILE-AMONT'

    def fake_fetch(rel_path):
        calls.append(rel_path)
        return PNG

    monkeypatch.setattr(mapserve, "_fetch_upstream", fake_fetch)
    try:
        z = 17  # jamais couvert par la maintenance (max_z 16 par defaut)
        x, y = _tile_of_cell(1054, 6882, z)
        resp = asyncio.run(mapserve.tile("aspect", z, x, f"{y}@2x.webp"))
        assert calls == [f"tiles/aspect/{z}/{x}/{y}@2x.webp"]
        assert resp.body == PNG
        assert resp.headers.get("X-Tile-Pending") != "1"  # servie, plus en attente
        # Mise en cache : la seconde demande ne repasse pas par l'amont
        asyncio.run(mapserve.tile("aspect", z, x, f"{y}@2x.webp"))
        assert len(calls) == 1
        assert tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "webp").is_file()
    finally:
        monkeypatch.delenv("LIDAR_TILE_CACHE_ONLY")


def test_tile_selection_always_answers(tmp_path, monkeypatch):
    """La sélection d'une dalle répond toujours, sans appel réseau : dalle
    absente comprise, avec nom, emprise L93 et cadre de sélection."""
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    def boom(*a, **k):
        raise AssertionError("aucun appel IGN dans /api/map/tile")
    monkeypatch.setattr("lidar_pipeline.fetch_ign.tile_ign_metadata", boom)
    d = mapserve.map_tile_info(lat=48.0, lng=2.0)             # loin de toute dalle
    assert d["found"] is False and d["name"].startswith("LHD_FXX_")
    assert len(d["corners"]) == 4 and d["l93_bbox"][2] - d["l93_bbox"][0] == 1000


def test_ign_metadata_cached_and_resilient(tmp_path, monkeypatch):
    """Fiche IGN : un seul appel au catalogue (cache disque) ; catalogue
    injoignable = message, jamais d'exception."""
    mapserve = _setup(tmp_path, monkeypatch)
    calls = []
    def fake(col, row, timeout=10):
        calls.append((col, row))
        return {"found": True, "download_url": "https://data.geopf.fr/x.copc.laz",
                "acquisition_start": "2022-03-05T16:20:51Z"}
    monkeypatch.setattr("lidar_pipeline.fetch_ign.tile_ign_metadata", fake)
    a = mapserve.map_tile_ign(col=1004, row=6832)
    b = mapserve.map_tile_ign(col=1004, row=6832)
    assert a["download_url"].endswith(".copc.laz") and b == a and len(calls) == 1
    def down(col, row, timeout=10):
        raise OSError("réseau coupé")
    monkeypatch.setattr("lidar_pipeline.fetch_ign.tile_ign_metadata", down)
    assert mapserve.map_tile_ign(col=1, row=2)["found"] is False


def test_ui_tile_card_and_wind_rose():
    from lidar_pipeline.mapui import _MAP_HTML, _MAP_JS
    assert 'id="rose"' in _MAP_HTML and 'id="roseCanvas"' in _MAP_HTML
    assert "api/map/ign?col=" in _MAP_JS and "Télécharger le nuage LiDAR" in _MAP_JS
    assert "compass + 90" in _MAP_JS          # même repère que le rendu (aspect = boussole + 90°)


def test_fresh_tiles_jump_the_maintenance_queue(monkeypatch):
    """Tuiles d'une dalle tout juste rendue : en tête de file, devant l'arriéré."""
    import lidar_pipeline.mapserve as mapserve
    monkeypatch.setitem(mapserve._bg, "queue", __import__("collections").deque())
    monkeypatch.setitem(mapserve._bg, "queued", set())
    backlog = [("relief_oriente", 16, i, 0) for i in range(5)]
    mapserve._bg_enqueue(backlog)
    fresh = [("relief_oriente", 12, 9, 9), ("relief_oriente", 16, 3, 0)]
    mapserve._bg_enqueue(fresh, front=True)
    q = list(mapserve._bg["queue"])
    assert q[:2] == fresh and len(q) == 6 and q.count(("relief_oriente", 16, 3, 0)) == 1


def test_cache_only_renders_unstored_levels(tmp_path, monkeypatch):
    """Cache seule (Pi) : un niveau non stocké est rendu à la volée — jamais
    une tuile transparente « en attente » qui n'arriverait jamais."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", True)
    monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 16)
    _setup(tmp_path, monkeypatch)
    monkeypatch.setenv("LIDAR_TILE_CACHE_ONLY", "1")
    z = 17
    x, y = _tile_of_cell(1054, 6882, z)
    resp = asyncio.run(mapserve.tile("aspect", z, x, f"{y}@2x.avif"))
    assert resp.headers.get("X-Tile-Pending") != "1" and resp.headers["X-Tile-Empty"] == "0"
    assert not tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "avif").exists()


def test_background_scan_only_queues_stored_levels(tmp_path, monkeypatch):
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", True)
    monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 16)
    _setup(tmp_path, monkeypatch)
    monkeypatch.setitem(mapserve._bg, "queue", __import__("collections").deque())
    monkeypatch.setitem(mapserve._bg, "queued", set())
    monkeypatch.setitem(mapserve._bg, "snapshot", None)
    mapserve.background_scan(tmp_path)
    zs = {z for (_l, z, _x, _y) in mapserve._bg["queue"]}
    assert zs and all(tiles.zoom_cached(z, 2) for z in zs)
    assert zs <= {5, 7, 9, 11, 13, 15}


def test_remote_fed_map_keeps_sources_locally(tmp_path, monkeypatch):
    """Carte alimentée par l'amont : les sources nécessaires sont rapatriées
    et gardées sans attendre de visite ; une source manquée (amont éteint)
    est reprise au scan suivant ; la dalle entière (doublon) n'est pas prise."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    S = tiles._Source
    thumb = S(tmp_path / "t.webp", (0, 0, 1000, 1000), 3.9, url="http://amont/t")
    quads = [S(tmp_path / f"q{i}.avif", (0, 0, 500, 500), 0.2, url=f"http://amont/q{i}") for i in range(4)]
    full = S(tmp_path / "d.avif", (0, 0, 1000, 1000), 0.2, url="http://amont/d")
    assert tiles.sources_to_keep([[thumb], quads, [full]]) == [thumb] + quads
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont")
    monkeypatch.setattr(mapserve, "OUTPUT_DIR", tmp_path)
    monkeypatch.setattr(tiles, "source_index", lambda *a, **k: {"relief_oriente": {(1, 2): [[thumb], quads, [full]]}})
    for key, val in (("queue", __import__("collections").deque()), ("queued", set()),
                     ("snapshot", None), ("prefetch", __import__("collections").deque())):
        monkeypatch.setitem(mapserve._bg, key, val)
    online = {"up": False}
    def fake_fetch(url, dest):
        if not online["up"]:
            return False
        dest.write_bytes(b"x"); return True
    monkeypatch.setattr(tiles, "_fetch_source", fake_fetch)
    mapserve.background_scan(tmp_path)
    while mapserve._bg["prefetch"]:
        mapserve._bg_process_one()
    assert not any(s_.path.exists() for s_ in [thumb] + quads)      # amont éteint
    online["up"] = True
    mapserve.background_scan(tmp_path)                               # dalle inchangée : reprise quand même
    while mapserve._bg["prefetch"]:
        mapserve._bg_process_one()
    assert all(s_.path.exists() for s_ in [thumb] + quads) and not full.path.exists()


def _bg_reset(monkeypatch, mapserve):
    import collections
    for key, val in (("queue", collections.deque()), ("queued", set()),
                     ("snapshot", None), ("prefetch", collections.deque()),
                     ("stats", {"rendues": 0, "a_jour": 0, "vides": 0, "scans": 0,
                                "dernier_scan": None, "ajouts_dernier_scan": 0})):
        monkeypatch.setitem(mapserve._bg, key, val)


def test_background_downloads_tiles_from_upstream(tmp_path, monkeypatch):
    """Maintenance de fond + LIDAR_MAPS_URL : les tuiles de la pyramide sont
    TÉLÉCHARGÉES en tâche de fond (pas seulement à l'affichage), sans rendu
    local — la carte se remplit même sans visite et reste servie ensuite
    sans l'amont."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    _setup(tmp_path, monkeypatch)
    _bg_reset(monkeypatch, mapserve)
    monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
    AVIF = b"\x00\x00\x00\x20ftypavif-TUILE-AMONT"
    calls = []
    monkeypatch.setattr(mapserve, "_fetch_upstream",
                        lambda rel: calls.append(rel) or AVIF)

    def no_render(*a, **k):
        raise AssertionError("rendu local inutile : la tuile vient de l'amont")

    monkeypatch.setattr(tiles, "get_tile", no_render)
    z = 13
    x, y = _tile_of_cell(1054, 6882, z)
    mapserve._bg_enqueue([("aspect", z, x, y)])
    assert mapserve._bg_process_one() == "telechargee"
    assert calls == [f"tiles/aspect/{z}/{x}/{y}@2x.avif"]
    cache = tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "avif")
    assert cache.read_bytes() == AVIF
    assert mapserve._bg["stats"]["telechargees"] == 1
    # Fraîche : servie en cache seule sans amont ni rendu
    data, state = tiles.cached_tile(tmp_path, "aspect", z, x, y, 2, "avif")
    assert state == "fresh" and data == AVIF


def test_background_downloads_without_pause(monkeypatch):
    """Tuiles téléchargées de l'amont enchaînées sans pause ; rendu local
    et rapatriement de source gardent la pause."""
    import lidar_pipeline.mapserve as mapserve
    results = iter(["telechargee", "telechargee", True])
    waits = []

    class Stop:
        def is_set(self):
            return len(waits) >= 1

        def wait(self, delay):
            waits.append(delay)

    monkeypatch.setattr(mapserve, "_bg", {**mapserve._bg, "stop": Stop()})
    monkeypatch.setattr(mapserve, "_bg_process_one", lambda: next(results))
    monkeypatch.setenv("LIDAR_TILE_BACKGROUND_PAUSE", "0.2")
    monkeypatch.setattr(mapserve.os, "nice", lambda n: 0)
    mapserve._bg_worker()
    assert waits == [0.2]


def test_background_renders_locally_when_upstream_down(tmp_path, monkeypatch):
    """Amont éteint : la maintenance rend la tuile localement (autonomie)."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    _setup(tmp_path, monkeypatch)
    _bg_reset(monkeypatch, mapserve)
    monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
    monkeypatch.setattr(mapserve, "_fetch_upstream", lambda rel: None)
    z = 13
    x, y = _tile_of_cell(1054, 6882, z)
    mapserve._bg_enqueue([("aspect", z, x, y)])
    assert mapserve._bg_process_one() is True
    assert tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "avif").is_file()
    assert mapserve._bg["stats"]["rendues"] == 1


def test_upstream_breaker_rearms_after_expiry(monkeypatch):
    """La suspension expirée, de nouveaux échecs la relancent (sinon chaque
    tuile repaie le délai réseau d'un amont éteint, indéfiniment)."""
    import lidar_pipeline.mapserve as mapserve
    monkeypatch.setattr(mapserve, "_UPSTREAM", {"fails": 0, "until": 0.0})
    for _ in range(3):
        mapserve._upstream_mark(False)
    assert mapserve._upstream_offline() is True
    mapserve._UPSTREAM["until"] = 1.0          # suspension expirée
    assert mapserve._upstream_offline() is False
    mapserve._upstream_mark(False)             # nouvel échec : resuspendu
    assert mapserve._upstream_offline() is True


def test_empty_tile_skips_upstream_when_inventory_is_upstream(tmp_path, monkeypatch):
    """Inventaire venu de l'amont (LIDAR_SOURCE_URL) et aucune dalle sous la
    tuile : l'amont n'en sait pas plus — pas de requête (20 s par tuile vide
    quand il est éteint)."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    _setup(tmp_path, monkeypatch)
    monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
    monkeypatch.setattr(tiles, "_contributing", lambda *a, **k: [])
    calls = []
    monkeypatch.setattr(mapserve, "_fetch_upstream", lambda rel: calls.append(rel))
    resp = asyncio.run(mapserve.tile("aspect", 15, 1, "1.png"))
    assert calls == []
    assert resp.headers["X-Tile-Empty"] == "1"


def test_cache_only_serves_stale_tile_while_pending(tmp_path, monkeypatch):
    """Cache seule, tuile périmée (dalle régénérée) et amont muet : l'ancienne
    tuile reste affichée en attendant la nouvelle — pas un trou transparent."""
    import os
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    _setup(tmp_path, monkeypatch)
    monkeypatch.setenv("LIDAR_TILE_CACHE_ONLY", "1")
    monkeypatch.setattr(mapserve, "MAPS_URL", "")
    z = 13
    x, y = _tile_of_cell(1054, 6882, z)
    cache = tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "avif")
    cache.parent.mkdir(parents=True, exist_ok=True)
    cache.write_bytes(b"ANCIENNE-TUILE")
    os.utime(cache, (1_000_000, 1_000_000))       # plus vieille que la dalle
    assert tiles.cached_tile(tmp_path, "aspect", z, x, y, 2, "avif")[1] == "pending"
    resp = asyncio.run(mapserve.tile("aspect", z, x, f"{y}@2x.avif"))
    assert resp.body == b"ANCIENNE-TUILE"
    assert resp.headers["Cache-Control"] == "no-store"   # redemandée plus tard
    assert resp.headers["X-Tile-Pending"] == "1"
    assert resp.headers["X-Tile-Empty"] == "0"


def test_background_tiles_before_sources(tmp_path, monkeypatch):
    """Tuiles d'abord (légères, visibles tout de suite), sources ensuite :
    des milliers de sources à rapatrier ne retardent pas la pyramide."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    _setup(tmp_path, monkeypatch)
    _bg_reset(monkeypatch, mapserve)
    monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
    monkeypatch.setattr(mapserve, "_fetch_upstream", lambda rel: b"TUILE")
    fetched = []

    class Src:
        def ensure(self):
            fetched.append(1)
            return True

    mapserve._bg["prefetch"].append(Src())
    z = 13
    x, y = _tile_of_cell(1054, 6882, z)
    mapserve._bg_enqueue([("aspect", z, x, y)])
    assert mapserve._bg_process_one() == "telechargee"
    assert fetched == [] and mapserve._bg["stats"]["telechargees"] == 1
    assert mapserve._bg_process_one() is True
    assert fetched == [1]


def test_zoom_animation_transforms_only_on_all_tile_layers():
    """Pendant l'animation de zoom, AUCUNE couche ne crée de niveau : un
    niveau neuf n'est pas composé par Chrome (animé sur le thread principal,
    gelé ~90 ms par le chargement) — le fond sautait quand le relief orienté,
    qui garde son niveau pair, glissait normalement : désynchronisés."""
    import re
    from lidar_pipeline.mapui import _MAP_JS
    assert "_animateZoom(e) { this._setView(e.center, e.zoom, true, true); }" in _MAP_JS
    # « zoom » part aussi au début de l'animation (Leaflet 1.9) : même garde
    assert "if (this._map && this._map._animatingZoom) {" in _MAP_JS
    # Sans quoi _setView recrée un niveau malgré noUpdate (tileZoomChanged)
    assert _MAP_JS.count("updateWhenZooming: false") >= 2
    # Le fond OSM ET les couches LiDAR héritent du comportement
    assert re.search(r"const base = new ZoomAnimTileLayer\(", _MAP_JS)
    assert "const EvenLevelTileLayer = ZoomAnimTileLayer.extend(" in _MAP_JS
    assert "L.tileLayer(" not in _MAP_JS.replace("L.tileLayer` standard", "")


def test_default_policy_pregenerates_every_level_up_to_native(monkeypatch):
    """Par défaut, toute la pyramide jusqu'au natif (18@2x = 0,2 m/px) est
    stockée et générée d'avance : aucun niveau rendu à la volée."""
    import importlib
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    for name in ("LIDAR_TILE_CACHE_MAX_Z", "LIDAR_TILE_EVEN_LEVELS",
                 "LIDAR_TILE_BACKGROUND_MAX_Z", "LIDAR_TILE_BACKGROUND_SCALE"):
        monkeypatch.delenv(name, raising=False)
    try:
        importlib.reload(tiles)
        assert all(tiles.zoom_cached(z, 2) for z in range(tiles.TILE_MIN_Z, 19))
        assert all(tiles.zoom_cached(z, 1) for z in range(tiles.TILE_MIN_Z, 20))
        assert mapserve._bg_conf()["max_z"] == 18 and mapserve._bg_conf()["scale"] == 2
        monkeypatch.setenv("LIDAR_TILE_BACKGROUND_SCALE", "1")
        assert mapserve._bg_conf()["max_z"] == 19
    finally:
        importlib.reload(tiles)


def test_background_scan_retries_cells_beyond_queue_max(tmp_path, monkeypatch):
    """File pleine : les dalles dont la pyramide n'a pas trouvé place sont
    reprises aux scans suivants jusqu'à ce que toutes leurs tuiles passent."""
    import collections
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", False)
    monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 19)
    _setup(tmp_path, monkeypatch)
    _make_dalle(tmp_path, 1055, 6882, ["aspect"])
    conf = dict(mapserve._bg_conf(), max_z=18, scale=2, queue_max=40)
    monkeypatch.setattr(mapserve, "_bg_conf", lambda: conf)
    for key, val in (("queue", collections.deque()), ("queued", set()),
                     ("snapshot", None), ("prefetch", collections.deque()),
                     ("incomplete", {})):
        monkeypatch.setitem(mapserve._bg, key, val)
    expected = set()
    for layer, per_cell in tiles.source_index(tmp_path, force=True).items():
        for col, row in per_cell:
            bbox = tiles.cell_bounds_wgs84(col, row)
            for z in range(tiles.TILE_MIN_Z, 19):
                expected.update((layer, z, x, y) for x, y in tiles.tiles_in_bounds(bbox, z))
    assert len(expected) > 3 * conf["queue_max"]
    seen = set()
    for _ in range(100):
        mapserve.background_scan(tmp_path)
        seen.update(mapserve._bg["queue"])
        mapserve._bg["queue"].clear()        # file vidée par les rendeurs
        mapserve._bg["queued"].clear()
        if not mapserve._bg["incomplete"]:
            break
    assert seen == expected
    # Pyramide passée en entier : les scans suivants se taisent
    assert mapserve.background_scan(tmp_path) == 0