"""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")): import lidar_pipeline.mapserve as mapserve from lidar_pipeline import tiles _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) meta = mapserve.map_meta() assert {l["key"] for l in meta["layers"]} == {"aspect", "slope"} assert all(l["label"] for l in meta["layers"]) assert meta["tile_url"] == "tiles/{layer}/{z}/{x}/{y}@2x.webp" assert meta["tile_size"] == 512 and meta["zoom_offset"] == -1 assert meta["max_native_zoom"] == 18 assert meta["bounds"] and meta["stamp"] > 0 assert "aspect" in meta["default_layers"] 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_offers_reorder_and_blend_modes(): """Pile de couches : réordonnancement au doigt ET modes de fusion. La superposition ne se réduit pas à de la transparence : chaque couche porte un mix-blend-mode, cantonné à la pile LiDAR par un conteneur isolé (sinon « Produit » assombrirait aussi le fond de carte). """ from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS # Modes proposés for mode in ("multiply", "screen", "overlay", "soft-light", "difference", "luminosity"): assert f"'{mode}'" in _MAP_JS, mode # Réordonnancement : glisser-déposer ET boutons (seuls utilisables au doigt) assert "function moveLayer(" in _MAP_JS assert "moveLayer(key, 1)" in _MAP_JS and "moveLayer(key, -1)" in _MAP_JS assert "dragstart" in _MAP_JS and "layer-move" in _MAP_CSS # Pile isolée + fusion par couche et fusion de la pile sur le fond assert "createPane('lidarStack')" in _MAP_JS assert "isolation = 'isolate'" in _MAP_JS assert "map.createPane(pane, stack)" in _MAP_JS assert "pane.style.mixBlendMode" in _MAP_JS assert "STATE.stackBlend" in _MAP_JS and 'id="stackBlend"' in _MAP_HTML # Le lien partagé transporte la fusion (4e champ, rétrocompatible) assert "(STATE.blend[k] || 'normal')" in _MAP_JS assert "const b = validBlend(f[3]);" in _MAP_JS def test_defaults_roundtrip(tmp_path, monkeypatch): """L'état réglé depuis la carte devient la configuration servie à tous.""" import lidar_pipeline.mapserve as mapserve _setup(tmp_path, monkeypatch) 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_order"] is None assert meta["default_stack_blend"] == "normal" req = mapserve.DefaultsRequest( order=["slope", "aspect"], on=["aspect"], opacity={"aspect": 0.6, "slope": 1.0}, blend={"aspect": "multiply", "slope": "normal"}, base={"on": True, "opacity": 0.4, "dark": False}, stack_blend="soft-light") assert mapserve.set_defaults(req)["enregistré"] is True meta = mapserve.map_meta() assert meta["defaults_saved"] is True assert meta["default_order"] == ["slope", "aspect"] assert meta["default_layers"] == ["aspect"] assert meta["default_opacity"]["aspect"] == 0.6 assert meta["default_blend"]["aspect"] == "multiply" assert meta["default_base"] == {"on": True, "opacity": 0.4, "dark": False} assert meta["default_stack_blend"] == "soft-light" # Persisté sur disque : survit au redémarrage du conteneur assert (tmp_path / ".map-defaults.json").is_file() assert mapserve.get_defaults()["defaults"]["on"] == ["aspect"] # 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_sanitised(tmp_path, monkeypatch): """Couches inconnues écartées, opacités bornées, fusions validées.""" import lidar_pipeline.mapserve as mapserve _setup(tmp_path, monkeypatch) monkeypatch.setattr(mapserve, "DEFAULTS_FILE", tmp_path / ".map-defaults.json") data = mapserve.set_defaults(mapserve.DefaultsRequest( order=["aspect", "inexistante"], on=["aspect", "inexistante"], opacity={"aspect": 5, "slope": "x", "inexistante": 0.5}, blend={"aspect": "vaudou", "slope": "screen"}, base={"opacity": -3}, stack_blend="vaudou"))["defaults"] assert "inexistante" not in data["order"] and "inexistante" not in data["on"] # Les couches présentes mais non citées complètent l'ordre assert set(data["order"]) == {"aspect", "slope"} assert data["opacity"]["aspect"] == 1.0 # borné à 1 assert data["opacity"]["slope"] == 1.0 # valeur illisible → opaque assert data["blend"] == {"slope": "screen"} # mode inconnu écarté assert data["base"]["opacity"] == 0.0 # borné à 0 assert data["stack_blend"] == "normal" # repli 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_order" in _MAP_JS assert "meta.default_base" in _MAP_JS assert "validBlend(meta.default_stack_blend)" 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_reorder_shows_destination(): """Réorganiser doit être lisible : repère de chute, fantôme, arrivée signalée. Sans repère, on lâche une couche sans savoir où elle atterrit — la pile n'est pas un détail, elle décide de ce qu'on voit. """ from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS # Barre d'insertion selon la moitié survolée, effacée entre deux survols assert ".layer-row.drop-before" in _MAP_CSS and ".layer-row.drop-after" in _MAP_CSS assert "clearDropMarks" in _MAP_JS assert "box.height / 2" in _MAP_JS assert "row.classList.add(before ? 'drop-before' : 'drop-after')" in _MAP_JS # Ligne déplacée estompée pendant le glisser assert ".layer-row.dragging" in _MAP_CSS and "classList.add('dragging')" in _MAP_JS # Insertion relative (avant/après la ligne visée), pas un simple échange assert "function reorderRelative(" in _MAP_JS # Arrivée mise en évidence puis ramenée dans le champ de vision assert "just-moved" in _MAP_CSS and "@keyframes moved" in _MAP_CSS assert "scrollIntoView" in _MAP_JS assert "renderPanel(key)" in _MAP_JS and "renderPanel(src)" in _MAP_JS # Extrémités de la pile nommées assert 'id="edgeTop"' in _MAP_HTML and 'id="edgeBottom"' in _MAP_HTML assert "Haut de pile" in _MAP_HTML and "Bas de pile" in _MAP_HTML 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(" in _MAP_JS 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 # 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)