"""Tests for the XYZ tile server and its discovery (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:
    """Minimal request: only the origin and the proxy headers matter."""

    def __init__(self, host="map.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": "map.local",
                                  "port": 8975})()


def _setup(tmp_path, monkeypatch, layers=("aspect", "slope"), panel=None):
    """Fake source tile; panel=None lifts the PANEL_VIZ restriction so that
    the test layers (aspect, slope) are served."""
    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):
    """The canonical tile is a 256 px RGBA PNG, with CORS headers."""
    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):
    """The @2x level serves 512 px WebP."""
    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):
    """Outside the data: transparent tile (not an error) for 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):
    """Versioned URL → immutable cache; no version → short revalidation."""
    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):
    """Tile name, layer, zoom and indices are validated (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"),      # invalid name
                                ("aspect", z, x, f"{y}.tif"),     # rejected format
                                ("unknown", z, x, f"{y}.png"),    # missing layer
                                ("../etc", z, x, f"{y}.png"),     # path traversal
                                ("aspect", 2, x, f"{y}.png"),     # zoom too low
                                ("aspect", 25, x, f"{y}.png"),    # zoom too high
                                ("aspect", z, 10 ** 9, f"{y}.png")):  # outside the grid
        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):
    """Tile missing locally: fetched from LIDAR_MAPS_URL and cached."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    _setup(tmp_path, monkeypatch)
    monkeypatch.setattr(mapserve, "MAPS_URL", "http://upstream: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)
    # Area without local data → local render fails, upstream takes over
    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 consecutive failures suspend upstream attempts; a success resets."""
    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: required fields, absolute XYZ URL, 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://map.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):
    """Layer missing from disk → 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):
    """Behind a TLS proxy, published URLs use 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):
    """JOSM imagery file: one TMS entry per layer, URL as {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):
    """WMTS capabilities: layers, GoogleMapsCompatible grid, REST template."""
    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` would first match the reference inside
    # TileMatrixSetLink (plain text): target the definition.
    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 describes the layers and the interface tiling contract."""
    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       # complete pyramid generated ahead of time
    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      # density missing from the fixture
    assert meta["default_mode"] == "relief"


def test_only_panel_layers_are_served(tmp_path, monkeypatch):
    """Only PANEL_VIZ layers are displayed and served; generation from the
    map produces only them."""
    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 identifies the tile under a point and reads its metadata."""
    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 outside the area: explicit answer, no error
    assert mapserve.map_tile_info(lat=45.0, lng=1.0)["found"] is False


def test_map_tile_includes_quality(tmp_path, monkeypatch):
    """/api/map/tile includes quality (density, acquisition dates) when known."""
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    from lidar_pipeline.quality import write_quality, QUALITY_VERSION
    name = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
    write_quality(tmp_path, name, {"version": QUALITY_VERSION, "ground_density": 12.5,
                                   "density_grid": [[12.5]], "empty_fraction": 0.1,
                                   "acq_start": "2021-03-02", "acq_end": "2021-03-02",
                                   "acq_source": "gps"})
    lat, lon = _center()
    d = mapserve.map_tile_info(lat=lat, lng=lon)
    assert d["quality"] == {"ground_density": 12.5, "acq_start": "2021-03-02",
                            "acq_end": "2021-03-02"}


def test_map_tile_quality_absent_is_none(tmp_path, monkeypatch):
    """Without a quality sidecar, the key is None (no error)."""
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    lat, lon = _center()
    assert mapserve.map_tile_info(lat=lat, lng=lon)["quality"] is None


def test_map_tile_quality_uses_real_basename(tmp_path, monkeypatch):
    """Quality is read under the real name of the rendered tile
    (best["basename"]), not the nominal name computed from col/row — they
    may differ."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import index as index_mod
    from lidar_pipeline.quality import write_quality, QUALITY_VERSION
    _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    nominal = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
    real = nominal + "_r0p2"
    write_quality(tmp_path, real, {"version": QUALITY_VERSION, "ground_density": 42.0,
                                   "density_grid": [[42.0]], "empty_fraction": 0.0,
                                   "acq_start": "2023-01-01", "acq_end": "2023-01-01",
                                   "acq_source": "gps"})
    fake_tile = {"col": 1054, "row": 6882, "resolution": 0.2, "basename": real, "viz": {}}
    monkeypatch.setattr(index_mod, "scan_tiles", lambda d: [fake_tile])
    monkeypatch.setattr(index_mod, "_collect_tile_metadata",
                        lambda t, dtm_dir: {"method": None, "generated": None, "viz": {}})
    lat, lon = _center()
    d = mapserve.map_tile_info(lat=lat, lng=lon)
    assert d["found"] is True
    assert d["name"] == real
    assert d["quality"]["ground_density"] == 42.0


def test_healthz_and_assets(tmp_path, monkeypatch):
    """Health probe and interface assets (revalidated, never immutable)."""
    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("missing.js")


def test_ui_main_layer_and_precision_modes():
    """A main display layer and precision: relief only, precision only, or a
    comparison on either side of a sliding bar.

    No more reorderable stack or per-layer blending. Quick keyboard toggle
    (P key).
    """
    from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
    for mode in ('data-mode="relief"', 'data-mode="precision"', 'data-mode="compare"'):
        assert mode in _MAP_HTML
    assert 'id="precLegend"' in _MAP_HTML
    assert 'id="mainSel"' in _MAP_HTML
    assert "createPane('lidarStack')" in _MAP_JS
    assert "e.key === 'p'" in _MAP_JS
    # Legend in exactly 16 steps (log density scale)
    assert "const PREC_LEVELS = 16;" in _MAP_JS and "repeat(16, 1fr)" in _MAP_CSS
    # Stack and blending removed
    for gone in ("function moveLayer(", "reorderRelative", "stackBlend", "addEventListener('dragstart'", "blendOptions"):
        assert gone not in _MAP_JS, gone
    assert "Haut de pile" not in _MAP_HTML and "Top of stack" not in _MAP_HTML
    # Shared link: main layer, mode, base map
    assert "'&M=' + STATE.main" in _MAP_JS
    assert "'&P=' + STATE.mode" in _MAP_JS


def test_ui_relief_intensity():
    """Relief intensity (contrast) slider: remembered, shared in the link
    (&I=, written only if ≠ 1), hidden in precision mode."""
    from lidar_pipeline.mapui import _MAP_HTML, _MAP_JS
    assert 'id="reliefIntensity"' in _MAP_HTML
    assert 'min="0.5"' in _MAP_HTML and 'max="2"' in _MAP_HTML and 'step="0.1"' in _MAP_HTML
    assert "function clampIntensity(" in _MAP_JS
    assert "'contrast(' +" in _MAP_JS
    assert "'&I=' +" in _MAP_JS and "seg.I" in _MAP_JS


def test_map_ui_tile_load_indicator_and_selection_pane():
    """Loading: remaining-tiles counter + visible tile outline.

    The counter reads the real tile state (a tile removed while loading
    emits no end event). The selection outline lives in a pane above the
    LiDAR stack (z-index 400 and above), otherwise the tiles hide it.
    """
    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_ui_tile_tab_replaces_popup_and_details_view():
    """A click on the map selects the tile (outline) and fills the Tile tab:
    no more Leaflet popup or separate "Details" view."""
    from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
    assert 'id="tilecard"' not in _MAP_HTML
    assert 'data-tab="tile"' in _MAP_HTML
    for id_ in ("tab-tile", "tileHint", "tileCard", "tileTitle", "tileBody", "tileTabLbl"):
        assert f'id="{id_}"' in _MAP_HTML, id_
    for gone in ("tileView", "tileBack", "showTilePopup", "openTileDetails",
                 "closeTileViews", "hideTileView", "restoreTabAfterDetails",
                 "RESTORING_TAB", "tileReturnTab", "L.popup("):
        assert gone not in _MAP_JS, gone
    assert "leaflet-popup" not in _MAP_CSS and "tile-pop" not in _MAP_CSS
    for fn in ("function selectTile(", "function clearTileSelection(", "function renderTabTile("):
        assert fn in _MAP_JS, fn
    # No selection while dragging the export frame or drawing an area
    click = _MAP_JS[_MAP_JS.index("map.on('click'"):]
    assert "genDrawing" in click[:400] and "PRINT.dragging" in click[:400]
    # Clicking the selected tile again: deselect rather than toggle
    assert "clearTileSelection();" in click[click.index("SELECTED_TILE"):click.index("});")]


def test_defaults_roundtrip(tmp_path, monkeypatch):
    """The display set from the map becomes the configuration served to everyone."""
    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")

    # Nothing saved: registry settings (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 "default_precision_opacity" not in meta

    req = mapserve.DefaultsRequest(main="aspect", mode="compare",
                                   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"] == "compare"
    assert meta["default_base"] == {"on": True, "opacity": 0.4, "dark": False}
    # Persisted on disk: survives a container restart
    assert (tmp_path / ".map-defaults.json").is_file()
    assert mapserve.get_defaults()["defaults"]["mode"] == "compare"

    # Removal: back to the registry settings
    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):
    """File from the former stack (order/on/blend) or vanished layer:
    registry main layer, relief mode; the base map is kept."""
    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": "voodoo"}), 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):
    """Unknown layer or precision rejected as main layer, mode validated, opacities clamped."""
    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="nonexistent", mode="voodoo",
        base={"opacity": -3}))["defaults"]
    assert data["main"] is None
    assert data["mode"] == "relief"
    assert "precision_opacity" not in data
    assert data["base"]["opacity"] == 0.0            # clamped to 0
    data = mapserve.set_defaults(mapserve.DefaultsRequest(main="densite_sol"))["defaults"]
    assert data["main"] is None                      # precision is not a main layer


def test_defaults_dark_none_is_not_a_choice(tmp_path, monkeypatch):
    """base.dark omitted/None is not a choice: it stays None (the base map
    follows the theme on the map side), unlike an explicit choice (True/False
    persist)."""
    import lidar_pipeline.mapserve as mapserve
    _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    monkeypatch.setattr(mapserve, "DEFAULTS_FILE", tmp_path / ".map-defaults.json")

    req = mapserve.DefaultsRequest(main="relief_oriente",
                                   base={"on": True, "opacity": 0.4})
    saved = mapserve.set_defaults(req)["defaults"]
    assert saved["base"]["dark"] is None
    assert saved["v"] == 3
    assert mapserve.map_meta()["default_base"]["dark"] is None

    # Explicit choice: kept as is
    req = mapserve.DefaultsRequest(main="relief_oriente",
                                   base={"on": True, "opacity": 0.4, "dark": True})
    mapserve.set_defaults(req)
    assert mapserve.map_meta()["default_base"]["dark"] is True


def test_defaults_legacy_dark_true_becomes_unset(tmp_path, monkeypatch):
    """Pre-v3 file (former webapp, base.dark:true always): served with dark
    None ("not chosen"), never imposed as the user's choice."""
    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({
        "main": "relief_oriente", "mode": "relief",
        "base": {"on": True, "opacity": 0.85, "dark": True}}), encoding="utf-8")
    meta = mapserve.map_meta()
    assert meta["default_base"]["dark"] is None
    assert meta["default_base"]["opacity"] == 0.85


def test_view_modes_compare_and_legacy_both(tmp_path, monkeypatch):
    """The former "both" mode is gone from VIEW_MODES in favour of
    "compare"; a file or request still sending it is read as "relief"
    (precision_opacity accepted for compatibility, but ignored)."""
    import lidar_pipeline.mapserve as mapserve
    _setup(tmp_path, monkeypatch, layers=("relief_oriente", "densite_sol"))
    from lidar_pipeline.index import VIEW_MODES
    assert VIEW_MODES == ('relief', 'precision', 'compare')
    monkeypatch.setattr(mapserve, "DEFAULTS_FILE", tmp_path / ".map-defaults.json")
    # Old frozen "both" default: read as relief
    mapserve.DEFAULTS_FILE.parent.mkdir(parents=True, exist_ok=True)
    mapserve.DEFAULTS_FILE.write_text('{"mode": "both", "precision_opacity": 0.6}', encoding="utf-8")
    meta = mapserve.map_meta()
    assert meta["default_mode"] == "relief"
    assert "default_precision_opacity" not in meta
    # Old page still open: precision_opacity accepted, ignored
    req = mapserve.DefaultsRequest(main=None, mode="both", precision_opacity=0.4, base={})
    assert mapserve._sanitize_defaults(req)["mode"] == "relief"
    assert "precision_opacity" not in mapserve._sanitize_defaults(req)
    req = mapserve.DefaultsRequest(main=None, mode="compare", base={})
    assert mapserve._sanitize_defaults(req)["mode"] == "compare"


def test_meta_serves_reading_text(tmp_path, monkeypatch):
    """/api/map/meta serves the "How to read" text of each layer."""
    import lidar_pipeline.mapserve as mapserve
    _setup(tmp_path, monkeypatch, layers=("relief_oriente", "densite_sol"))
    layers = {l["key"]: l for l in mapserve.map_meta()["layers"]}
    reading = layers["relief_oriente"]["reading"]
    assert isinstance(reading, list) and len(reading) >= 3
    text = " ".join(reading)
    assert "no ground point" in text and "1.5" in text and "1 m" in text
    assert layers["densite_sol"]["reading"]


def test_ui_reading_legend_from_meta():
    """The "How to read" legend: sentences from META.layers[].reading, never
    hard-coded in the interface (VIZ_LEGENDS stays the single source)."""
    from lidar_pipeline.mapui import _MAP_HTML, _MAP_JS
    assert 'id="readingBlock"' in _MAP_HTML and 'id="readingList"' in _MAP_HTML
    assert "<details" in _MAP_HTML and "How to read the map" in _MAP_HTML
    assert "function renderReading(" in _MAP_JS and ".reading" in _MAP_JS
    assert "no ground point" not in _MAP_JS and "no ground point" not in _MAP_HTML


def test_ui_compare_mode_replaces_both():
    """The "Compare" mode: a sliding bar separates two layers (relief, precision
    or OSM alone); the former "both" (multiplied) mode is gone and an old
    P=both link opens in relief."""
    from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
    assert 'data-mode="compare"' in _MAP_HTML and 'data-mode="both"' not in _MAP_HTML
    assert 'id="precOpacity"' not in _MAP_HTML and "precOpacity" not in _MAP_JS
    assert "mixBlendMode" not in _MAP_JS and "multiply" not in _MAP_JS
    for id_ in ("compareBar", "compareLeft", "compareRight", "compareCtl"):
        assert f'id="{id_}"' in _MAP_HTML, id_
    assert "function applyCompareClip(" in _MAP_JS and "clipPath" in _MAP_JS
    assert "map.on('move', applyCompareClip)" in _MAP_JS
    assert "'&C=' +" in _MAP_JS
    # Old P=both link: relief mode
    parse = _MAP_JS[_MAP_JS.index("function parseHash("):_MAP_JS.index("function viewHash(")]
    assert "'both'" in parse and "'relief'" in parse
    assert "#compareBar" in _MAP_CSS and "touch-action: none" in _MAP_CSS

    # The handle stays out of the panel and the tool column: the displayed
    # split is clamped (bar and clip together), including while dragging, and
    # recomputed when the panel changes width.
    assert "function compareBounds(" in _MAP_JS
    bounds = _MAP_JS[_MAP_JS.index("function compareBounds("):_MAP_JS.index("function applyCompareClip(")]
    assert "el('panel').getBoundingClientRect()" in bounds and "el('tools').getBoundingClientRect()" in bounds
    clip = _MAP_JS[_MAP_JS.index("function applyCompareClip("):_MAP_JS.index("(function compareDrag(")]
    assert "compareBounds()" in clip
    drag = _MAP_JS[_MAP_JS.index("(function compareDrag("):]
    assert "compareBounds()" in drag[:drag.index("})();")]
    assert "document.addEventListener('panel:tab', applyCompareClip)" in _MAP_JS
    # Handle centred on the map area visible above the sheet (phone)
    assert "top: calc((100% - var(--sheet-h)) / 2)" in _MAP_CSS


def test_ui_applies_server_defaults():
    """The interface starts from the server defaults and can (re)define them."""
    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 "default_precision_opacity" not in _MAP_JS
    assert "precision_opacity" not in _MAP_JS
    # Reset forgets the local state before taking the server's
    assert "lsDel(LS_KEY)" in _MAP_JS


def test_ui_base_dark_follows_theme_unless_chosen():
    """base.dark is never an implicit choice: None = follow the theme.

    Client: defaultState keeps None (instead of forcing a boolean); loadState
    ignores the base.dark of an old lidarMapView_v2 (before v3, when the
    former webapp persisted dark:true on every save); saveState writes v: 3.
    The OSM base CSS class starts from the current theme, not a frozen
    "base-dark"."""
    from lidar_pipeline.mapui import _MAP_JS
    assert "dark: b.dark == null ? null : !!b.dark" in _MAP_JS
    assert "if (s.v !== 3) delete sBase.dark;" in _MAP_JS
    assert "lsSet(LS_KEY, Object.assign({ v: 3 }, STATE))" in _MAP_JS
    assert "className: themeIsDark() ? 'base-dark' : ''" in _MAP_JS


def test_ui_escape_deselects_tile_before_collapsing_panel():
    """Escape deselects the tile (if any is selected) before collapsing the
    panel; an area being drawn is cancelled first, as before."""
    from lidar_pipeline.mapui import _MAP_JS
    keydown = _MAP_JS[_MAP_JS.index("document.addEventListener('keydown'"):]
    esc = keydown[keydown.rindex("if (e.key === 'Escape')"):]
    assert "genSetDrawing(false)" in esc[:200]
    assert "clearTileSelection()" in esc[:400]
    assert "setSheet('closed')" in esc[:600] and "setCollapsed(true)" in esc[:600]


def test_ui_saved_print_tab_does_not_rezoom_shared_link():
    """The remembered Export tab does not reframe the view on the very first
    panel render (otherwise it overrides the zoom of a shared link or of the
    initial URL): BOOTSTRAPPING becomes false right after THAT first render —
    on success AND failure of the /api/map/meta load (otherwise a server
    unreachable at startup keeps BOOTSTRAPPING true forever: no reframing
    ever again, even after a later explicit opening of the tab)."""
    from lidar_pipeline.mapui import _MAP_JS
    assert "let BOOTSTRAPPING = true;" in _MAP_JS
    assert "printSetOpen(e.detail.visible === 'print', !BOOTSTRAPPING);" in _MAP_JS
    then_branch = _MAP_JS[_MAP_JS.index("panelRender();\n  BOOTSTRAPPING = false;"):]
    assert then_branch.startswith("panelRender();\n  BOOTSTRAPPING = false;")
    catch_branch = _MAP_JS[_MAP_JS.index(".catch(() => {\n  toast('Tile server unreachable."):]
    assert "BOOTSTRAPPING = false;" in catch_branch[:400]


def test_ui_tile_selection_does_not_affect_print_frame():
    """Tile selection lives apart from the tabs: it neither closes nor reopens
    the export frame (unlike the former "Details" card, which hid the other
    tabs on top of the map)."""
    from lidar_pipeline.mapui import _MAP_JS
    select_fn = _MAP_JS[_MAP_JS.index("function selectTile("):]
    assert "printSetOpen" not in select_fn[:select_fn.index("function clearTileSelection(")]
    open_fn = _MAP_JS[_MAP_JS.index("function printSetOpen("):]
    assert "explicit && !map.getBounds().contains(PRINT.center)" in open_fn[:700]
    assert "printRefreshFrame(!!explicit);" in open_fn[:700]


def test_ui_colors_are_tokenised():
    """No more hard-coded colour for the selection outline, the GPS marker and
    the generation log (AGENTS.md "no hard-coded colour in components");
    #1b1b1b stays hard-coded on the export handle (fixed contrast wanted on
    --print), documented in a comment."""
    from lidar_pipeline.mapui import _MAP_CSS, _MAP_JS
    assert "color: cssVar('--tile-sel')" in _MAP_JS
    assert "'#ffd24a'" not in _MAP_JS
    assert "cssVar('--gps')" in _MAP_JS
    assert "'#39d98a'" not in _MAP_JS
    assert "--tile-sel:" in _MAP_CSS and "--gps:" in _MAP_CSS
    assert ".gen-log { font:" in _MAP_CSS
    gen_log = _MAP_CSS[_MAP_CSS.index(".gen-log {"):]
    assert "rgba(0,0,0,0.35)" not in gen_log[:250]
    assert "var(--surface-2)" in gen_log[:250] and "var(--text)" in gen_log[:250]
    assert "#1b1b1b" in _MAP_CSS   # intentional, commented just above


def test_ui_guards_corrupt_local_storage():
    """A corrupt value in lidar-panel/lidar-print (another app on the same
    origin, quota exceeded while writing…) does not break the interface:
    type/value whitelist rather than direct use of lsGet."""
    from lidar_pipeline.mapui import _MAP_JS
    assert "function _sanitizedPanelLS()" in _MAP_JS
    panel_ls = _MAP_JS[_MAP_JS.index("function _sanitizedPanelLS("):]
    assert "typeof s.tab === 'string'" in panel_ls[:700]
    assert "typeof s.collapsed === 'boolean'" in panel_ls[:700]
    assert "['closed', 'half', 'full'].includes(s.sheet)" in panel_ls[:700]
    print_load = _MAP_JS[_MAP_JS.index("function printLoad("):]
    assert "raw && typeof raw === 'object' && !Array.isArray(raw)" in print_load[:400]


def test_ui_uses_standard_tilelayer():
    """The interface relies on Leaflet's native LOD, not on a home-made level scheme."""
    from lidar_pipeline.mapui import _MAP_JS, render_html, ui_version
    assert "L.TileLayer.extend(" in _MAP_JS      # Leaflet TileLayer, native LOD
    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
    # The odd native level (19 = 0.2 m/px) is requested as is, never folded
    # onto the even level below (display capped at 0.4 m/px).
    assert "z % 2 && (maxN === undefined || z < maxN)" in _MAP_JS
    # None of the old map machinery left
    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):
    """Written assets carry the version and embed the vendored Leaflet."""
    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 starts the pre-render in the background, then reports."""
    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):
    """Per-tile lock: shared during rendering, removed once released.

    Without a counter, the registry grows forever (millions of tiles) or a
    lock disappears while another thread waits on it — two concurrent
    renders of the same tile.
    """
    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):
    """Two simultaneous requests for the same tile render it only once."""
    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)
    # Later calls read the disk cache: a single effective render
    assert len(calls) == 4          # every request calls get_tile…
    assert mapserve._locks == {}    # …but serialized, and the locks released


def test_tile_bytes_survives_render_error(tmp_path, monkeypatch):
    """A failing render returns an empty tile without leaking a lock."""
    import lidar_pipeline.mapserve as mapserve
    _setup(tmp_path, monkeypatch)

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

    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):
    """The concurrent-render cap holds (small server protected).

    Without it, as many renders as in-flight requests: each one decodes a
    source tile of several tens of MB — a Raspberry Pi's memory and CPU are
    saturated.
    """
    import threading
    import lidar_pipeline.mapserve as mapserve
    _setup(tmp_path, monkeypatch)

    # Deliberately low cap for the 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)
    # DIFFERENT tiles: the per-tile lock does not hide the semaphore
    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):
    """The caps come from the environment, with a modest default."""
    import importlib
    import lidar_pipeline.mapserve as mapserve
    import lidar_pipeline.tiles as tiles

    # Defaults: 2 renders, 2 downloads — sized for a small server
    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
        # Absurd value: never zero threads (would block the service entirely)
        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):
    """Cache-only mode: missing tile = transparent, not cacheable."""
    from lidar_pipeline import tiles as _t
    monkeypatch.setattr(_t, "TILE_EVEN_LEVELS", False)   # legacy rule: everything is stored
    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()
    # Once rendered elsewhere: served from the cache, cacheable
    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):
    """The maintenance detects source tiles, fills the pyramid, then goes quiet."""
    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:
        # First scan: everything is new, low levels at the head of the queue
        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()
        # Nothing changed: no more additions
        assert mapserve.background_scan(tmp_path) == 0
        # A new source tile (worker): its pyramid alone is queued again
        _make_dalle(tmp_path, 1055, 6882, ["slope"])
        assert mapserve.background_scan(tmp_path) > 0
    finally:
        _reset_bg(mapserve)


# ---------------------------------------------------------------------------
# Tile generation (ported from the former webapp)
# ---------------------------------------------------------------------------

import types

from fastapi import HTTPException
import pytest


class _ClientRequest:
    """Minimal request with a connection IP (LIDAR_REGEN_CIDR restriction)."""

    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():
    """A WGS84 bbox inside a 1 km L93 cell selects that cell."""
    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
    # Cell (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):
    """The preview returns the cells to process with their GPS corners."""
    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=["unknown"])
    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):
    """Global pass without any tile in input/: empty list (nothing to complete).

    The 400 "no tile to complete" error only concerns the launch
    (/api/generate → _resolve_request): the preview stays informative.
    """
    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
    # Sorted north → south (decreasing row), then increasing col
    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 in progress: the request is queued (running work never interrupted)
    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 reports available (local pipeline) and 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 outside the local network: generation forbidden (tab hidden)
    d = mapserve.generation_status(_ClientRequest("203.0.113.7"))
    assert d["regen_allowed"] is False

    # X-Forwarded-For honoured only from a trusted peer
    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

    # Lightweight image without pipeline or LIDAR_GENERATION_URL: unavailable
    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):
    """Lightweight map: /api/status relays the worker state; 503 = offline."""
    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, "processing machine unreachable")

    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):
    """A client outside LIDAR_REGEN_CIDR cannot start a generation."""
    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 returns the tile under a point (manual selection)."""
    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 serves index_tiles.json (lightweight LIDAR_SOURCE_URL machines)."""
    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
    # Unchanged ?stamp → light response
    light = mapserve.tiles_inventory(stamp=d["stamp"])
    assert light["tiles"] is None


def test_generation_always_ign_with_edge_buffer(tmp_path, monkeypatch):
    """IGN classification (ground) and 100 m edge stitching are enforced,
    whatever the client sends."""
    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():
    """The interface embeds the generation controls (＋ Area, 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="tab-gen"' 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 and edge stitching enforced: no more settings
    for gone in ('id="genClass"', 'id="genIgnClasses"', 'id="genReclass"', 'id="genEdge"'):
        assert gone not in _MAP_HTML
    # Hidden by default: revealed only if /api/status allows it
    assert "GEN.available && GEN.allowed" in _MAP_JS
    # Area drawing + run outlines + status polling
    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
    # End of run: layer refresh + maintenance restart
    assert "genAfterRun" in _MAP_JS and "api/map/background" in _MAP_JS


def test_cache_only_fetches_missing_tile_upstream(tmp_path, monkeypatch):
    """Cache-only + LIDAR_MAPS_URL: a missing tile is fetched from the
    upstream server (no local render: zooms beyond the maintenance display).
    """
    from lidar_pipeline import tiles as _t
    monkeypatch.setattr(_t, "TILE_EVEN_LEVELS", False)   # legacy rule: everything is stored
    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://upstream:8975")
    monkeypatch.setattr(mapserve, "_UPSTREAM", {"fails": 0, "until": 0.0})
    calls = []
    PNG = bytes([0x89, 0x50, 0x4E, 0x47, 13, 10, 0x1A, 10]) + b'UPSTREAM-TILE'

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

    monkeypatch.setattr(mapserve, "_fetch_upstream", fake_fetch)
    try:
        z = 17  # stored level, not yet produced (no maintenance running in this test)
        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"  # served, no longer pending
        # Cached: the second request does not go upstream again
        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):
    """Tile selection always answers, without a network call: including a
    missing tile, with name, L93 extent and selection outline."""
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    def boom(*a, **k):
        raise AssertionError("no IGN call in /api/map/tile")
    monkeypatch.setattr("lidar_pipeline.fetch_ign.tile_ign_metadata", boom)
    d = mapserve.map_tile_info(lat=48.0, lng=2.0)             # far from any tile
    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):
    """IGN record: a single catalogue call (disk cache); unreachable
    catalogue = message, never an 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("network down")
    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="legendBlock"' in _MAP_HTML and 'id="roseCanvas"' in _MAP_HTML
    assert "api/map/ign?col=" in _MAP_JS and "Download the IGN LiDAR point cloud" in _MAP_JS
    assert "compass + 90" in _MAP_JS          # same frame as the rendering (aspect = compass + 90°)


def test_fresh_tiles_jump_the_maintenance_queue(monkeypatch):
    """Tiles of a freshly rendered source tile: at the head of the queue, ahead of the backlog."""
    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-only (Pi): a level that is not stored is rendered on the fly —
    never a transparent "pending" tile that would never arrive."""
    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):
    """Map fed by upstream: the needed sources are fetched and kept without
    waiting for a visit; a missed source (upstream down) is retried at the
    next scan; the full tile (duplicate) is not fetched."""
    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://upstream/t")
    quads = [S(tmp_path / f"q{i}.avif", (0, 0, 500, 500), 0.2, url=f"http://upstream/q{i}") for i in range(4)]
    full = S(tmp_path / "d.avif", (0, 0, 1000, 1000), 0.2, url="http://upstream/d")
    assert tiles.sources_to_keep([[thumb], quads, [full]]) == [thumb] + quads
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://upstream")
    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)      # upstream down
    online["up"] = True
    mapserve.background_scan(tmp_path)                               # unchanged tile: retried anyway
    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):
    """Background maintenance + LIDAR_MAPS_URL: the pyramid tiles are
    DOWNLOADED in the background (not only on display), without local
    rendering — the map fills up even without visits and is then served
    without upstream."""
    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://upstream:8975")
    AVIF = b"\x00\x00\x00\x20ftypavif-UPSTREAM-TILE"
    calls = []
    monkeypatch.setattr(mapserve, "_fetch_upstream",
                        lambda rel: calls.append(rel) or AVIF)

    def no_render(*a, **k):
        raise AssertionError("local render not needed: the tile comes from upstream")

    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
    # Fresh: served in cache-only mode without upstream or rendering
    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):
    """Tiles downloaded from upstream are chained without a pause; local
    rendering and source fetching keep the 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):
    """Upstream down: the maintenance renders the tile locally (autonomy)."""
    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://upstream: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):
    """Once the suspension has expired, new failures restart it (otherwise
    every tile pays the network timeout of a down upstream again, forever)."""
    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 expired
    assert mapserve._upstream_offline() is False
    mapserve._upstream_mark(False)             # new failure: suspended again
    assert mapserve._upstream_offline() is True


def test_empty_tile_skips_upstream_when_inventory_is_upstream(tmp_path, monkeypatch):
    """Inventory from upstream (LIDAR_SOURCE_URL) and no source tile under the
    tile: upstream knows no more — no request (20 s per empty tile when it
    is down)."""
    import lidar_pipeline.mapserve as mapserve
    from lidar_pipeline import tiles
    _setup(tmp_path, monkeypatch)
    monkeypatch.setattr(mapserve, "MAPS_URL", "http://upstream:8975")
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://upstream: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-only, stale tile (source tile regenerated) and silent upstream:
    the old tile stays displayed until the new one arrives — not a
    transparent hole."""
    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"OLD-TILE")
    os.utime(cache, (1_000_000, 1_000_000))       # older than the source tile
    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"OLD-TILE"
    assert resp.headers["Cache-Control"] == "no-store"   # requested again later
    assert resp.headers["X-Tile-Pending"] == "1"
    assert resp.headers["X-Tile-Empty"] == "0"


def test_background_tiles_before_sources(tmp_path, monkeypatch):
    """Tiles first (light, visible immediately), sources afterwards:
    thousands of sources to fetch do not delay the pyramid."""
    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://upstream:8975")
    monkeypatch.setattr(mapserve, "_fetch_upstream", lambda rel: b"TILE")
    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():
    """During the zoom animation, NO layer creates a level: a new level is not
    composited by Chrome (animated on the main thread, frozen ~90 ms by
    loading) — the base map jumped while the oriented relief, which kept its
    even level, slid normally: out of sync."""
    import re
    from lidar_pipeline.mapui import _MAP_JS
    assert "_animateZoom(e) { this._setView(e.center, e.zoom, true, true); }" in _MAP_JS
    # "zoom" also fires at the start of the animation (Leaflet 1.9): same guard
    assert "if (this._map && this._map._animatingZoom) {" in _MAP_JS
    # Otherwise _setView recreates a level despite noUpdate (tileZoomChanged)
    assert _MAP_JS.count("updateWhenZooming: false") >= 2
    # The OSM base AND the LiDAR layers inherit the behaviour
    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):
    """By default, the whole pyramid up to native (18@2x = 0.2 m/px) is
    stored and generated ahead of time: no level rendered on the fly."""
    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):
    """Full queue: source tiles whose pyramid did not fit are resumed at the
    next scans until all their tiles have been queued."""
    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()        # queue drained by the renderers
        mapserve._bg["queued"].clear()
        if not mapserve._bg["incomplete"]:
            break
    assert seen == expected
    # Whole pyramid queued: later scans stay quiet
    assert mapserve.background_scan(tmp_path) == 0


def _center(col=1054, row=6882):
    from lidar_pipeline.tiles import _transformer
    lon, lat = _transformer("EPSG:2154", "EPSG:4326").transform(col * 1000 + 500, (row - 1) * 1000 + 500)
    return lat, lon


def test_export_frame_returns_corners(tmp_path, monkeypatch):
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    lat, lon = _center()
    d = mapserve.export_frame(lat, lon, "A4", "paysage", 2000)
    assert len(d["corners"]) == 4 and d["width_m"] > 0


def test_export_pdf_ok(tmp_path, monkeypatch):
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    lat, lon = _center()
    resp = mapserve.export_pdf_file(lat, lon, "A4", "paysage", 2000, None)
    assert resp.media_type == "application/pdf" and resp.body.startswith(b"%PDF")
    assert 'attachment; filename="relief_' in resp.headers["Content-Disposition"]


def test_export_pdf_errors(tmp_path, monkeypatch):
    import pytest
    from fastapi import HTTPException
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    lat, lon = _center()
    with pytest.raises(HTTPException) as e:
        mapserve.export_pdf_file(lat, lon, "A4", "paysage", 1234, None)
    assert e.value.status_code == 400
    with pytest.raises(HTTPException) as e:
        mapserve.export_pdf_file(46.5, 2.5, "A4", "paysage", 2000, None)
    assert e.value.status_code == 404


def test_export_pdf_rejects_non_finite(tmp_path, monkeypatch):
    import pytest
    from fastapi import HTTPException
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    for lat, lon in ((float("nan"), 2.0), (48.0, float("inf"))):
        with pytest.raises(HTTPException) as e:
            mapserve.export_pdf_file(lat, lon, "A4", "paysage", 2000, None)
        assert e.value.status_code == 400
        with pytest.raises(HTTPException) as e:
            mapserve.export_frame(lat, lon, "A4", "paysage", 2000)
        assert e.value.status_code == 400


def test_export_pdf_concurrent_429(tmp_path, monkeypatch):
    import pytest
    from fastapi import HTTPException
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    lat, lon = _center()
    assert mapserve._export_lock.acquire(blocking=False)
    try:
        with pytest.raises(HTTPException) as e:
            mapserve.export_pdf_file(lat, lon, "A4", "paysage", 2000, None)
        assert e.value.status_code == 429
    finally:
        mapserve._export_lock.release()
    assert mapserve.export_pdf_file(lat, lon, "A4", "paysage", 2000, None).body.startswith(b"%PDF")


def test_export_pdf_never_delegated(tmp_path, monkeypatch):
    """Lightweight map with a remote generator: the export stays local."""
    mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
    monkeypatch.setattr(mapserve, "GENERATION_URL", "http://unreachable.invalid")
    lat, lon = _center()
    assert mapserve.export_pdf_file(lat, lon, "A4", "paysage", 2000, None).body.startswith(b"%PDF")


def test_map_ui_has_pdf_export():
    from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
    for id_ in ("tab-print", "printPaper", "printOrient", "printScale",
                "printTitle", "printGo", "printMsg", "printInfo"):
        assert f'id="{id_}"' in _MAP_HTML, id_
    for scale in ("1000", "2000", "5000", "10000"):
        assert f'value="{scale}"' in _MAP_HTML
    assert "api/export/frame?" in _MAP_JS and "api/export/pdf?" in _MAP_JS
    assert "createPane('printFrame')" in _MAP_JS
    # frame placed on the ground: it no longer follows the view, it is dragged
    assert "map.on('moveend', printRefreshFrame)" not in _MAP_JS
    assert "draggable: true" in _MAP_JS and "on('dragend', printDragEnd)" in _MAP_JS
    assert "PRINT.center" in _MAP_JS
    for id_ in ("printHere", "printFit"):
        assert f'id="{id_}"' in _MAP_HTML, id_
    # tab always offered (independent of /api/status.available); no more ⎙
    # button or card outside the panel
    assert 'data-tab="print"' in _MAP_HTML
    for gone in ('id="btnPrint"', 'id="printcard"', 'id="printClose"'):
        assert gone not in _MAP_HTML, gone
    tools = _MAP_HTML.split('<nav id="tools">')[1].split("</nav>")[0]
    assert "⎙" not in tools
    assert "e.detail.visible === 'print'" in _MAP_JS


def test_ui_tokens_themes_and_components():
    from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
    for var in ("--surface", "--on-accent", "--print", "--fs-sm", "--r-md", "--sp-2",
                "--shadow-lg", "--z-panel", "--z-toast"):
        assert var + ":" in _MAP_CSS, var
    # Dark palette: dark system theme (unless light forced) or dark forced
    assert '@media (prefers-color-scheme: dark)' in _MAP_CSS
    assert ':root:not([data-theme="light"])' in _MAP_CSS
    assert ':root[data-theme="dark"]' in _MAP_CSS
    for cls in (".btn-primary", ".btn-ghost", ".btn-icon", ".field", ".seg", ".hint"):
        assert cls in _MAP_CSS, cls
    assert 'id="btnTheme"' in _MAP_HTML
    assert "const THEME_LS = 'lidar-theme'" in _MAP_JS and "function applyTheme(" in _MAP_JS
    # Local storage always guarded (private browsing): no direct
    # localStorage access outside the lsGet/lsSet/lsDel block.
    import re
    start = _MAP_JS.index("function lsGet(")
    end = _MAP_JS.index("// --- end of storage")
    outside = _MAP_JS[:start] + _MAP_JS[end:]
    assert not re.search(r"localStorage\.(getItem|setItem|removeItem)\(", outside)
    # Dead code removed
    for gone in ("genClass", "gen-input", "genClassInfo", ".grip", "layer-row", "#gps.busy"):
        assert gone not in _MAP_CSS + _MAP_HTML, gone
    # No more interface colours hard-coded outside the palettes
    rules = _MAP_CSS[_MAP_CSS.rindex(":root[data-theme=\"dark\"]"):]
    rules = rules[rules.index("}") + 1:]
    for hard in ("#0b0e13", "#10141b", "#f4a261", "#4cc9f0", "#e76f51", "rgba(90, 160, 255"):
        assert hard not in rules, hard


def test_ui_single_tabbed_panel():
    from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
    for id_ in ("panel", "panelTabs", "panelBody", "panelCollapse", "sheetGrip",
                "tab-view", "tab-tile", "tab-print", "tab-gen", "tab-share"):
        assert f'id="{id_}"' in _MAP_HTML, id_
    for tab in ("view", "tile", "print", "gen", "share"):
        assert f'data-tab="{tab}"' in _MAP_HTML, tab
    # No more floating elements: everything goes into the panel
    for gone in ('id="layercard"', 'id="printcard"', 'id="gencard"', 'id="usecard"',
                 'id="gentools"', 'id="btnPrint"', 'id="rose"', 'id="btnUse"'):
        assert gone not in _MAP_HTML, gone
    # Settings and buttons moved, identifiers kept
    for id_ in ("printPaper", "printGo", "genGo", "btnZone", "btnComplete", "btnShare",
                "btnDefault", "btnReset", "useBody", "modeSeg", "roseCanvas", "baseOpacity"):
        assert f'id="{id_}"' in _MAP_HTML, id_
    assert "const PANEL_LS = 'lidar-panel'" in _MAP_JS
    for fn in ("function panelTabs(", "function openTab(", "function setSheet(",
               "function setCollapsed(", "function panelVisibleTab("):
        assert fn in _MAP_JS, fn
    assert "'panel:tab'" in _MAP_JS
    # Generation hidden: tab removed from the list (keys 1–5 included)
    assert "GEN.available && GEN.allowed" in _MAP_JS.split("function panelTabs(")[1].split("}")[0]
    assert "@media (max-width: 720px)" in _MAP_CSS and "#panel" in _MAP_CSS
    for state in ('data-sheet="closed"', 'data-sheet="half"', 'data-sheet="full"'):
        assert state.split('"')[1] in _MAP_CSS, state


def test_ui_panel_restores_tab_and_ignores_collapse_on_phone():
    from lidar_pipeline.mapui import _MAP_JS
    render = _MAP_JS.split("function panelRender(")[1].split("\nfunction ")[0]
    # Chosen tab kept aside: Generation restored once /api/status is known
    assert "PANEL.want = PANEL.tab;" in _MAP_JS
    assert "PANEL.tab = tabs.includes(PANEL.want) ? PANEL.want : 'view';" in render
    assert "tab: PANEL.want" in _MAP_JS.split("function panelSave(")[1].split("\n")[0]
    assert "PANEL.want = PANEL.tab = tab;" in _MAP_JS.split("function openTab(")[1].split("\nfunction ")[0]
    # Collapse to an icon strip: desktop only
    assert "PANEL.collapsed && !isPhone()" in render
    assert "toggle('collapsed', PANEL.collapsed)" not in render
    assert "toggle('panel-collapsed', PANEL.collapsed)" not in render


def test_ui_tile_tab_label_stays_dalle_col_row_in_title():
    """Final review (finding 2): 5 tabs in 340 px — the Tile tab label always
    stays "Tile" (never truncated), col · row goes into the tab title
    (hover) and #tileTitle (already in the card). Display/PDF export/
    Generation shortened to fit, Share kept."""
    from lidar_pipeline.mapui import _MAP_HTML, _MAP_JS
    assert '<span class="tab-lbl">View</span>' in _MAP_HTML
    assert '<span class="tab-lbl">PDF</span>' in _MAP_HTML
    assert '<span class="tab-lbl">Generate</span>' in _MAP_HTML
    assert '<span class="tab-lbl" id="tileTabLbl">Tile</span>' in _MAP_HTML
    assert 'data-tab="share"' in _MAP_HTML and '>Share</span>' in _MAP_HTML
    # The full title (hover) stays explicit despite the short label
    assert 'title="Display (1)"' in _MAP_HTML
    assert 'title="PDF export (3)"' in _MAP_HTML
    assert 'title="Generation (4)"' in _MAP_HTML
    assert "function tileTabTitle(" in _MAP_JS
    assert "textContent = tileTabLabel" not in _MAP_JS   # no more dynamic label
    assert "button[data-tab=\"tile\"]').title = tileTabTitle()" in _MAP_JS


def test_ui_tile_card_refetched_after_gen_run():
    """Finding 3: the card of a tile selected before/during a run becomes
    stale (resolution, date, layers...) — genAfterRun reloads it through
    api/map/tile without ever deselecting (no clearTileSelection)."""
    from lidar_pipeline.mapui import _MAP_JS
    fn = _MAP_JS[_MAP_JS.index("function genAfterRun("):_MAP_JS.index("function genUpdateStatus(")]
    assert "SELECTED_TILE" in fn and "lastTileQuery" in fn
    assert "api/map/tile?lat=" in fn
    assert "selectTile(d, lastTileQuery)" in fn
    assert "clearTileSelection" not in fn


def test_ui_double_click_and_stale_click_response_ignored():
    """M1/M2: a double click (Leaflet zoom, two single clicks, the second
    with detail > 1) must not toggle back the selection set by the first;
    a response arriving after a more recent click (sequence token) is
    ignored rather than overwriting the state."""
    from lidar_pipeline.mapui import _MAP_JS
    click = _MAP_JS[_MAP_JS.index("map.on('click'"):]
    body = click[:click.index("});") + 3]
    assert "e.originalEvent && e.originalEvent.detail > 1" in body
    assert "let tileClickSeq = 0;" in _MAP_JS
    assert "const seq = ++tileClickSeq;" in body
    assert "if (seq !== tileClickSeq) return;" in body


def test_ui_regen_button_appears_when_gen_status_resolves_late():
    """M3: a tile selected before /api/status answers must not stay without
    a Regenerate button — the card is re-rendered as soon as GEN is known,
    in the same block that sets GEN.available/allowed."""
    from lidar_pipeline.mapui import _MAP_JS
    boot = _MAP_JS[_MAP_JS.index("fetch('api/map/meta').then(r => r.json()).then(meta => {"):]
    status_block = boot[:boot.index("}).catch(() => {\n  toast('Tile server unreachable.'")]
    assert "GEN.available = !!d.available;" in status_block
    assert "if (SELECTED_TILE) renderTabTile();" in status_block


def test_ui_intensity_row_visible_outside_precision_block():
    """M4 (spec binding): #intensityRow is a sibling of #precBlock, not a
    child — visible in the Display tab even without a precision layer,
    hidden only in precision mode (renderPanel)."""
    from lidar_pipeline.mapui import _MAP_HTML
    assert '<div id="precBlock" hidden>' in _MAP_HTML
    # #intensityRow opens right after #precBlock closes (sibling, not
    # child): nothing of precBlock (modeSeg/compareCtl/precLegend) follows it.
    after_prec_close = _MAP_HTML[_MAP_HTML.index('<div class="prec-legend" id="precLegend"'):]
    close_idx = after_prec_close.index("</div>\n      </div>")
    between = after_prec_close[:close_idx]
    assert 'id="intensityRow"' not in between
    assert '<div class="row" id="intensityRow">' in after_prec_close[close_idx:close_idx + 200]


def test_ui_full_shared_link_without_intensity_resets_to_one():
    """M5: a full link (layer or mode in the hash) without &I means 1×
    (viewHash writes &I only if ≠ 1), never the intensity left in this
    browser's localStorage."""
    from lidar_pipeline.mapui import _MAP_JS
    load = _MAP_JS[_MAP_JS.index("function loadState("):_MAP_JS.index("function saveState(")]
    assert "else if (hash.main || hash.view) st.intensity = 1;" in load


def test_ui_compare_clip_skips_when_already_off():
    """M7: applyCompareClip returns early when the mode is not Compare and the
    layers are already unclipped — avoids re-reading/rewriting the style of
    every layer on each map move when Compare has never been opened."""
    from lidar_pipeline.mapui import _MAP_JS
    clip = _MAP_JS[_MAP_JS.index("function applyCompareClip("):_MAP_JS.index("(function compareDrag(")]
    assert "let compareClipsCleared = true;" in _MAP_JS
    assert "if (!on && compareClipsCleared) { el('compareBar').hidden = true; return; }" in clip
    assert "compareClipsCleared = !on;" in clip