72b10cd228
Retour/Échap depuis la fiche détaillée ne retouche plus la hauteur du volet téléphone (openTab remplacé par un rétablissement direct de l'onglet) ; le fond OSM suit le thème sauf choix explicite (migration du localStorage et des défauts serveur en v3, plus de dark:true hérité de l'ex-webapp) et part du bon thème dès sa création (plus de flash sombre) ; l'onglet Export mémorisé ne recadre plus la vue au premier rendu (lien partagé préservé) et referme son cadre pendant que la fiche détaillée est ouverte ; la qualité affichée suit le nom réel de la dalle rendue ; un second clic referme la fiche au lieu de déplacer son contour derrière elle ; couleurs en dur restantes tokenisées (--tile-sel, --gps, .gen-log) et lisibles en thème clair ; localStorage corrompu (lidar-panel, lidar-print) n'y casse plus l'interface. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1607 lines
74 KiB
Python
1607 lines
74 KiB
Python
"""Tests du serveur de tuiles XYZ et de sa découverte (TileJSON, WMTS, JOSM)."""
|
||
|
||
import asyncio
|
||
import json
|
||
import xml.etree.ElementTree as ET
|
||
|
||
from lidar_pipeline.tests.test_tiles import _make_dalle, _tile_of_cell
|
||
|
||
|
||
class _FakeRequest:
|
||
"""Requête minimale : seules l'origine et les en-têtes de proxy comptent."""
|
||
|
||
def __init__(self, host="carte.local:8975", proto=None, scheme="http"):
|
||
self.headers = {"host": host}
|
||
if proto:
|
||
self.headers["x-forwarded-proto"] = proto
|
||
self.url = type("U", (), {"scheme": scheme, "hostname": "carte.local",
|
||
"port": 8975})()
|
||
|
||
|
||
def _setup(tmp_path, monkeypatch, layers=("aspect", "slope"), panel=None):
|
||
"""Dalle factice ; panel=None lève la restriction PANEL_VIZ pour que les
|
||
couches de test (aspect, slope) soient servies."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import index, tiles
|
||
monkeypatch.setattr(index, "PANEL_VIZ", panel)
|
||
_make_dalle(tmp_path, 1054, 6882, list(layers))
|
||
monkeypatch.setattr(mapserve, "OUTPUT_DIR", tmp_path)
|
||
tiles.source_index(tmp_path, force=True)
|
||
return mapserve
|
||
|
||
|
||
def test_tile_png_is_256_rgba(tmp_path, monkeypatch):
|
||
"""La tuile canonique est un PNG RGBA de 256 px, avec en-têtes CORS."""
|
||
import io
|
||
from PIL import Image
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
resp = asyncio.run(mapserve.tile("aspect", z, x, f"{y}.png"))
|
||
assert resp.media_type == "image/png"
|
||
assert resp.headers["Access-Control-Allow-Origin"] == "*"
|
||
assert resp.headers["X-Tile-Empty"] == "0"
|
||
img = Image.open(io.BytesIO(resp.body))
|
||
assert img.size == (256, 256) and img.mode in ("RGBA", "P")
|
||
|
||
|
||
def test_tile_2x_is_512_webp(tmp_path, monkeypatch):
|
||
"""Le palier @2x sert du WebP 512 px (interface interne)."""
|
||
import io
|
||
from PIL import Image
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
resp = asyncio.run(mapserve.tile("aspect", z, x, f"{y}@2x.webp"))
|
||
assert resp.media_type == "image/webp"
|
||
assert Image.open(io.BytesIO(resp.body)).size == (512, 512)
|
||
|
||
|
||
def test_tile_outside_data_is_transparent(tmp_path, monkeypatch):
|
||
"""Hors données : tuile transparente (et pas une erreur) pour JOSM/iD."""
|
||
import io
|
||
from PIL import Image
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
resp = asyncio.run(mapserve.tile("aspect", 15, 1, "1.png"))
|
||
assert resp.headers["X-Tile-Empty"] == "1"
|
||
img = Image.open(io.BytesIO(resp.body)).convert("RGBA")
|
||
assert img.getextrema()[3] == (0, 0)
|
||
|
||
|
||
def test_tile_cache_headers(tmp_path, monkeypatch):
|
||
"""URL versionnée → cache immuable ; sans version → revalidation courte."""
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
plain = asyncio.run(mapserve.tile("aspect", z, x, f"{y}.png"))
|
||
assert "must-revalidate" in plain.headers["Cache-Control"]
|
||
versioned = asyncio.run(mapserve.tile("aspect", z, x, f"{y}.png", v="123"))
|
||
assert "immutable" in versioned.headers["Cache-Control"]
|
||
|
||
|
||
def test_tile_rejects_bad_input(tmp_path, monkeypatch):
|
||
"""Nom de tuile, couche, zoom et indices sont validés (404)."""
|
||
from fastapi import HTTPException
|
||
import pytest
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
for layer, zz, xx, name in (("aspect", z, x, "abc.png"), # nom invalide
|
||
("aspect", z, x, f"{y}.tif"), # format refusé
|
||
("inconnue", z, x, f"{y}.png"), # couche absente
|
||
("../etc", z, x, f"{y}.png"), # traversée
|
||
("aspect", 2, x, f"{y}.png"), # zoom trop bas
|
||
("aspect", 25, x, f"{y}.png"), # zoom trop haut
|
||
("aspect", z, 10 ** 9, f"{y}.png")): # hors grille
|
||
with pytest.raises(HTTPException) as e:
|
||
asyncio.run(mapserve.tile(layer, zz, xx, name))
|
||
assert e.value.status_code == 404
|
||
|
||
|
||
def test_tile_upstream_fallback(tmp_path, monkeypatch):
|
||
"""Tuile absente localement : rapatriée depuis LIDAR_MAPS_URL et mise en cache."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
_setup(tmp_path, monkeypatch)
|
||
monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
|
||
monkeypatch.setattr(mapserve, "_UPSTREAM", {"fails": 0, "until": 0.0})
|
||
called = {}
|
||
|
||
def fake_fetch(rel_path):
|
||
called["path"] = rel_path
|
||
return b"\x89PNG\r\n\x1a\nFAKE"
|
||
|
||
monkeypatch.setattr(mapserve, "_fetch_upstream", fake_fetch)
|
||
# Zone sans donnée locale → le rendu local échoue, l'amont prend le relais
|
||
resp = asyncio.run(mapserve.tile("aspect", 15, 1, "1.png"))
|
||
assert called["path"] == "tiles/aspect/15/1/1.png"
|
||
assert resp.body == b"\x89PNG\r\n\x1a\nFAKE"
|
||
assert tiles.tile_cache_path(tmp_path, "aspect", 15, 1, 1).is_file()
|
||
|
||
|
||
def test_upstream_breaker(monkeypatch):
|
||
"""3 échecs consécutifs suspendent les tentatives amont ; un succès réarme."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
monkeypatch.setattr(mapserve, "_UPSTREAM", {"fails": 0, "until": 0.0})
|
||
for _ in range(2):
|
||
mapserve._upstream_mark(False)
|
||
assert mapserve._upstream_offline() is False
|
||
mapserve._upstream_mark(False)
|
||
assert mapserve._upstream_offline() is True
|
||
mapserve._upstream_mark(True)
|
||
assert mapserve._upstream_offline() is False
|
||
|
||
|
||
def test_tilejson_is_valid(tmp_path, monkeypatch):
|
||
"""TileJSON 3.0.0 : champs obligatoires, URL XYZ absolue, attribution."""
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
data = json.loads(mapserve.tilejson("aspect", _FakeRequest()).body)
|
||
assert data["tilejson"] == "3.0.0"
|
||
assert data["scheme"] == "xyz"
|
||
assert data["tiles"] == ["http://carte.local:8975/tiles/aspect/{z}/{x}/{y}.png"]
|
||
assert data["minzoom"] < data["maxzoom"] <= 19
|
||
assert len(data["bounds"]) == 4 and data["bounds"][0] < data["bounds"][2]
|
||
assert "IGN" in data["attribution"]
|
||
|
||
|
||
def test_tilejson_unknown_layer(tmp_path, monkeypatch):
|
||
"""Couche absente du disque → 404."""
|
||
from fastapi import HTTPException
|
||
import pytest
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
with pytest.raises(HTTPException) as e:
|
||
mapserve.tilejson("wavelet", _FakeRequest())
|
||
assert e.value.status_code == 404
|
||
|
||
|
||
def test_base_url_honours_proxy(tmp_path, monkeypatch):
|
||
"""Derrière un proxy TLS, les URL publiées sont en https."""
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
data = json.loads(mapserve.tilejson(
|
||
"aspect", _FakeRequest(host="lidar.example.fr", proto="https")).body)
|
||
assert data["tiles"][0].startswith("https://lidar.example.fr/tiles/")
|
||
|
||
|
||
def test_josm_imagery_xml(tmp_path, monkeypatch):
|
||
"""Fichier d'imagerie JOSM : une entrée TMS par couche, URL en {zoom}/{x}/{y}."""
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
xml = mapserve.josm_imagery(_FakeRequest()).body.decode("utf-8")
|
||
root = ET.fromstring(xml)
|
||
ns = {"j": "http://josm.openstreetmap.de/maps-1.0"}
|
||
entries = root.findall("j:entry", ns)
|
||
assert {e.find("j:id", ns).text for e in entries} == {
|
||
"lidar-hd-aspect", "lidar-hd-slope"}
|
||
for entry in entries:
|
||
assert entry.find("j:type", ns).text == "tms"
|
||
assert "{zoom}/{x}/{y}.png" in entry.find("j:url", ns).text
|
||
assert entry.find("j:max-zoom", ns).text == "19"
|
||
assert entry.find("j:bounds", ns) is not None
|
||
|
||
|
||
def test_wmts_capabilities(tmp_path, monkeypatch):
|
||
"""Capacités WMTS : couches, grille GoogleMapsCompatible, gabarit REST."""
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
xml = mapserve.wmts_capabilities(_FakeRequest()).body.decode("utf-8")
|
||
root = ET.fromstring(xml)
|
||
ns = {"w": "http://www.opengis.net/wmts/1.0",
|
||
"ows": "http://www.opengis.net/ows/1.1"}
|
||
ids = [l.find("ows:Identifier", ns).text
|
||
for l in root.findall(".//w:Layer", ns)]
|
||
assert set(ids) == {"aspect", "slope"}
|
||
# `.//w:TileMatrixSet` attraperait d'abord le renvoi de
|
||
# TileMatrixSetLink (simple texte) : on vise la définition.
|
||
tms = root.find("w:Contents/w:TileMatrixSet", ns)
|
||
assert tms.find("ows:Identifier", ns).text == "GoogleMapsCompatible"
|
||
levels = [m.find("ows:Identifier", ns).text
|
||
for m in tms.findall("w:TileMatrix", ns)]
|
||
assert levels[0] == "5" and levels[-1] == "19"
|
||
template = root.find(".//w:ResourceURL", ns).get("template")
|
||
assert template.endswith("/{TileMatrix}/{TileCol}/{TileRow}.png")
|
||
|
||
|
||
def test_map_meta(tmp_path, monkeypatch):
|
||
"""/api/map/meta décrit les couches et le contrat de tuilage de l'interface."""
|
||
mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente", "aspect", "slope"))
|
||
meta = mapserve.map_meta()
|
||
assert {l["key"] for l in meta["layers"]} == {"relief_oriente", "aspect", "slope"}
|
||
assert all(l["label"] for l in meta["layers"])
|
||
assert meta["tile_url"] == "tiles/{layer}/{z}/{x}/{y}@2x.avif"
|
||
assert meta["even_levels"] is False # pyramide complète générée d'avance
|
||
assert meta["tile_size"] == 512 and meta["zoom_offset"] == -1
|
||
assert meta["max_native_zoom"] == 19
|
||
assert meta["bounds"] and meta["stamp"] > 0
|
||
assert meta["default_main"] == "relief_oriente"
|
||
assert meta["precision_layer"] is None # densité absente de la fixture
|
||
assert meta["default_mode"] == "relief"
|
||
|
||
|
||
def test_only_panel_layers_are_served(tmp_path, monkeypatch):
|
||
"""Seules les couches de PANEL_VIZ sont affichées et servies ; la
|
||
génération depuis la carte ne produit qu'elles."""
|
||
from fastapi import HTTPException
|
||
mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente", "aspect", "slope"),
|
||
panel=("relief_oriente",))
|
||
meta = mapserve.map_meta()
|
||
assert [l["key"] for l in meta["layers"]] == ["relief_oriente"]
|
||
with pytest.raises(HTTPException):
|
||
asyncio.run(mapserve.tile("aspect", 15, 1, "1.png"))
|
||
assert mapserve._panel_viz_steps() == ["relief_oriente"]
|
||
|
||
|
||
def test_map_tile_info(tmp_path, monkeypatch):
|
||
"""/api/map/tile identifie la dalle sous un point et lit ses métadonnées."""
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
dtm = tmp_path / "DTM"
|
||
dtm.mkdir(parents=True, exist_ok=True)
|
||
(dtm / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_dtm_method.txt").write_text(
|
||
"ign", encoding="utf-8")
|
||
from lidar_pipeline.tiles import _transformer
|
||
lon, lat = _transformer("EPSG:2154", "EPSG:4326").transform(1054500, 6881500)
|
||
data = mapserve.map_tile_info(lat=lat, lng=lon)
|
||
assert data["found"] is True
|
||
assert (data["col"], data["row"]) == (1054, 6882)
|
||
assert data["method"] == "ign"
|
||
assert data["layers"] == ["aspect", "slope"]
|
||
# Point hors zone : réponse explicite, sans erreur
|
||
assert mapserve.map_tile_info(lat=45.0, lng=1.0)["found"] is False
|
||
|
||
|
||
def test_map_tile_includes_quality(tmp_path, monkeypatch):
|
||
"""/api/map/tile joint la qualité (densité, dates d'acquisition) si connue."""
|
||
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):
|
||
"""Sans sidecar qualité, la clé vaut None (pas d'erreur)."""
|
||
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):
|
||
"""La qualité lue est celle du nom réel de la dalle rendue (best["basename"]),
|
||
pas celle du nom nominal calculé depuis col/row — ils peuvent différer."""
|
||
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):
|
||
"""Sonde de santé et assets de l'interface (revalidés, jamais immuables)."""
|
||
from fastapi import HTTPException
|
||
import pytest
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
assert mapserve.healthz()["layers"] == 2
|
||
resp = mapserve.assets("app.js")
|
||
assert resp.headers["Cache-Control"] == "no-cache, must-revalidate"
|
||
with pytest.raises(HTTPException):
|
||
mapserve.assets("introuvable.js")
|
||
|
||
|
||
def test_ui_main_layer_and_precision_modes():
|
||
"""Une couche d'affichage principal et la précision : relief seul,
|
||
précision seule, ou précision en « produit » sur le relief.
|
||
|
||
Plus de pile réordonnable ni de fusion par couche. Le produit est
|
||
cantonné à la pile LiDAR par un conteneur isolé (sinon il assombrirait
|
||
aussi le fond de carte). Bascule rapide au clavier (touche P).
|
||
"""
|
||
from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
|
||
for mode in ('data-mode="relief"', 'data-mode="precision"', 'data-mode="both"'):
|
||
assert mode in _MAP_HTML
|
||
assert 'id="precOpacity"' in _MAP_HTML and 'id="precLegend"' in _MAP_HTML
|
||
assert 'id="mainSel"' in _MAP_HTML
|
||
assert "blend = 'multiply'" in _MAP_JS
|
||
assert "createPane('lidarStack')" in _MAP_JS and "isolation = 'isolate'" in _MAP_JS
|
||
assert "e.key === 'p'" in _MAP_JS
|
||
# Légende en 16 paliers exacts (échelle log de la densité)
|
||
assert "const PREC_LEVELS = 16;" in _MAP_JS and "repeat(16, 1fr)" in _MAP_CSS
|
||
# Pile et fusions supprimées
|
||
for gone in ("function moveLayer(", "reorderRelative", "stackBlend", "addEventListener('dragstart'", "blendOptions"):
|
||
assert gone not in _MAP_JS, gone
|
||
assert "Haut de pile" not in _MAP_HTML
|
||
# Lien partagé : principale, mode:opacité, fond
|
||
assert "'&M=' + STATE.main" in _MAP_JS
|
||
assert "'&P=' + STATE.mode" in _MAP_JS
|
||
|
||
|
||
def test_map_ui_tile_load_indicator_and_selection_pane():
|
||
"""Chargement : compteur de tuiles restantes + contour de dalle visible.
|
||
|
||
Le compteur lit l'état réel des tuiles (une tuile retirée en cours de
|
||
chargement n'émet aucun événement de fin). Le contour de sélection vit
|
||
dans un panneau au-dessus de la pile LiDAR (z-index 400), sinon les
|
||
tuiles le masquent.
|
||
"""
|
||
from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
|
||
assert 'id="tileload"' in _MAP_HTML and 'id="tileloadText"' in _MAP_HTML
|
||
assert "@keyframes tlspin" in _MAP_CSS
|
||
assert "ZoomAnimTileLayer.addInitHook" in _MAP_JS
|
||
assert "t.current && !t.loaded" in _MAP_JS
|
||
assert "['tileSel', 680]" in _MAP_JS
|
||
assert "pane: 'tileSel'" in _MAP_JS
|
||
|
||
|
||
def test_ui_tile_popup_and_details_view():
|
||
"""La fiche de dalle est une bulle Leaflet + une vue détaillée dans le panneau."""
|
||
from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
|
||
assert 'id="tilecard"' not in _MAP_HTML
|
||
for id_ in ("tileView", "tileBack", "tileTitle", "tileBody"):
|
||
assert f'id="{id_}"' in _MAP_HTML, id_
|
||
for fn in ("function showTilePopup(", "function openTileDetails(", "function closeTileViews("):
|
||
assert fn in _MAP_JS, fn
|
||
assert "L.popup(" in _MAP_JS and "Détails" in _MAP_JS
|
||
# Pas de bulle pendant le déplacement du cadre d'export ni un tracé
|
||
click = _MAP_JS[_MAP_JS.index("map.on('click'"):]
|
||
assert "genDrawing" in click[:400] and "PRINT.dragging" in click[:400]
|
||
# Changer d'onglet pendant que la fiche est ouverte nettoie le contour
|
||
# de sélection (pas de fuite de tileSel) : openTab passe par le même
|
||
# nettoyage que closeTileViews, sans se rappeler lui-même.
|
||
assert "function hideTileView(" in _MAP_JS
|
||
open_tab = _MAP_JS[_MAP_JS.index("function openTab("):]
|
||
assert "hideTileView()" in open_tab[:200]
|
||
|
||
|
||
def test_defaults_roundtrip(tmp_path, monkeypatch):
|
||
"""L'affichage réglé depuis la carte devient la configuration servie à tous."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
_setup(tmp_path, monkeypatch, layers=("relief_oriente", "aspect", "densite_sol"))
|
||
monkeypatch.setattr(mapserve, "DEFAULTS_FILE", tmp_path / ".map-defaults.json")
|
||
|
||
# Sans enregistrement : réglages du registre (index.py)
|
||
meta = mapserve.map_meta()
|
||
assert meta["defaults_saved"] is False
|
||
assert meta["default_main"] == "relief_oriente"
|
||
assert meta["precision_layer"] == "densite_sol"
|
||
assert meta["default_mode"] == "relief"
|
||
assert meta["default_precision_opacity"] == 0.6
|
||
|
||
req = mapserve.DefaultsRequest(main="aspect", mode="both", precision_opacity=0.4,
|
||
base={"on": True, "opacity": 0.4, "dark": False})
|
||
assert mapserve.set_defaults(req)["enregistré"] is True
|
||
|
||
meta = mapserve.map_meta()
|
||
assert meta["defaults_saved"] is True
|
||
assert meta["default_main"] == "aspect"
|
||
assert meta["default_mode"] == "both"
|
||
assert meta["default_precision_opacity"] == 0.4
|
||
assert meta["default_base"] == {"on": True, "opacity": 0.4, "dark": False}
|
||
# Persisté sur disque : survit au redémarrage du conteneur
|
||
assert (tmp_path / ".map-defaults.json").is_file()
|
||
assert mapserve.get_defaults()["defaults"]["mode"] == "both"
|
||
|
||
# Retrait : retour aux réglages du registre
|
||
assert mapserve.clear_defaults()["supprimé"] is True
|
||
assert mapserve.map_meta()["defaults_saved"] is False
|
||
|
||
|
||
def test_defaults_legacy_stack_file_falls_back(tmp_path, monkeypatch):
|
||
"""Fichier de l'ancienne pile (order/on/blend) ou couche disparue :
|
||
couche principale du registre, mode relief ; le fond est conservé."""
|
||
import json
|
||
import lidar_pipeline.mapserve as mapserve
|
||
_setup(tmp_path, monkeypatch, layers=("relief_oriente",))
|
||
defaults = tmp_path / ".map-defaults.json"
|
||
monkeypatch.setattr(mapserve, "DEFAULTS_FILE", defaults)
|
||
defaults.write_text(json.dumps({
|
||
"order": ["aspect"], "on": ["aspect"], "opacity": {}, "blend": {},
|
||
"base": {"on": True, "opacity": 0.5, "dark": True}, "stack_blend": "normal"}),
|
||
encoding="utf-8")
|
||
meta = mapserve.map_meta()
|
||
assert meta["default_main"] == "relief_oriente"
|
||
assert meta["default_mode"] == "relief"
|
||
assert meta["default_base"]["opacity"] == 0.5
|
||
defaults.write_text(json.dumps({"main": "aspect", "mode": "vaudou"}), encoding="utf-8")
|
||
meta = mapserve.map_meta()
|
||
assert meta["default_main"] == "relief_oriente" and meta["default_mode"] == "relief"
|
||
|
||
|
||
def test_defaults_sanitised(tmp_path, monkeypatch):
|
||
"""Couche inconnue ou précision refusée comme principale, mode validé, opacités bornées."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
_setup(tmp_path, monkeypatch, layers=("relief_oriente", "densite_sol"))
|
||
monkeypatch.setattr(mapserve, "DEFAULTS_FILE", tmp_path / ".map-defaults.json")
|
||
data = mapserve.set_defaults(mapserve.DefaultsRequest(
|
||
main="inexistante", mode="vaudou", precision_opacity=5,
|
||
base={"opacity": -3}))["defaults"]
|
||
assert data["main"] is None
|
||
assert data["mode"] == "relief"
|
||
assert data["precision_opacity"] == 1.0 # borné à 1
|
||
assert data["base"]["opacity"] == 0.0 # borné à 0
|
||
data = mapserve.set_defaults(mapserve.DefaultsRequest(main="densite_sol"))["defaults"]
|
||
assert data["main"] is None # la précision n'est pas une principale
|
||
|
||
|
||
def test_defaults_dark_none_is_not_a_choice(tmp_path, monkeypatch):
|
||
"""base.dark omis/None n'est pas un choix : il reste None (le fond suit le
|
||
thème côté carte), distinct d'un choix explicite (True/False persistent)."""
|
||
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
|
||
|
||
# Choix explicite : conservé tel quel
|
||
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):
|
||
"""Fichier d'avant v3 (ex-webapp, base.dark:true systématique) : servi avec
|
||
dark None (« non choisi »), jamais imposé comme choix de l'utilisateur."""
|
||
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_ui_applies_server_defaults():
|
||
"""L'interface part des défauts du serveur et sait les (re)définir."""
|
||
from lidar_pipeline.mapui import _MAP_HTML, _MAP_JS
|
||
assert 'id="btnDefault"' in _MAP_HTML and 'id="btnReset"' in _MAP_HTML
|
||
assert "api/map/defaults" in _MAP_JS
|
||
assert "meta.default_main" in _MAP_JS and "meta.default_mode" in _MAP_JS
|
||
assert "meta.default_base" in _MAP_JS
|
||
assert "meta.default_precision_opacity" in _MAP_JS
|
||
# Réinitialiser oublie l'état local avant de reprendre celui du serveur
|
||
assert "lsDel(LS_KEY)" in _MAP_JS
|
||
|
||
|
||
def test_ui_base_dark_follows_theme_unless_chosen():
|
||
"""base.dark ne vaut jamais un choix implicite : None = suit le thème.
|
||
|
||
Client : defaultState garde None (au lieu de forcer un booléen) ; loadState
|
||
ignore le base.dark d'un ancien lidarMapView_v2 (avant v3, où l'ex-webapp
|
||
persistait dark:true à chaque sauvegarde) ; saveState écrit v: 3. La classe
|
||
CSS du fond OSM part du thème courant, pas d'un « base-dark » figé."""
|
||
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_phone_return_and_escape_keep_sheet_height():
|
||
"""« ← Retour » et Échap depuis la fiche de dalle rétablissent l'onglet
|
||
précédent SANS repasser par openTab (qui retoucherait la hauteur du volet
|
||
en re-touchant l'onglet déjà actif) : la fiche ferme via un rétablissement
|
||
direct de PANEL.tab/want, partagé par closeTileViews et showTilePopup."""
|
||
from lidar_pipeline.mapui import _MAP_JS
|
||
assert "function restoreTabAfterDetails()" in _MAP_JS
|
||
restore = _MAP_JS[_MAP_JS.index("function restoreTabAfterDetails("):]
|
||
assert "PANEL.tab = PANEL.want = tileReturnTab || 'view';" in restore[:300]
|
||
assert "openTab(" not in restore[:300]
|
||
close_tiles = _MAP_JS[_MAP_JS.index("function closeTileViews("):]
|
||
assert "restoreTabAfterDetails()" in close_tiles[:250]
|
||
# Second clic sur la carte pendant que la fiche est ouverte : referme la
|
||
# fiche plutôt que de bouger le contour derrière elle (M5)
|
||
popup_fn = _MAP_JS[_MAP_JS.index("function showTilePopup("):]
|
||
assert "restoreTabAfterDetails()" in popup_fn[:400]
|
||
|
||
|
||
def test_ui_saved_print_tab_does_not_rezoom_shared_link():
|
||
"""L'onglet Export mémorisé ne recadre pas la vue au tout premier rendu du
|
||
panneau (sinon il écrase le zoom d'un lien partagé ou de l'URL initiale) :
|
||
BOOTSTRAPPING passe à faux juste après ce premier rendu."""
|
||
from lidar_pipeline.mapui import _MAP_JS
|
||
assert "let BOOTSTRAPPING = true;" in _MAP_JS
|
||
assert "printSetOpen(e.detail.visible === 'print', !BOOTSTRAPPING);" in _MAP_JS
|
||
boot = _MAP_JS[_MAP_JS.index("panelRender();\n BOOTSTRAPPING = false;"):]
|
||
assert boot.startswith("panelRender();\n BOOTSTRAPPING = false;")
|
||
|
||
|
||
def test_ui_details_view_hides_print_frame():
|
||
"""Ouvrir la fiche détaillée pendant que l'onglet Export est affiché
|
||
referme le cadre d'export (sinon cadre + poignée restent visibles
|
||
derrière la fiche)."""
|
||
from lidar_pipeline.mapui import _MAP_JS
|
||
open_details = _MAP_JS[_MAP_JS.index("function openTileDetails("):]
|
||
assert "printSetOpen(false);" in open_details[:600]
|
||
|
||
|
||
def test_ui_colors_are_tokenised():
|
||
"""Plus de couleur en dur pour le contour de sélection, le marqueur GPS et
|
||
le journal de génération (AGENTS.md « aucune couleur en dur dans les
|
||
composants ») ; #1b1b1b reste en dur sur la poignée d'export (contraste
|
||
fixe voulu sur --print), documenté en commentaire."""
|
||
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 # volontaire, commenté juste au-dessus
|
||
|
||
|
||
def test_ui_guards_corrupt_local_storage():
|
||
"""Une valeur corrompue dans lidar-panel/lidar-print (autre appli sur la
|
||
même origine, quota dépassé en écriture…) ne casse pas l'interface :
|
||
whitelist de type/valeur plutôt qu'un usage direct de 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():
|
||
"""L'interface s'appuie sur le LOD natif de Leaflet, pas sur un palier maison."""
|
||
from lidar_pipeline.mapui import _MAP_JS, render_html, ui_version
|
||
assert "L.TileLayer.extend(" in _MAP_JS # TileLayer Leaflet, LOD natif
|
||
assert "tileSize: META.tile_size" in _MAP_JS
|
||
assert "zoomOffset: META.zoom_offset" in _MAP_JS
|
||
assert "maxNativeZoom: META.max_native_zoom" in _MAP_JS
|
||
# Le niveau natif impair (19 = 0,2 m/px) est demandé tel quel, jamais
|
||
# replié sur le pair inférieur (affichage plafonné à 0,4 m/px).
|
||
assert "z % 2 && (maxN === undefined || z < maxN)" in _MAP_JS
|
||
# Plus aucune mécanique de l'ancienne carte
|
||
for banned in ("quadMatrix3d", "FULL_CAP", "ovTick", "layoutTiles"):
|
||
assert banned not in _MAP_JS
|
||
html = render_html()
|
||
assert ui_version() in html and "__UI_VERSION__" not in html
|
||
|
||
|
||
def test_write_map_assets(tmp_path):
|
||
"""Les assets écrits portent la version et embarquent Leaflet vendorisé."""
|
||
from lidar_pipeline.mapui import _MAP_JS, ui_version, write_map_assets
|
||
write_map_assets(tmp_path / "ui")
|
||
js = (tmp_path / "ui" / "app.js").read_text(encoding="utf-8")
|
||
assert js == _MAP_JS.replace("__UI_VERSION__", ui_version())
|
||
assert (tmp_path / "ui" / "app.css").read_text(encoding="utf-8").startswith("/*")
|
||
assert (tmp_path / "ui" / "vendor" / "leaflet" / "leaflet.js").is_file()
|
||
|
||
|
||
def test_warm_endpoint(tmp_path, monkeypatch):
|
||
"""/api/map/warm lance le pré-calcul en tâche de fond puis rend compte."""
|
||
import time
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
monkeypatch.setattr(mapserve, "_warm", {"running": False, "done": None,
|
||
"result": None})
|
||
req = mapserve.WarmRequest(layers=["aspect"], z_min=12, z_max=12)
|
||
assert mapserve.map_warm(req)["couches"] == ["aspect"]
|
||
for _ in range(200):
|
||
if not mapserve._warm["running"]:
|
||
break
|
||
time.sleep(0.05)
|
||
assert mapserve.map_warm_status()["result"]["rendues"] >= 1
|
||
|
||
|
||
def test_tile_locks_are_refcounted(tmp_path, monkeypatch):
|
||
"""Verrou par tuile : partagé pendant le rendu, retiré une fois libéré.
|
||
|
||
Sans compteur, le registre grossit sans fin (des millions de tuiles) ou un
|
||
verrou disparaît alors qu'un autre fil l'attend — deux rendus concurrents
|
||
de la même tuile.
|
||
"""
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
key = ("aspect", 15, 1, 1, 1, "png")
|
||
a = mapserve._acquire_lock(key)
|
||
b = mapserve._acquire_lock(key)
|
||
assert a is b and a[1] == 2
|
||
mapserve._release_lock(key, a)
|
||
assert key in mapserve._locks
|
||
mapserve._release_lock(key, b)
|
||
assert key not in mapserve._locks
|
||
|
||
|
||
def test_concurrent_requests_render_once(tmp_path, monkeypatch):
|
||
"""Deux requêtes simultanées sur la même tuile ne la rendent qu'une fois."""
|
||
import threading
|
||
import lidar_pipeline.mapserve as mapserve
|
||
_setup(tmp_path, monkeypatch)
|
||
calls = []
|
||
real = mapserve.tiles_mod.get_tile
|
||
|
||
def slow_get_tile(*args, **kwargs):
|
||
calls.append(args)
|
||
threading.Event().wait(0.05)
|
||
return real(*args, **kwargs)
|
||
|
||
monkeypatch.setattr(mapserve.tiles_mod, "get_tile", slow_get_tile)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
results = []
|
||
threads = [threading.Thread(
|
||
target=lambda: results.append(mapserve._tile_bytes("aspect", z, x, y, 1, "png")))
|
||
for _ in range(4)]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join()
|
||
assert all(r == results[0] for r in results)
|
||
# Les appels suivants lisent le cache disque : un seul rendu effectif
|
||
assert len(calls) == 4 # chaque requête interroge get_tile…
|
||
assert mapserve._locks == {} # …mais sérialisées, et les verrous libérés
|
||
|
||
|
||
def test_tile_bytes_survives_render_error(tmp_path, monkeypatch):
|
||
"""Un rendu qui échoue rend une tuile vide sans laisser de verrou fuité."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
_setup(tmp_path, monkeypatch)
|
||
|
||
def boom(*a, **k):
|
||
raise RuntimeError("source corrompue")
|
||
|
||
monkeypatch.setattr(mapserve.tiles_mod, "get_tile", boom)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
assert mapserve._tile_bytes("aspect", z, x, y, 1, "png") is None
|
||
assert mapserve._locks == {}
|
||
|
||
|
||
def test_render_concurrency_is_capped(tmp_path, monkeypatch):
|
||
"""Le plafond de rendus simultanés est tenu (petit serveur protégé).
|
||
|
||
Sans lui, autant de rendus que de requêtes en vol : chacun décode une dalle
|
||
de plusieurs dizaines de Mo — mémoire et CPU d'un Raspberry Pi saturés.
|
||
"""
|
||
import threading
|
||
import lidar_pipeline.mapserve as mapserve
|
||
_setup(tmp_path, monkeypatch)
|
||
|
||
# Plafond volontairement bas pour le test
|
||
monkeypatch.setattr(mapserve, "_render_sem", threading.Semaphore(2))
|
||
state = {"now": 0, "peak": 0}
|
||
guard = threading.Lock()
|
||
real = mapserve.tiles_mod.get_tile
|
||
|
||
def slow(*args, **kwargs):
|
||
with guard:
|
||
state["now"] += 1
|
||
state["peak"] = max(state["peak"], state["now"])
|
||
try:
|
||
threading.Event().wait(0.05)
|
||
return real(*args, **kwargs)
|
||
finally:
|
||
with guard:
|
||
state["now"] -= 1
|
||
|
||
monkeypatch.setattr(mapserve.tiles_mod, "get_tile", slow)
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
# Tuiles DIFFÉRENTES : le verrou par tuile ne masque pas le sémaphore
|
||
threads = [threading.Thread(target=mapserve._tile_bytes,
|
||
args=("aspect", z, x + i, y, 1, "png"))
|
||
for i in range(8)]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join()
|
||
assert state["peak"] <= 2, state["peak"]
|
||
assert mapserve._locks == {}
|
||
|
||
|
||
def test_worker_limits_are_configurable(monkeypatch):
|
||
"""Les plafonds viennent de l'environnement, avec un défaut modeste."""
|
||
import importlib
|
||
import lidar_pipeline.mapserve as mapserve
|
||
import lidar_pipeline.tiles as tiles
|
||
|
||
# Défauts : 2 rendus, 2 téléchargements — taillés pour un petit serveur
|
||
assert mapserve.TILE_WORKERS == 2
|
||
assert tiles.FETCH_WORKERS == 2
|
||
|
||
monkeypatch.setenv("LIDAR_TILE_WORKERS", "6")
|
||
monkeypatch.setenv("LIDAR_TILE_FETCH_WORKERS", "3")
|
||
try:
|
||
importlib.reload(tiles)
|
||
importlib.reload(mapserve)
|
||
assert mapserve.TILE_WORKERS == 6
|
||
assert tiles.FETCH_WORKERS == 3
|
||
# Valeur absurde : jamais zéro fil (blocage complet du service)
|
||
monkeypatch.setenv("LIDAR_TILE_WORKERS", "0")
|
||
importlib.reload(mapserve)
|
||
assert mapserve.TILE_WORKERS == 1
|
||
finally:
|
||
monkeypatch.delenv("LIDAR_TILE_WORKERS", raising=False)
|
||
monkeypatch.delenv("LIDAR_TILE_FETCH_WORKERS", raising=False)
|
||
importlib.reload(tiles)
|
||
importlib.reload(mapserve)
|
||
|
||
|
||
def _reset_bg(mapserve):
|
||
mapserve._bg["queue"].clear()
|
||
mapserve._bg["queued"].clear()
|
||
mapserve._bg["snapshot"] = None
|
||
mapserve._bg["incomplete"] = {}
|
||
mapserve._bg["stats"].update(rendues=0, a_jour=0, vides=0, scans=0,
|
||
dernier_scan=None, ajouts_dernier_scan=0)
|
||
|
||
|
||
def test_tile_cache_only_never_renders(tmp_path, monkeypatch):
|
||
"""Mode cache seule : tuile manquante = transparente non mémorisable."""
|
||
from lidar_pipeline import tiles as _t
|
||
monkeypatch.setattr(_t, "TILE_EVEN_LEVELS", False) # règle historique : tout est stocké
|
||
monkeypatch.setattr(_t, "TILE_CACHE_MAX_Z", 99)
|
||
mapserve = _setup(tmp_path, monkeypatch, layers=("slope",))
|
||
from lidar_pipeline import tiles
|
||
monkeypatch.setenv("LIDAR_TILE_CACHE_ONLY", "1")
|
||
z = 15
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
resp = asyncio.run(mapserve.tile("slope", z, x, f"{y}.png"))
|
||
assert resp.headers["X-Tile-Pending"] == "1"
|
||
assert resp.headers["X-Tile-Empty"] == "1"
|
||
assert resp.headers["Cache-Control"] == "no-store"
|
||
assert not tiles.tile_cache_path(tmp_path, "slope", z, x, y).exists()
|
||
# Une fois rendue par ailleurs : servie depuis le cache, mémorisable
|
||
tiles.get_tile(tmp_path, "slope", z, x, y)
|
||
resp = asyncio.run(mapserve.tile("slope", z, x, f"{y}.png", v="s"))
|
||
assert resp.headers["X-Tile-Empty"] == "0"
|
||
assert "X-Tile-Pending" not in resp.headers
|
||
assert "immutable" in resp.headers["Cache-Control"]
|
||
|
||
|
||
def test_background_scan_detects_and_fills(tmp_path, monkeypatch):
|
||
"""La maintenance détecte les dalles, remplit la pyramide, puis se tait."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
_make_dalle(tmp_path, 1054, 6882, ["slope"])
|
||
monkeypatch.setattr(mapserve, "OUTPUT_DIR", tmp_path)
|
||
monkeypatch.setenv("LIDAR_TILE_BACKGROUND_MAX_Z", "14")
|
||
monkeypatch.setenv("LIDAR_TILE_BACKGROUND_SCALE", "1")
|
||
monkeypatch.setenv("LIDAR_TILE_BACKGROUND_FMT", "png")
|
||
monkeypatch.setenv("LIDAR_TILE_BACKGROUND_PAUSE", "0")
|
||
_reset_bg(mapserve)
|
||
try:
|
||
# Premier scan : tout est nouveau, bas niveaux en tête de file
|
||
assert mapserve.background_scan(tmp_path) > 0
|
||
zs = [t[1] for t in mapserve._bg["queue"]]
|
||
assert zs == sorted(zs) and max(zs) <= 14
|
||
processed = 0
|
||
while mapserve._bg["queue"] and processed < 1000:
|
||
mapserve._bg_process_one()
|
||
processed += 1
|
||
assert not mapserve._bg["queue"]
|
||
assert mapserve._bg["stats"]["rendues"] > 0
|
||
x, y = _tile_of_cell(1054, 6882, 14)
|
||
assert tiles.tile_cache_path(
|
||
tmp_path, "slope", 14, x, y, 1, "png").exists()
|
||
# Rien n'a changé : plus aucun ajout
|
||
assert mapserve.background_scan(tmp_path) == 0
|
||
# Une dalle nouvelle (worker) : sa pyramide repart seule en file
|
||
_make_dalle(tmp_path, 1055, 6882, ["slope"])
|
||
assert mapserve.background_scan(tmp_path) > 0
|
||
finally:
|
||
_reset_bg(mapserve)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Génération de tuiles (portée de la webapp historique)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
import types
|
||
|
||
from fastapi import HTTPException
|
||
import pytest
|
||
|
||
|
||
class _ClientRequest:
|
||
"""Requête minimale avec IP de connexion (restriction LIDAR_REGEN_CIDR)."""
|
||
|
||
def __init__(self, ip, xff=None):
|
||
self.client = types.SimpleNamespace(host=ip)
|
||
self.headers = {"x-forwarded-for": xff} if xff else {}
|
||
|
||
|
||
def _gen_setup(tmp_path, monkeypatch):
|
||
import lidar_pipeline.mapserve as mapserve
|
||
_make_dalle(tmp_path, 1054, 6882, ["aspect", "slope"])
|
||
monkeypatch.setattr(mapserve, "OUTPUT_DIR", tmp_path)
|
||
monkeypatch.setattr(mapserve, "INPUT_DIR", tmp_path / "input")
|
||
monkeypatch.setattr(mapserve, "GENERATION_URL", "")
|
||
monkeypatch.setattr(mapserve, "LOG_FILE", tmp_path / ".generation.log")
|
||
monkeypatch.setattr(mapserve, "JOB_FILE", tmp_path / ".generation.job.json")
|
||
monkeypatch.setattr(mapserve, "QUEUE_FILE", tmp_path / ".generation.queue.json")
|
||
mapserve._queue.clear()
|
||
mapserve._pipeline_import_cache.clear()
|
||
return mapserve
|
||
|
||
|
||
def test_bbox_to_cells_covers_1km():
|
||
"""Une bbox d'1 km² WGS84 sélectionne la cellule L93 correspondante."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline.tiles import _transformer
|
||
x, y = 1054500, 6881500
|
||
w, n = _transformer("EPSG:2154", "EPSG:4326").transform(x - 200, y + 200)
|
||
e, s = _transformer("EPSG:2154", "EPSG:4326").transform(x + 200, y - 200)
|
||
cells = mapserve.bbox_to_cells(w, s, e, n)
|
||
assert (1054, 6882) in cells
|
||
# Cellule (col, row) : X ∈ [col, col+1] km, Y ∈ [row-1, row] km
|
||
assert all(1053 <= c <= 1055 and 6881 <= r <= 6883 for c, r in cells)
|
||
|
||
|
||
def test_preview_bbox_lists_cells_with_corners(tmp_path, monkeypatch):
|
||
"""Le preview retourne les cellules à traiter avec leurs coins GPS."""
|
||
mapserve = _gen_setup(tmp_path, monkeypatch)
|
||
from lidar_pipeline.tiles import _transformer
|
||
x, y = 1054500, 6881500
|
||
w, n = _transformer("EPSG:2154", "EPSG:4326").transform(x - 100, y + 100)
|
||
e, s = _transformer("EPSG:2154", "EPSG:4326").transform(x + 100, y - 100)
|
||
req = mapserve.PreviewRequest(bbox=[w, s, e, n], viz=["aspect"])
|
||
d = mapserve.preview(req)
|
||
assert d["capped"] is False
|
||
cells = {(c["col"], c["row"]) for c in d["cells"]}
|
||
assert (1054, 6882) in cells
|
||
first = d["cells"][0]
|
||
assert first["corners"] and len(first["corners"]) == 4
|
||
|
||
|
||
def test_preview_rejects_unknown_viz(tmp_path, monkeypatch):
|
||
mapserve = _gen_setup(tmp_path, monkeypatch)
|
||
req = mapserve.PreviewRequest(bbox=[0.0, 44.0, 0.01, 44.01],
|
||
viz=["inconnue"])
|
||
with pytest.raises(HTTPException) as e:
|
||
mapserve.preview(req)
|
||
assert e.value.status_code == 400
|
||
|
||
|
||
def test_preview_all_missing_without_laz(tmp_path, monkeypatch):
|
||
"""Passe globale sans dalle dans input/ : liste vide (rien à compléter).
|
||
|
||
L'erreur 400 « aucune tuile à compléter » ne concerne que le lancement
|
||
(/api/generate → _resolve_request) : le preview reste informatif.
|
||
"""
|
||
mapserve = _gen_setup(tmp_path, monkeypatch)
|
||
(tmp_path / "input").mkdir()
|
||
req = mapserve.PreviewRequest(all_missing=True, viz=["aspect"])
|
||
d = mapserve.preview(req)
|
||
assert d["count"] == 0 and d["cells"] == []
|
||
|
||
|
||
def test_generate_validates_request(tmp_path, monkeypatch):
|
||
mapserve = _gen_setup(tmp_path, monkeypatch)
|
||
for kwargs in ({"tiles": []},
|
||
{"tiles": [[1054]]},
|
||
{"tiles": [[c, 6882] for c in range(500)]}):
|
||
req = mapserve.GenerateRequest(**kwargs)
|
||
with pytest.raises(HTTPException) as e:
|
||
mapserve._resolve_request(req)
|
||
assert e.value.status_code == 400
|
||
# Tri nord → sud (row décroissante) puis col croissante
|
||
req = mapserve.GenerateRequest(tiles=[[1055, 6882], [1054, 6883],
|
||
[1054, 6882]])
|
||
tiles, _viz = mapserve._resolve_request(req)
|
||
assert tiles == [(1054, 6883), (1054, 6882), (1055, 6882)]
|
||
|
||
|
||
def test_generate_launches_and_queues(tmp_path, monkeypatch):
|
||
mapserve = _gen_setup(tmp_path, monkeypatch)
|
||
launched = []
|
||
monkeypatch.setattr(mapserve, "_launch_job",
|
||
lambda tiles, viz, req, qid=None:
|
||
launched.append((tiles, qid)) or ["fake", "cmd"])
|
||
|
||
req = mapserve.GenerateRequest(tiles=[[1054, 6882]], viz=["aspect"])
|
||
d = mapserve.generate(req)
|
||
assert d["demarré"] is True and d["tuiles"] == 1
|
||
|
||
# Run en cours : la demande part en file (jamais de coupure du travail)
|
||
import threading
|
||
with mapserve._job_lock:
|
||
mapserve._job["proc"] = types.SimpleNamespace(
|
||
poll=lambda: None, pid=4242, terminate=lambda: None, wait=lambda: None)
|
||
d = mapserve.generate(req)
|
||
assert d["demarré"] is False and d["en_file"] == 1
|
||
assert mapserve.generate(req)["en_file"] == 2
|
||
assert len(mapserve._queue_summary()) == 2
|
||
d = mapserve.queue_clear()
|
||
assert d["retirées"] == 2
|
||
with mapserve._job_lock:
|
||
mapserve._job["proc"] = None
|
||
mapserve._queue.clear()
|
||
|
||
|
||
def test_status_reports_rights_and_availability(tmp_path, monkeypatch):
|
||
"""/api/status annonce available (pipeline local) et regen_allowed (IP)."""
|
||
mapserve = _gen_setup(tmp_path, monkeypatch)
|
||
monkeypatch.setattr(mapserve, "_pipeline_import_cache", {"ok": True})
|
||
d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
|
||
assert d["available"] is True and d["regen_allowed"] is True
|
||
assert d["running"] is False and d["tiles"] == []
|
||
|
||
# Client hors réseau local : génération interdite (boutons masqués)
|
||
d = mapserve.generation_status(_ClientRequest("203.0.113.7"))
|
||
assert d["regen_allowed"] is False
|
||
|
||
# X-Forwarded-For honoré seulement depuis un pair de confiance
|
||
d = mapserve.generation_status(
|
||
_ClientRequest("172.25.0.1", xff="203.0.113.7"))
|
||
assert d["regen_allowed"] is False
|
||
d = mapserve.generation_status(
|
||
_ClientRequest("172.25.0.1", xff="192.168.3.10"))
|
||
assert d["regen_allowed"] is True
|
||
|
||
# Image légère sans pipeline ni LIDAR_GENERATION_URL : indisponible
|
||
monkeypatch.setattr(mapserve, "_pipeline_import_cache", {"ok": False})
|
||
d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
|
||
assert d["available"] is False
|
||
|
||
|
||
def test_status_proxies_to_worker(tmp_path, monkeypatch):
|
||
"""Webapp légère : /api/status relaie l'état du worker, 503 hors ligne."""
|
||
mapserve = _gen_setup(tmp_path, monkeypatch)
|
||
monkeypatch.setattr(mapserve, "GENERATION_URL", "http://worker:8973")
|
||
monkeypatch.setattr(mapserve, "_pipeline_import_cache", {"ok": False})
|
||
|
||
def fake_proxy(method, path, payload=None, timeout=30):
|
||
assert (method, path) == ("GET", "/api/status")
|
||
return {"running": True, "tiles": []}
|
||
|
||
monkeypatch.setattr(mapserve, "_proxy_generation", fake_proxy)
|
||
d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
|
||
assert d["distant"] is True and d["running"] is True
|
||
assert d["available"] is True and d["regen_allowed"] is True
|
||
|
||
def offline(method, path, payload=None, timeout=30):
|
||
raise HTTPException(503, "machine de traitement injoignable")
|
||
|
||
monkeypatch.setattr(mapserve, "_proxy_generation", offline)
|
||
d = mapserve.generation_status(_ClientRequest("192.168.3.10"))
|
||
assert d["worker_offline"] is True and d["running"] is None
|
||
|
||
|
||
def test_generate_requires_lan(tmp_path, monkeypatch):
|
||
"""Un client hors LIDAR_REGEN_CIDR ne peut pas lancer de génération."""
|
||
mapserve = _gen_setup(tmp_path, monkeypatch)
|
||
with pytest.raises(HTTPException) as e:
|
||
mapserve._require_lan_for_generation(_ClientRequest("203.0.113.7"))
|
||
assert e.value.status_code == 403
|
||
|
||
|
||
def test_cell_at_point(tmp_path, monkeypatch):
|
||
"""/api/cell donne la dalle sous un point (sélection manuelle)."""
|
||
mapserve = _gen_setup(tmp_path, monkeypatch)
|
||
from lidar_pipeline.tiles import _transformer
|
||
lon, lat = _transformer("EPSG:2154", "EPSG:4326").transform(1054500, 6881500)
|
||
d = mapserve.cell_at_point(lat=lat, lng=lon)
|
||
assert (d["col"], d["row"]) == (1054, 6882)
|
||
assert d["corners"] and len(d["corners"]) == 4
|
||
|
||
|
||
def test_tiles_inventory_endpoint(tmp_path, monkeypatch):
|
||
"""/api/tiles sert index_tiles.json (machines légères LIDAR_SOURCE_URL)."""
|
||
mapserve = _gen_setup(tmp_path, monkeypatch)
|
||
payload = {"tiles": [{"col": 1054, "row": 6882,
|
||
"viz": {"aspect": {"full": "visualisations/x.avif?v=1"}}}],
|
||
"viz_meta": {}, "stats": {"n_tiles": 1}}
|
||
(tmp_path / "index_tiles.json").write_text(
|
||
__import__("json").dumps(payload), encoding="utf-8")
|
||
d = mapserve.tiles_inventory()
|
||
assert d["tiles"] == payload["tiles"] and d["stamp"] > 0
|
||
# ?stamp inchangé → réponse allégée
|
||
light = mapserve.tiles_inventory(stamp=d["stamp"])
|
||
assert light["tiles"] is None
|
||
|
||
|
||
def test_generation_always_ign_with_edge_buffer(tmp_path, monkeypatch):
|
||
"""Classification IGN (sol) et raccord des bords de 100 m imposés, quel
|
||
que soit ce que le client envoie."""
|
||
mapserve = _gen_setup(tmp_path, monkeypatch)
|
||
cmd = mapserve._build_command([(1054, 6882)])
|
||
i = cmd.index("--edge-buffer")
|
||
assert cmd[i + 1] == "100"
|
||
assert cmd[cmd.index("--ground-classification") + 1] == "ign"
|
||
assert cmd[cmd.index("--ign-classes") + 1] == "sol"
|
||
req = mapserve.GenerateRequest(tiles=[[1054, 6882]], ground_class="csf", edge_buffer=False)
|
||
assert not hasattr(req, "ground_class") or getattr(req, "ground_class", None) != "csf"
|
||
|
||
|
||
def test_ui_offers_generation_controls():
|
||
"""L'interface embarque les contrôles de génération (＋ Zone, options, run)."""
|
||
from lidar_pipeline.mapui import _MAP_CSS, _MAP_HTML, _MAP_JS
|
||
assert 'id="btnZone"' in _MAP_HTML and 'id="btnComplete"' in _MAP_HTML
|
||
assert 'id="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 et raccord des bords imposés : plus de réglage
|
||
for gone in ('id="genClass"', 'id="genIgnClasses"', 'id="genReclass"', 'id="genEdge"'):
|
||
assert gone not in _MAP_HTML
|
||
# Boutons masqués par défaut : révélés seulement si /api/status autorise
|
||
assert "GEN.available && GEN.allowed" in _MAP_JS
|
||
# Dessin de zone + cadres de run + sondage d'état
|
||
assert "genSetDrawing" in _MAP_JS and "pane: 'genDraw'" in _MAP_JS
|
||
assert "gen-frame-running" in _MAP_CSS and "genDrawFrames" in _MAP_JS
|
||
assert "api/status" in _MAP_JS and "api/preview" in _MAP_JS
|
||
assert "api/generate" in _MAP_JS and "api/stop" in _MAP_JS
|
||
assert "all_missing: true" in _MAP_JS
|
||
# Fin de run : rafraîchissement des couches + relance de la maintenance
|
||
assert "genAfterRun" in _MAP_JS and "api/map/background" in _MAP_JS
|
||
|
||
|
||
def test_cache_only_fetches_missing_tile_upstream(tmp_path, monkeypatch):
|
||
"""Cache seule + LIDAR_MAPS_URL : une tuile absente est rapatriee du serveur
|
||
amont (aucun rendu local : les zooms au-dela de la maintenance s'affichent).
|
||
"""
|
||
from lidar_pipeline import tiles as _t
|
||
monkeypatch.setattr(_t, "TILE_EVEN_LEVELS", False) # règle historique : tout est stocké
|
||
monkeypatch.setattr(_t, "TILE_CACHE_MAX_Z", 99)
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
_setup(tmp_path, monkeypatch)
|
||
monkeypatch.setenv("LIDAR_TILE_CACHE_ONLY", "1")
|
||
monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
|
||
monkeypatch.setattr(mapserve, "_UPSTREAM", {"fails": 0, "until": 0.0})
|
||
calls = []
|
||
PNG = bytes([0x89, 0x50, 0x4E, 0x47, 13, 10, 0x1A, 10]) + b'TUILE-AMONT'
|
||
|
||
def fake_fetch(rel_path):
|
||
calls.append(rel_path)
|
||
return PNG
|
||
|
||
monkeypatch.setattr(mapserve, "_fetch_upstream", fake_fetch)
|
||
try:
|
||
z = 17 # jamais couvert par la maintenance (max_z 16 par defaut)
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
resp = asyncio.run(mapserve.tile("aspect", z, x, f"{y}@2x.webp"))
|
||
assert calls == [f"tiles/aspect/{z}/{x}/{y}@2x.webp"]
|
||
assert resp.body == PNG
|
||
assert resp.headers.get("X-Tile-Pending") != "1" # servie, plus en attente
|
||
# Mise en cache : la seconde demande ne repasse pas par l'amont
|
||
asyncio.run(mapserve.tile("aspect", z, x, f"{y}@2x.webp"))
|
||
assert len(calls) == 1
|
||
assert tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "webp").is_file()
|
||
finally:
|
||
monkeypatch.delenv("LIDAR_TILE_CACHE_ONLY")
|
||
|
||
|
||
def test_tile_selection_always_answers(tmp_path, monkeypatch):
|
||
"""La sélection d'une dalle répond toujours, sans appel réseau : dalle
|
||
absente comprise, avec nom, emprise L93 et cadre de sélection."""
|
||
mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
|
||
def boom(*a, **k):
|
||
raise AssertionError("aucun appel IGN dans /api/map/tile")
|
||
monkeypatch.setattr("lidar_pipeline.fetch_ign.tile_ign_metadata", boom)
|
||
d = mapserve.map_tile_info(lat=48.0, lng=2.0) # loin de toute dalle
|
||
assert d["found"] is False and d["name"].startswith("LHD_FXX_")
|
||
assert len(d["corners"]) == 4 and d["l93_bbox"][2] - d["l93_bbox"][0] == 1000
|
||
|
||
|
||
def test_ign_metadata_cached_and_resilient(tmp_path, monkeypatch):
|
||
"""Fiche IGN : un seul appel au catalogue (cache disque) ; catalogue
|
||
injoignable = message, jamais d'exception."""
|
||
mapserve = _setup(tmp_path, monkeypatch)
|
||
calls = []
|
||
def fake(col, row, timeout=10):
|
||
calls.append((col, row))
|
||
return {"found": True, "download_url": "https://data.geopf.fr/x.copc.laz",
|
||
"acquisition_start": "2022-03-05T16:20:51Z"}
|
||
monkeypatch.setattr("lidar_pipeline.fetch_ign.tile_ign_metadata", fake)
|
||
a = mapserve.map_tile_ign(col=1004, row=6832)
|
||
b = mapserve.map_tile_ign(col=1004, row=6832)
|
||
assert a["download_url"].endswith(".copc.laz") and b == a and len(calls) == 1
|
||
def down(col, row, timeout=10):
|
||
raise OSError("réseau coupé")
|
||
monkeypatch.setattr("lidar_pipeline.fetch_ign.tile_ign_metadata", down)
|
||
assert mapserve.map_tile_ign(col=1, row=2)["found"] is False
|
||
|
||
|
||
def test_ui_tile_card_and_wind_rose():
|
||
from lidar_pipeline.mapui import _MAP_HTML, _MAP_JS
|
||
assert 'id="legendBlock"' in _MAP_HTML and 'id="roseCanvas"' in _MAP_HTML
|
||
assert "api/map/ign?col=" in _MAP_JS and "Télécharger le nuage LiDAR" in _MAP_JS
|
||
assert "compass + 90" in _MAP_JS # même repère que le rendu (aspect = boussole + 90°)
|
||
|
||
|
||
def test_fresh_tiles_jump_the_maintenance_queue(monkeypatch):
|
||
"""Tuiles d'une dalle tout juste rendue : en tête de file, devant l'arriéré."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
monkeypatch.setitem(mapserve._bg, "queue", __import__("collections").deque())
|
||
monkeypatch.setitem(mapserve._bg, "queued", set())
|
||
backlog = [("relief_oriente", 16, i, 0) for i in range(5)]
|
||
mapserve._bg_enqueue(backlog)
|
||
fresh = [("relief_oriente", 12, 9, 9), ("relief_oriente", 16, 3, 0)]
|
||
mapserve._bg_enqueue(fresh, front=True)
|
||
q = list(mapserve._bg["queue"])
|
||
assert q[:2] == fresh and len(q) == 6 and q.count(("relief_oriente", 16, 3, 0)) == 1
|
||
|
||
|
||
def test_cache_only_renders_unstored_levels(tmp_path, monkeypatch):
|
||
"""Cache seule (Pi) : un niveau non stocké est rendu à la volée — jamais
|
||
une tuile transparente « en attente » qui n'arriverait jamais."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", True)
|
||
monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 16)
|
||
_setup(tmp_path, monkeypatch)
|
||
monkeypatch.setenv("LIDAR_TILE_CACHE_ONLY", "1")
|
||
z = 17
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
resp = asyncio.run(mapserve.tile("aspect", z, x, f"{y}@2x.avif"))
|
||
assert resp.headers.get("X-Tile-Pending") != "1" and resp.headers["X-Tile-Empty"] == "0"
|
||
assert not tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "avif").exists()
|
||
|
||
|
||
def test_background_scan_only_queues_stored_levels(tmp_path, monkeypatch):
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", True)
|
||
monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 16)
|
||
_setup(tmp_path, monkeypatch)
|
||
monkeypatch.setitem(mapserve._bg, "queue", __import__("collections").deque())
|
||
monkeypatch.setitem(mapserve._bg, "queued", set())
|
||
monkeypatch.setitem(mapserve._bg, "snapshot", None)
|
||
mapserve.background_scan(tmp_path)
|
||
zs = {z for (_l, z, _x, _y) in mapserve._bg["queue"]}
|
||
assert zs and all(tiles.zoom_cached(z, 2) for z in zs)
|
||
assert zs <= {5, 7, 9, 11, 13, 15}
|
||
|
||
|
||
def test_remote_fed_map_keeps_sources_locally(tmp_path, monkeypatch):
|
||
"""Carte alimentée par l'amont : les sources nécessaires sont rapatriées
|
||
et gardées sans attendre de visite ; une source manquée (amont éteint)
|
||
est reprise au scan suivant ; la dalle entière (doublon) n'est pas prise."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
S = tiles._Source
|
||
thumb = S(tmp_path / "t.webp", (0, 0, 1000, 1000), 3.9, url="http://amont/t")
|
||
quads = [S(tmp_path / f"q{i}.avif", (0, 0, 500, 500), 0.2, url=f"http://amont/q{i}") for i in range(4)]
|
||
full = S(tmp_path / "d.avif", (0, 0, 1000, 1000), 0.2, url="http://amont/d")
|
||
assert tiles.sources_to_keep([[thumb], quads, [full]]) == [thumb] + quads
|
||
monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont")
|
||
monkeypatch.setattr(mapserve, "OUTPUT_DIR", tmp_path)
|
||
monkeypatch.setattr(tiles, "source_index", lambda *a, **k: {"relief_oriente": {(1, 2): [[thumb], quads, [full]]}})
|
||
for key, val in (("queue", __import__("collections").deque()), ("queued", set()),
|
||
("snapshot", None), ("prefetch", __import__("collections").deque())):
|
||
monkeypatch.setitem(mapserve._bg, key, val)
|
||
online = {"up": False}
|
||
def fake_fetch(url, dest):
|
||
if not online["up"]:
|
||
return False
|
||
dest.write_bytes(b"x"); return True
|
||
monkeypatch.setattr(tiles, "_fetch_source", fake_fetch)
|
||
mapserve.background_scan(tmp_path)
|
||
while mapserve._bg["prefetch"]:
|
||
mapserve._bg_process_one()
|
||
assert not any(s_.path.exists() for s_ in [thumb] + quads) # amont éteint
|
||
online["up"] = True
|
||
mapserve.background_scan(tmp_path) # dalle inchangée : reprise quand même
|
||
while mapserve._bg["prefetch"]:
|
||
mapserve._bg_process_one()
|
||
assert all(s_.path.exists() for s_ in [thumb] + quads) and not full.path.exists()
|
||
|
||
|
||
def _bg_reset(monkeypatch, mapserve):
|
||
import collections
|
||
for key, val in (("queue", collections.deque()), ("queued", set()),
|
||
("snapshot", None), ("prefetch", collections.deque()),
|
||
("stats", {"rendues": 0, "a_jour": 0, "vides": 0, "scans": 0,
|
||
"dernier_scan": None, "ajouts_dernier_scan": 0})):
|
||
monkeypatch.setitem(mapserve._bg, key, val)
|
||
|
||
|
||
def test_background_downloads_tiles_from_upstream(tmp_path, monkeypatch):
|
||
"""Maintenance de fond + LIDAR_MAPS_URL : les tuiles de la pyramide sont
|
||
TÉLÉCHARGÉES en tâche de fond (pas seulement à l'affichage), sans rendu
|
||
local — la carte se remplit même sans visite et reste servie ensuite
|
||
sans l'amont."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
_setup(tmp_path, monkeypatch)
|
||
_bg_reset(monkeypatch, mapserve)
|
||
monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
|
||
AVIF = b"\x00\x00\x00\x20ftypavif-TUILE-AMONT"
|
||
calls = []
|
||
monkeypatch.setattr(mapserve, "_fetch_upstream",
|
||
lambda rel: calls.append(rel) or AVIF)
|
||
|
||
def no_render(*a, **k):
|
||
raise AssertionError("rendu local inutile : la tuile vient de l'amont")
|
||
|
||
monkeypatch.setattr(tiles, "get_tile", no_render)
|
||
z = 13
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
mapserve._bg_enqueue([("aspect", z, x, y)])
|
||
assert mapserve._bg_process_one() == "telechargee"
|
||
assert calls == [f"tiles/aspect/{z}/{x}/{y}@2x.avif"]
|
||
cache = tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "avif")
|
||
assert cache.read_bytes() == AVIF
|
||
assert mapserve._bg["stats"]["telechargees"] == 1
|
||
# Fraîche : servie en cache seule sans amont ni rendu
|
||
data, state = tiles.cached_tile(tmp_path, "aspect", z, x, y, 2, "avif")
|
||
assert state == "fresh" and data == AVIF
|
||
|
||
|
||
def test_background_downloads_without_pause(monkeypatch):
|
||
"""Tuiles téléchargées de l'amont enchaînées sans pause ; rendu local
|
||
et rapatriement de source gardent la pause."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
results = iter(["telechargee", "telechargee", True])
|
||
waits = []
|
||
|
||
class Stop:
|
||
def is_set(self):
|
||
return len(waits) >= 1
|
||
|
||
def wait(self, delay):
|
||
waits.append(delay)
|
||
|
||
monkeypatch.setattr(mapserve, "_bg", {**mapserve._bg, "stop": Stop()})
|
||
monkeypatch.setattr(mapserve, "_bg_process_one", lambda: next(results))
|
||
monkeypatch.setenv("LIDAR_TILE_BACKGROUND_PAUSE", "0.2")
|
||
monkeypatch.setattr(mapserve.os, "nice", lambda n: 0)
|
||
mapserve._bg_worker()
|
||
assert waits == [0.2]
|
||
|
||
|
||
def test_background_renders_locally_when_upstream_down(tmp_path, monkeypatch):
|
||
"""Amont éteint : la maintenance rend la tuile localement (autonomie)."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
_setup(tmp_path, monkeypatch)
|
||
_bg_reset(monkeypatch, mapserve)
|
||
monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
|
||
monkeypatch.setattr(mapserve, "_fetch_upstream", lambda rel: None)
|
||
z = 13
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
mapserve._bg_enqueue([("aspect", z, x, y)])
|
||
assert mapserve._bg_process_one() is True
|
||
assert tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "avif").is_file()
|
||
assert mapserve._bg["stats"]["rendues"] == 1
|
||
|
||
|
||
def test_upstream_breaker_rearms_after_expiry(monkeypatch):
|
||
"""La suspension expirée, de nouveaux échecs la relancent (sinon chaque
|
||
tuile repaie le délai réseau d'un amont éteint, indéfiniment)."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
monkeypatch.setattr(mapserve, "_UPSTREAM", {"fails": 0, "until": 0.0})
|
||
for _ in range(3):
|
||
mapserve._upstream_mark(False)
|
||
assert mapserve._upstream_offline() is True
|
||
mapserve._UPSTREAM["until"] = 1.0 # suspension expirée
|
||
assert mapserve._upstream_offline() is False
|
||
mapserve._upstream_mark(False) # nouvel échec : resuspendu
|
||
assert mapserve._upstream_offline() is True
|
||
|
||
|
||
def test_empty_tile_skips_upstream_when_inventory_is_upstream(tmp_path, monkeypatch):
|
||
"""Inventaire venu de l'amont (LIDAR_SOURCE_URL) et aucune dalle sous la
|
||
tuile : l'amont n'en sait pas plus — pas de requête (20 s par tuile vide
|
||
quand il est éteint)."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
_setup(tmp_path, monkeypatch)
|
||
monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
|
||
monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
|
||
monkeypatch.setattr(tiles, "_contributing", lambda *a, **k: [])
|
||
calls = []
|
||
monkeypatch.setattr(mapserve, "_fetch_upstream", lambda rel: calls.append(rel))
|
||
resp = asyncio.run(mapserve.tile("aspect", 15, 1, "1.png"))
|
||
assert calls == []
|
||
assert resp.headers["X-Tile-Empty"] == "1"
|
||
|
||
|
||
def test_cache_only_serves_stale_tile_while_pending(tmp_path, monkeypatch):
|
||
"""Cache seule, tuile périmée (dalle régénérée) et amont muet : l'ancienne
|
||
tuile reste affichée en attendant la nouvelle — pas un trou transparent."""
|
||
import os
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
_setup(tmp_path, monkeypatch)
|
||
monkeypatch.setenv("LIDAR_TILE_CACHE_ONLY", "1")
|
||
monkeypatch.setattr(mapserve, "MAPS_URL", "")
|
||
z = 13
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
cache = tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "avif")
|
||
cache.parent.mkdir(parents=True, exist_ok=True)
|
||
cache.write_bytes(b"ANCIENNE-TUILE")
|
||
os.utime(cache, (1_000_000, 1_000_000)) # plus vieille que la dalle
|
||
assert tiles.cached_tile(tmp_path, "aspect", z, x, y, 2, "avif")[1] == "pending"
|
||
resp = asyncio.run(mapserve.tile("aspect", z, x, f"{y}@2x.avif"))
|
||
assert resp.body == b"ANCIENNE-TUILE"
|
||
assert resp.headers["Cache-Control"] == "no-store" # redemandée plus tard
|
||
assert resp.headers["X-Tile-Pending"] == "1"
|
||
assert resp.headers["X-Tile-Empty"] == "0"
|
||
|
||
|
||
def test_background_tiles_before_sources(tmp_path, monkeypatch):
|
||
"""Tuiles d'abord (légères, visibles tout de suite), sources ensuite :
|
||
des milliers de sources à rapatrier ne retardent pas la pyramide."""
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
_setup(tmp_path, monkeypatch)
|
||
_bg_reset(monkeypatch, mapserve)
|
||
monkeypatch.setattr(mapserve, "MAPS_URL", "http://amont:8975")
|
||
monkeypatch.setattr(mapserve, "_fetch_upstream", lambda rel: b"TUILE")
|
||
fetched = []
|
||
|
||
class Src:
|
||
def ensure(self):
|
||
fetched.append(1)
|
||
return True
|
||
|
||
mapserve._bg["prefetch"].append(Src())
|
||
z = 13
|
||
x, y = _tile_of_cell(1054, 6882, z)
|
||
mapserve._bg_enqueue([("aspect", z, x, y)])
|
||
assert mapserve._bg_process_one() == "telechargee"
|
||
assert fetched == [] and mapserve._bg["stats"]["telechargees"] == 1
|
||
assert mapserve._bg_process_one() is True
|
||
assert fetched == [1]
|
||
|
||
|
||
def test_zoom_animation_transforms_only_on_all_tile_layers():
|
||
"""Pendant l'animation de zoom, AUCUNE couche ne crée de niveau : un
|
||
niveau neuf n'est pas composé par Chrome (animé sur le thread principal,
|
||
gelé ~90 ms par le chargement) — le fond sautait quand le relief orienté,
|
||
qui garde son niveau pair, glissait normalement : désynchronisés."""
|
||
import re
|
||
from lidar_pipeline.mapui import _MAP_JS
|
||
assert "_animateZoom(e) { this._setView(e.center, e.zoom, true, true); }" in _MAP_JS
|
||
# « zoom » part aussi au début de l'animation (Leaflet 1.9) : même garde
|
||
assert "if (this._map && this._map._animatingZoom) {" in _MAP_JS
|
||
# Sans quoi _setView recrée un niveau malgré noUpdate (tileZoomChanged)
|
||
assert _MAP_JS.count("updateWhenZooming: false") >= 2
|
||
# Le fond OSM ET les couches LiDAR héritent du comportement
|
||
assert re.search(r"const base = new ZoomAnimTileLayer\(", _MAP_JS)
|
||
assert "const EvenLevelTileLayer = ZoomAnimTileLayer.extend(" in _MAP_JS
|
||
assert "L.tileLayer(" not in _MAP_JS.replace("L.tileLayer` standard", "")
|
||
|
||
|
||
def test_default_policy_pregenerates_every_level_up_to_native(monkeypatch):
|
||
"""Par défaut, toute la pyramide jusqu'au natif (18@2x = 0,2 m/px) est
|
||
stockée et générée d'avance : aucun niveau rendu à la volée."""
|
||
import importlib
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
for name in ("LIDAR_TILE_CACHE_MAX_Z", "LIDAR_TILE_EVEN_LEVELS",
|
||
"LIDAR_TILE_BACKGROUND_MAX_Z", "LIDAR_TILE_BACKGROUND_SCALE"):
|
||
monkeypatch.delenv(name, raising=False)
|
||
try:
|
||
importlib.reload(tiles)
|
||
assert all(tiles.zoom_cached(z, 2) for z in range(tiles.TILE_MIN_Z, 19))
|
||
assert all(tiles.zoom_cached(z, 1) for z in range(tiles.TILE_MIN_Z, 20))
|
||
assert mapserve._bg_conf()["max_z"] == 18 and mapserve._bg_conf()["scale"] == 2
|
||
monkeypatch.setenv("LIDAR_TILE_BACKGROUND_SCALE", "1")
|
||
assert mapserve._bg_conf()["max_z"] == 19
|
||
finally:
|
||
importlib.reload(tiles)
|
||
|
||
|
||
def test_background_scan_retries_cells_beyond_queue_max(tmp_path, monkeypatch):
|
||
"""File pleine : les dalles dont la pyramide n'a pas trouvé place sont
|
||
reprises aux scans suivants jusqu'à ce que toutes leurs tuiles passent."""
|
||
import collections
|
||
import lidar_pipeline.mapserve as mapserve
|
||
from lidar_pipeline import tiles
|
||
monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", False)
|
||
monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 19)
|
||
_setup(tmp_path, monkeypatch)
|
||
_make_dalle(tmp_path, 1055, 6882, ["aspect"])
|
||
conf = dict(mapserve._bg_conf(), max_z=18, scale=2, queue_max=40)
|
||
monkeypatch.setattr(mapserve, "_bg_conf", lambda: conf)
|
||
for key, val in (("queue", collections.deque()), ("queued", set()),
|
||
("snapshot", None), ("prefetch", collections.deque()),
|
||
("incomplete", {})):
|
||
monkeypatch.setitem(mapserve._bg, key, val)
|
||
expected = set()
|
||
for layer, per_cell in tiles.source_index(tmp_path, force=True).items():
|
||
for col, row in per_cell:
|
||
bbox = tiles.cell_bounds_wgs84(col, row)
|
||
for z in range(tiles.TILE_MIN_Z, 19):
|
||
expected.update((layer, z, x, y) for x, y in tiles.tiles_in_bounds(bbox, z))
|
||
assert len(expected) > 3 * conf["queue_max"]
|
||
seen = set()
|
||
for _ in range(100):
|
||
mapserve.background_scan(tmp_path)
|
||
seen.update(mapserve._bg["queue"])
|
||
mapserve._bg["queue"].clear() # file vidée par les rendeurs
|
||
mapserve._bg["queued"].clear()
|
||
if not mapserve._bg["incomplete"]:
|
||
break
|
||
assert seen == expected
|
||
# Pyramide passée en entier : les scans suivants se taisent
|
||
assert mapserve.background_scan(tmp_path) == 0
|
||
|
||
|
||
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):
|
||
"""Carte légère avec générateur distant : l'export reste local."""
|
||
mapserve = _setup(tmp_path, monkeypatch, layers=("relief_oriente",))
|
||
monkeypatch.setattr(mapserve, "GENERATION_URL", "http://injoignable.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
|
||
# cadre posé sur le terrain : il ne suit plus la vue, on le fait glisser
|
||
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_
|
||
# onglet toujours proposé (indépendant de /api/status.available) ; plus de
|
||
# bouton ⎙ ni de carte hors du panneau
|
||
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
|
||
# Palette sombre : système sombre (sauf clair forcé) ou sombre forcé
|
||
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
|
||
# Stockage local toujours protégé (navigation privée) : aucun accès
|
||
# direct à localStorage hors du bloc lsGet/lsSet/lsDel.
|
||
import re
|
||
start = _MAP_JS.index("function lsGet(")
|
||
end = _MAP_JS.index("// --- fin stockage")
|
||
outside = _MAP_JS[:start] + _MAP_JS[end:]
|
||
assert not re.search(r"localStorage\.(getItem|setItem|removeItem)\(", outside)
|
||
# Code mort retiré
|
||
for gone in ("genClass", "gen-input", "genClassInfo", ".grip", "layer-row", "#gps.busy"):
|
||
assert gone not in _MAP_CSS + _MAP_HTML, gone
|
||
# Plus de couleurs d'interface écrites en dur hors des 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-print", "tab-gen", "tab-share"):
|
||
assert f'id="{id_}"' in _MAP_HTML, id_
|
||
for tab in ("view", "print", "gen", "share"):
|
||
assert f'data-tab="{tab}"' in _MAP_HTML, tab
|
||
# Plus d'éléments flottants : tout passe dans le panneau
|
||
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
|
||
# Réglages et boutons déplacés, identifiants conservés
|
||
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
|
||
# Génération masquée : onglet retiré de la liste (touches 1–4 comprises)
|
||
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]
|
||
# Onglet choisi gardé à part : Génération rétablie une fois /api/status connu
|
||
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]
|
||
# Repli en bande d'icônes : ordinateur seulement
|
||
assert "PANEL.collapsed && !isPhone()" in render
|
||
assert "toggle('collapsed', PANEL.collapsed)" not in render
|
||
assert "toggle('panel-collapsed', PANEL.collapsed)" not in render
|