0d3aa41008
Au rechargement de la page, l'index obsolète du worker (16 couches) polluait le /api/tiles fusionné et mergeTilesData réajoutait les couches retirées au menu. viz_meta est désormais filtré aux couches du panneau (PANEL_VIZ) côté serveur, et le panneau client est figé au chargement (clés VIZ_META initiales) : aucun flux ne peut réintroduire de couche hors panneau, même avec un index distant périmé.
829 lines
35 KiB
Python
829 lines
35 KiB
Python
"""Tests du serveur web de génération de zones (webapp)."""
|
|
|
|
|
|
def test_bbox_to_cells_single_km_cell():
|
|
"""Une bbox couvrant ~1 km² retourne la cellule L93 correspondante."""
|
|
from lidar_pipeline.webapp import bbox_to_cells
|
|
# Cellule 1054,6882 : X∈[1054000,1055000], Y∈[6881000,6882000] (L93)
|
|
from rasterio.warp import transform as warp_transform
|
|
lons, lats = warp_transform('EPSG:2154', 'EPSG:4326',
|
|
[1054100, 1054900], [6881100, 6881900])
|
|
cells = bbox_to_cells(min(lons), min(lats), max(lons), max(lats))
|
|
assert (1054, 6882) in cells
|
|
# La sélection reste locale : pas de cellule lointaine
|
|
for (c, r) in cells:
|
|
assert abs(c - 1054) <= 1 and abs(r - 6882) <= 1
|
|
|
|
|
|
def test_bbox_to_cells_empty_for_tiny_bbox():
|
|
"""Une bbox quasi ponctuelle ne sélectionne rien (rétrécie sous 1 m)."""
|
|
from lidar_pipeline.webapp import bbox_to_cells
|
|
assert bbox_to_cells(7.850000, 48.930000, 7.850001, 48.930001) == []
|
|
|
|
|
|
def test_processed_cells(tmp_path):
|
|
"""processed_cells lit les dossiers de visualisations."""
|
|
from lidar_pipeline.webapp import processed_cells
|
|
vis = tmp_path / "visualisations" / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_r0p2"
|
|
vis.mkdir(parents=True)
|
|
(vis / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_aspect.avif").write_bytes(b"x")
|
|
assert processed_cells(tmp_path) == {(1054, 6882)}
|
|
|
|
|
|
def test_missing_cells_filters_processed(tmp_path):
|
|
"""Les cellules déjà traitées sont exclues, les autres gardent leurs coins."""
|
|
from lidar_pipeline.webapp import missing_cells_with_corners
|
|
vis = tmp_path / "visualisations" / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
|
|
vis.mkdir(parents=True)
|
|
(vis / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_aspect.avif").write_bytes(b"x")
|
|
todo = missing_cells_with_corners([(1054, 6882), (1055, 6882)], tmp_path)
|
|
assert len(todo) == 1
|
|
assert todo[0]['col'] == 1055 and todo[0]['row'] == 6882
|
|
assert len(todo[0]['corners']) == 4 # SW, SE, NE, NW
|
|
|
|
|
|
def test_missing_cells_include_done(tmp_path):
|
|
"""include_done=True conserve les cellules déjà traitées (régénération)."""
|
|
from lidar_pipeline.webapp import missing_cells_with_corners
|
|
vis = tmp_path / "visualisations" / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
|
|
vis.mkdir(parents=True)
|
|
(vis / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_aspect.avif").write_bytes(b"x")
|
|
todo = missing_cells_with_corners([(1054, 6882), (1055, 6882)], tmp_path,
|
|
include_done=True)
|
|
assert {(t['col'], t['row']) for t in todo} == {(1054, 6882), (1055, 6882)}
|
|
|
|
|
|
def _make_tile(tmp_path, col, row, viz_keys, resolutions=(0.5, 0.2)):
|
|
"""Crée les dossiers de visualisations d'une tuile (une entrée par résolution).
|
|
|
|
Le suffixe de résolution n'est que sur le DOSSIER : les fichiers gardent
|
|
le basename nu (cf. _expected_output_path dans le pipeline).
|
|
"""
|
|
base = f"LHD_FXX_{col:04d}_{row:04d}_PTS_LAMB93_IGN69"
|
|
for res in resolutions:
|
|
suffix = "" if res == 0.5 else "_r" + str(res).replace('.', 'p')
|
|
vis = tmp_path / "visualisations" / (base + suffix)
|
|
vis.mkdir(parents=True, exist_ok=True)
|
|
for key in viz_keys:
|
|
(vis / f"{base}_{key}.avif").write_bytes(b"x")
|
|
|
|
|
|
def test_missing_cells_includes_incomplete_tile(tmp_path):
|
|
"""Une tuile existante mais incomplète reste incluse (visualisation manquante)."""
|
|
from lidar_pipeline.webapp import missing_cells_with_corners
|
|
_make_tile(tmp_path, 1054, 6882, ["aspect"])
|
|
# aspect aux deux résolutions → complète pour ['aspect']
|
|
assert missing_cells_with_corners([(1054, 6882)], tmp_path, viz=["aspect"]) == []
|
|
# slope manquante → la tuile est incluse pour être complétée
|
|
todo = missing_cells_with_corners([(1054, 6882)], tmp_path,
|
|
viz=["aspect", "slope"])
|
|
assert [(t['col'], t['row']) for t in todo] == [(1054, 6882)]
|
|
|
|
|
|
def test_missing_cells_requires_generation_resolution(tmp_path):
|
|
"""Une tuile complète à 0.5 mais sans 0.2 reste à traiter (seule 0.2 est générée)."""
|
|
from lidar_pipeline.webapp import missing_cells_with_corners
|
|
_make_tile(tmp_path, 1054, 6882, ["aspect"], resolutions=(0.5,))
|
|
assert len(missing_cells_with_corners([(1054, 6882)], tmp_path, viz=["aspect"])) == 1
|
|
_make_tile(tmp_path, 1054, 6882, ["aspect"], resolutions=(0.2,))
|
|
assert missing_cells_with_corners([(1054, 6882)], tmp_path, viz=["aspect"]) == []
|
|
|
|
|
|
def test_complete_cells_maps_step_names(tmp_path):
|
|
"""Les noms d'étapes sont convertis en mots-clés de fichiers (hillshade → hillshade_multi)."""
|
|
from lidar_pipeline.webapp import complete_cells
|
|
_make_tile(tmp_path, 1054, 6882, ["hillshade_multi"])
|
|
assert complete_cells(tmp_path, ["hillshade"]) == {(1054, 6882)}
|
|
# 'pos_open' → 'positive_openness' : absent → tuile non complète
|
|
assert complete_cells(tmp_path, ["pos_open"]) == set()
|
|
|
|
|
|
def test_preview_rejects_unknown_viz():
|
|
"""/api/preview refuse un nom de visualisation inconnu (HTTPException 400)."""
|
|
from fastapi import HTTPException
|
|
from lidar_pipeline.webapp import preview, PreviewRequest
|
|
req = PreviewRequest(bbox=[7.85, 48.93, 7.86, 48.94], viz=["inconnu"])
|
|
try:
|
|
preview(req)
|
|
assert False, "une HTTPException était attendue"
|
|
except HTTPException as e:
|
|
assert e.status_code == 400
|
|
assert "inconnu" in e.detail
|
|
|
|
|
|
def test_build_command_resolutions_match_completeness():
|
|
"""La commande -r et la détection de complétude utilisent les mêmes résolutions."""
|
|
from lidar_pipeline.webapp import GENERATE_RESOLUTIONS, _build_command
|
|
cmd = _build_command([(1054, 6882)])
|
|
i = cmd.index("-r")
|
|
assert cmd[i + 1] == ",".join(str(r) for r in GENERATE_RESOLUTIONS)
|
|
|
|
|
|
def test_job_state_persisted_across_reload(tmp_path, monkeypatch):
|
|
"""L'état du job survit au redémarrage du serveur (file .generation.job.json)."""
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "JOB_FILE", tmp_path / "job.json")
|
|
saved = {k: webapp._job.get(k) for k in ("proc", "started", "returncode", "cmd", "finished")}
|
|
try:
|
|
webapp._job.update({"started": 123.0, "returncode": 0,
|
|
"cmd": ["python", "-m", "lidar_pipeline"], "finished": 456.0})
|
|
webapp._save_job_state()
|
|
assert (tmp_path / "job.json").exists()
|
|
# Simulation du redémarrage : état mémoire vide, rechargé depuis disque
|
|
webapp._job.update({"proc": None, "started": None, "returncode": None,
|
|
"cmd": None, "finished": None})
|
|
webapp._load_job_state()
|
|
assert webapp._job["started"] == 123.0
|
|
assert webapp._job["returncode"] == 0
|
|
assert webapp._job["finished"] == 456.0
|
|
finally:
|
|
webapp._job.update(saved)
|
|
|
|
|
|
def test_laz_cells_parses_input(tmp_path):
|
|
"""laz_cells repère les dalles LHD présentes dans input/ (les autres ignorés)."""
|
|
from lidar_pipeline.webapp import laz_cells
|
|
(tmp_path / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69.copc.laz").write_bytes(b"x")
|
|
(tmp_path / "LHD_FXX_1055_6881_PTS_LAMB93_IGN69.laz").write_bytes(b"x")
|
|
(tmp_path / "LHD_FXX_1056_6880_PTS_LAMB93_IGN69.copc.las").write_bytes(b"x")
|
|
(tmp_path / "autre_fichier.laz").write_bytes(b"x")
|
|
assert laz_cells(tmp_path) == [(1054, 6882), (1055, 6881), (1056, 6880)]
|
|
|
|
|
|
def test_preview_all_missing(tmp_path, monkeypatch):
|
|
"""all_missing: seulement les dalles présentes dans input/ et incomplètes."""
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "INPUT_DIR", tmp_path)
|
|
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
|
|
(tmp_path / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69.copc.laz").write_bytes(b"x")
|
|
(tmp_path / "LHD_FXX_1055_6881_PTS_LAMB93_IGN69.copc.laz").write_bytes(b"x")
|
|
# 1054,6882 : aspect aux deux résolutions → complète pour ['aspect']
|
|
base = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
|
|
for suffix in ("", "_r0p2"):
|
|
vis = tmp_path / "visualisations" / (base + suffix)
|
|
vis.mkdir(parents=True)
|
|
(vis / f"{base}_aspect.avif").write_bytes(b"x")
|
|
d = webapp.preview(webapp.PreviewRequest(all_missing=True, viz=["aspect"]))
|
|
assert d["count"] == 1
|
|
assert d["cells"][0]["col"] == 1055 and d["cells"][0]["row"] == 6881
|
|
|
|
|
|
def test_generate_all_missing(tmp_path, monkeypatch):
|
|
"""all_missing: traite toutes les dalles présentes (au-delà de la limite 400)."""
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "INPUT_DIR", tmp_path)
|
|
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
|
|
for i in range(1054, 1455): # 401 dalles → au-delà de MAX_CELLS
|
|
(tmp_path / f"LHD_FXX_{i:04d}_6882_PTS_LAMB93_IGN69.copc.laz").write_bytes(b"x")
|
|
|
|
captured = {}
|
|
|
|
def fake_popen(cmd, **kwargs):
|
|
captured["cmd"] = cmd
|
|
|
|
class _P:
|
|
def wait(self):
|
|
return 0
|
|
|
|
def poll(self):
|
|
return 0
|
|
return _P()
|
|
|
|
# Stub du module subprocess : patcher le vrai module global casserait ses
|
|
# autres usages (matplotlib l'appelle à l'import, ex. fc-list) et la
|
|
# restauration serait un no-op (valeur déjà falsifiée).
|
|
import subprocess as _real_subprocess
|
|
|
|
class _SubprocessStub:
|
|
Popen = staticmethod(fake_popen)
|
|
run = staticmethod(_real_subprocess.run)
|
|
STDOUT = _real_subprocess.STDOUT
|
|
|
|
monkeypatch.setattr(webapp, "subprocess", _SubprocessStub)
|
|
req = webapp.GenerateRequest(tiles=[], all_missing=True, viz=["aspect"])
|
|
result = webapp.generate(req)
|
|
assert result["demarré"] is True and result["tuiles"] == 401
|
|
i = captured["cmd"].index("--file")
|
|
assert len(captured["cmd"][i + 1:]) == 401
|
|
assert "LHD_FXX_1054_6882_PTS_LAMB93_IGN69.copc.laz" in captured["cmd"]
|
|
|
|
|
|
def test_generate_all_missing_empty(tmp_path, monkeypatch):
|
|
"""all_missing sans dalle présente → 400."""
|
|
from fastapi import HTTPException
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "INPUT_DIR", tmp_path)
|
|
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
|
|
try:
|
|
webapp.generate(webapp.GenerateRequest(tiles=[], all_missing=True))
|
|
assert False, "une HTTPException était attendue"
|
|
except HTTPException as e:
|
|
assert e.status_code == 400
|
|
|
|
|
|
def test_build_command_regenerate():
|
|
"""regenerate --force seul (classification conservée) ; reclassify ajoute --force-classification."""
|
|
from lidar_pipeline.webapp import _build_command
|
|
cmd = " ".join(_build_command([(1054, 6882)], regenerate=True))
|
|
assert "--force" in cmd
|
|
assert "--force-classification" not in cmd
|
|
cmd = " ".join(_build_command([(1054, 6882)], regenerate=True, reclassify=True))
|
|
assert "--force" in cmd
|
|
assert "--force-classification" in cmd
|
|
cmd = " ".join(_build_command([(1054, 6882)]))
|
|
assert "--force" not in cmd
|
|
assert "--force-classification" not in cmd
|
|
|
|
|
|
def test_build_command_ground_classification():
|
|
"""La commande utilise la méthode de classification demandée (défaut : ign)."""
|
|
from lidar_pipeline.webapp import _build_command, GROUND_CLASS_METHODS
|
|
# Défaut : ign (pré-classification)
|
|
cmd = _build_command([(1054, 6882)])
|
|
i = cmd.index("--ground-classification")
|
|
assert cmd[i + 1] == "ign"
|
|
# Chaque méthode valide est transmise telle quelle ; --force suit
|
|
# regenerate, --force-classification suit reclassify (indépendants)
|
|
for method in GROUND_CLASS_METHODS:
|
|
for regenerate in (False, True):
|
|
for reclassify in (False, True):
|
|
cmd = _build_command([(1054, 6882)], regenerate=regenerate,
|
|
ground_class=method, reclassify=reclassify)
|
|
i = cmd.index("--ground-classification")
|
|
assert cmd[i + 1] == method
|
|
assert ("--force" in cmd) == regenerate
|
|
assert ("--force-classification" in cmd) == reclassify
|
|
|
|
|
|
def test_stop_generation(monkeypatch):
|
|
"""/api/stop termine le processus en cours (SIGTERM) ou refuse (409)."""
|
|
import time
|
|
from fastapi import HTTPException
|
|
import lidar_pipeline.webapp as webapp
|
|
terminated = []
|
|
|
|
class FakeProc:
|
|
pid = 4242
|
|
def poll(self):
|
|
return None # en cours
|
|
def terminate(self):
|
|
terminated.append("SIGTERM")
|
|
def wait(self, timeout=None):
|
|
terminated.append("wait")
|
|
|
|
saved = {k: webapp._job.get(k) for k in ("proc", "started", "returncode", "cmd", "finished")}
|
|
try:
|
|
webapp._job.update({"proc": FakeProc(), "started": time.time(),
|
|
"returncode": None, "cmd": ["x"], "finished": None})
|
|
assert webapp.stop_generation() == {"arrêt": "demandé"}
|
|
assert "SIGTERM" in terminated
|
|
# Le fil d'escalade peut mettre un instant avant son wait : laisser
|
|
# la file de threads vider avant de restaurer l'état du job.
|
|
for _ in range(50):
|
|
if "wait" in terminated:
|
|
break
|
|
time.sleep(0.02)
|
|
# Sans job en cours : refus propre
|
|
webapp._job.update({"proc": None})
|
|
try:
|
|
webapp.stop_generation()
|
|
assert False, "une HTTPException était attendue"
|
|
except HTTPException as e:
|
|
assert e.status_code == 409
|
|
finally:
|
|
webapp._job.update(saved)
|
|
|
|
|
|
def test_build_command_default_viz_aspect():
|
|
"""Sans choix de visualisation, la commande génère uniquement aspect."""
|
|
from lidar_pipeline.webapp import _build_command
|
|
cmd = _build_command([(1054, 6882)])
|
|
i = cmd.index("--only")
|
|
assert cmd[i + 1] == "aspect"
|
|
|
|
|
|
def test_build_command_custom_viz():
|
|
"""Les visualisations demandées sont passées à --only dans l'ordre."""
|
|
from lidar_pipeline.webapp import _build_command
|
|
cmd = _build_command([(1054, 6882)], viz=["aspect", "wavelet", "slope"])
|
|
i = cmd.index("--only")
|
|
assert cmd[i + 1:i + 4] == ["aspect", "wavelet", "slope"]
|
|
|
|
|
|
def test_build_command_incremental_index():
|
|
"""La génération web reconstruit l'index après chaque tuile (affichage en direct)."""
|
|
from lidar_pipeline.webapp import _build_command
|
|
cmd = _build_command([(1054, 6882)])
|
|
assert "--incremental-index" in cmd
|
|
|
|
|
|
def test_tiles_endpoint_stamp_diffing(tmp_path, monkeypatch):
|
|
"""/api/tiles sert index_tiles.json, allégé quand le stamp correspond."""
|
|
import json
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
|
|
|
|
# Fichier absent (aucune tuile encore indexée) → réponse vide
|
|
assert webapp.tiles_data() == {"stamp": None, "tiles": None}
|
|
|
|
# Fichier présent → données complètes puis réponse allégée au stamp
|
|
(tmp_path / "index_tiles.json").write_text(
|
|
json.dumps({"tiles": [{"col": 1054, "corners": []}],
|
|
"viz_meta": {}, "stats": {}}), encoding="utf-8")
|
|
full = webapp.tiles_data()
|
|
assert full["tiles"][0]["col"] == 1054
|
|
light = webapp.tiles_data(stamp=full["stamp"])
|
|
assert light["tiles"] is None and light["stamp"] == full["stamp"]
|
|
# Un stamp périmé redonne les données complètes
|
|
assert webapp.tiles_data(stamp=full["stamp"] - 10)["tiles"] is not None
|
|
|
|
|
|
def test_tiles_endpoint_merges_remote_index(tmp_path, monkeypatch):
|
|
"""L'index du worker prime ; le cache local complète les positions perdues.
|
|
|
|
viz_meta est filtré aux couches du panneau (PANEL_VIZ) : une couche
|
|
retirée (absente de PANEL_VIZ) proposée par un index obsolète n'est
|
|
pas réintroduite.
|
|
"""
|
|
import json
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
|
|
monkeypatch.setattr(webapp, "GENERATION_URL", "http://distant:8973")
|
|
(tmp_path / "index_tiles.json").write_text(json.dumps({
|
|
"tiles": [
|
|
{"col": 1054, "row": 6882, "resolution": 0.2, "viz": {}}, # commun
|
|
{"col": 1050, "row": 6882, "resolution": 0.2, "viz": {}}, # local seul
|
|
],
|
|
"viz_meta": {
|
|
"aspect": {"label": "Aspect local"},
|
|
"slope": {"label": "Pente locale"},
|
|
},
|
|
"stats": {},
|
|
}), encoding="utf-8")
|
|
import os
|
|
remote_stamp = os.stat(tmp_path / "index_tiles.json").st_mtime + 1000
|
|
monkeypatch.setattr(webapp, "_remote_tiles_data", lambda: {
|
|
"stamp": remote_stamp,
|
|
"tiles": [
|
|
# 1054 régénéré côté worker : sa version (viz remplie) doit primer
|
|
{"col": 1054, "row": 6882, "resolution": 0.2, "viz": {"slope": {}}},
|
|
{"col": 1055, "row": 6882, "resolution": 0.2, "viz": {}}, # distant seul
|
|
],
|
|
"viz_meta": {
|
|
"slope": {"label": "Pente"},
|
|
"wavelet": {"label": "Ondulets"}, # hors panneau : filtrée
|
|
},
|
|
})
|
|
d = webapp.tiles_data()
|
|
entries = {(t["col"], json.dumps(t["viz"], sort_keys=True)) for t in d["tiles"]}
|
|
assert (1054, '{"slope": {}}') in entries # version distante retenue
|
|
assert (1055, '{}') in entries # tuile non cachée visible
|
|
assert (1050, '{}') in entries # historique local conservé
|
|
assert d["viz_meta"]["slope"]["label"] == "Pente" # distant prime
|
|
assert d["viz_meta"]["aspect"]["label"] == "Aspect local" # local complété
|
|
assert "wavelet" not in d["viz_meta"] # couche hors panneau filtrée
|
|
assert d["stamp"] == remote_stamp # max des deux stamps
|
|
|
|
|
|
def test_tiles_endpoint_worker_offline_serves_local(tmp_path, monkeypatch):
|
|
"""Worker injoignable : index local servi seul, sans erreur."""
|
|
import json
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
|
|
monkeypatch.setattr(webapp, "GENERATION_URL", "http://distant:8973")
|
|
(tmp_path / "index_tiles.json").write_text(
|
|
json.dumps({"tiles": [{"col": 1054}], "viz_meta": {}, "stats": {}}),
|
|
encoding="utf-8")
|
|
monkeypatch.setattr(webapp, "_remote_tiles_data", lambda: None)
|
|
d = webapp.tiles_data()
|
|
assert d["tiles"][0]["col"] == 1054
|
|
assert d["stamp"] is not None
|
|
|
|
|
|
def test_worker_offline_breaker(monkeypatch):
|
|
"""3 échecs consécutifs → tentatives suspendues ; un succès réarme."""
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "_WORKER_HEALTH", {"fails": 0, "until": 0.0})
|
|
webapp._worker_mark(False)
|
|
webapp._worker_mark(False)
|
|
assert webapp._worker_offline() is False # pas encore armé
|
|
webapp._worker_mark(False)
|
|
assert webapp._worker_offline() is True # disjoncteur armé
|
|
until = webapp._WORKER_HEALTH["until"]
|
|
webapp._worker_mark(False)
|
|
assert webapp._WORKER_HEALTH["until"] == until # fenêtre non étendue
|
|
webapp._worker_mark(True)
|
|
assert webapp._worker_offline() is False # réarmé immédiatement
|
|
|
|
|
|
def test_safe_rel_path():
|
|
"""Seuls les chemins images des préfixes autorisés sont rapatrieables."""
|
|
from lidar_pipeline.webapp import _safe_rel_path
|
|
assert _safe_rel_path("/visualisations/dalle/img.avif") == \
|
|
"visualisations/dalle/img.avif"
|
|
assert _safe_rel_path("/index_thumbs/a_b.jpg") == "index_thumbs/a_b.jpg"
|
|
assert _safe_rel_path("/index_subtiles/x_0_1.avif") == "index_subtiles/x_0_1.avif"
|
|
assert _safe_rel_path("/DTM/x.tif") is None # préfixe interdit
|
|
assert _safe_rel_path("/assets/app.js") is None # préfixe interdit
|
|
assert _safe_rel_path("/visualisations/../x.avif") is None # traversal
|
|
assert _safe_rel_path("/visualisations/a%20b.avif") is None # caractères
|
|
|
|
|
|
def test_fetch_remote_to_cache(tmp_path, monkeypatch):
|
|
"""Rapatriement atomique depuis le worker ; échec sans fichier ni erreur."""
|
|
import io
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
|
|
monkeypatch.setattr(webapp, "GENERATION_URL", "http://distant:8973")
|
|
monkeypatch.setattr(webapp, "_WORKER_HEALTH", {"fails": 0, "until": 0.0})
|
|
|
|
class _Resp(io.BytesIO):
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
captured = {}
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
captured["url"] = req.full_url
|
|
return _Resp(b"IMAGEDATA")
|
|
|
|
monkeypatch.setattr(webapp.urllib.request, "urlopen", fake_urlopen)
|
|
assert webapp._fetch_remote_to_cache("visualisations/dalle/img.avif") is True
|
|
p = tmp_path / "visualisations" / "dalle" / "img.avif"
|
|
assert p.read_bytes() == b"IMAGEDATA"
|
|
assert captured["url"] == "http://distant:8973/visualisations/dalle/img.avif"
|
|
assert not list(tmp_path.rglob("*.part")) # pas de résidu temporaire
|
|
assert webapp._WORKER_HEALTH["fails"] == 0 # succès → disjoncteur sain
|
|
|
|
def boom(req, timeout=None):
|
|
raise OSError("injoignable")
|
|
|
|
monkeypatch.setattr(webapp.urllib.request, "urlopen", boom)
|
|
assert webapp._fetch_remote_to_cache("visualisations/dalle/x.avif") is False
|
|
assert not (tmp_path / "visualisations" / "dalle" / "x.avif").exists()
|
|
assert webapp._WORKER_HEALTH["fails"] == 1
|
|
|
|
|
|
def test_status_worker_offline_no_error(monkeypatch):
|
|
"""Worker hors ligne : running=null (ignoré par l'UI), pas de 503."""
|
|
from fastapi import HTTPException
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "GENERATION_URL", "http://distant:8973")
|
|
|
|
def unreachable(method, path, payload=None, timeout=30):
|
|
raise HTTPException(503, "machine de traitement injoignable")
|
|
|
|
monkeypatch.setattr(webapp, "_proxy_api", unreachable)
|
|
d = webapp.status(_FakeRequest(host="192.168.1.7"))
|
|
assert d["running"] is None
|
|
assert d["worker_offline"] is True
|
|
assert d["regen_allowed"] is True
|
|
|
|
|
|
def test_viz_step_names_match_pipeline():
|
|
"""Les noms acceptés par l'API sont exactement les étapes VIZ_STEPS du pipeline."""
|
|
from lidar_pipeline.webapp import _viz_step_names
|
|
from lidar_pipeline.pipeline import VIZ_STEPS
|
|
assert _viz_step_names() == [n for n, _ in VIZ_STEPS]
|
|
|
|
|
|
def test_generate_rejects_unknown_viz():
|
|
"""L'API refuse un nom de visualisation inconnu (HTTPException 400)."""
|
|
from fastapi import HTTPException
|
|
from lidar_pipeline.webapp import generate, GenerateRequest
|
|
req = GenerateRequest(tiles=[[1054, 6882]], viz=["wavelet", "inconnu"])
|
|
try:
|
|
generate(req)
|
|
assert False, "une HTTPException était attendue"
|
|
except HTTPException as e:
|
|
assert e.status_code == 400
|
|
assert "inconnu" in e.detail
|
|
|
|
|
|
def test_generate_accepts_wavelet_and_slope(monkeypatch):
|
|
"""L'API accepte wavelet + slope et les transmet à --only."""
|
|
import lidar_pipeline.webapp as webapp
|
|
captured = {}
|
|
|
|
def fake_popen(cmd, **kwargs):
|
|
captured["cmd"] = cmd
|
|
|
|
class _P:
|
|
def wait(self):
|
|
return 0
|
|
|
|
def poll(self):
|
|
return 0
|
|
return _P()
|
|
|
|
# Stub du module subprocess (voir test_generate_all_missing) : ne pas
|
|
# patcher le vrai module global.
|
|
import subprocess as _real_subprocess
|
|
|
|
class _SubprocessStub:
|
|
Popen = staticmethod(fake_popen)
|
|
run = staticmethod(_real_subprocess.run)
|
|
STDOUT = _real_subprocess.STDOUT
|
|
|
|
monkeypatch.setattr(webapp, "subprocess", _SubprocessStub)
|
|
req = webapp.GenerateRequest(tiles=[[1054, 6882]],
|
|
viz=["wavelet", "slope"])
|
|
result = webapp.generate(req)
|
|
assert result["demarré"] is True
|
|
i = captured["cmd"].index("--only")
|
|
assert captured["cmd"][i + 1:i + 3] == ["wavelet", "slope"]
|
|
|
|
|
|
def test_available_layers_scans_disk(tmp_path, monkeypatch):
|
|
"""/api/layers retourne les couches présentes (clé → label)."""
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
|
|
vis = tmp_path / "visualisations" / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
|
|
vis.mkdir(parents=True)
|
|
(vis / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_wavelet.avif").write_bytes(b"x")
|
|
layers = webapp.available_layers()
|
|
assert layers.get("wavelet")
|
|
assert "aspect" not in layers
|
|
|
|
|
|
def test_rebuild_index_background(tmp_path, monkeypatch):
|
|
"""POST /api/rebuild lance build_index en arrière-plan puis termine."""
|
|
import time as _time
|
|
import lidar_pipeline.webapp as webapp
|
|
import lidar_pipeline.index as index_mod
|
|
calls = []
|
|
monkeypatch.setattr(index_mod, "build_index", lambda out: calls.append(out))
|
|
assert webapp.rebuild_index()["demarré"] is True
|
|
for _ in range(100):
|
|
if not webapp.rebuild_status()["running"]:
|
|
break
|
|
_time.sleep(0.05)
|
|
assert calls # build_index a bien été appelé
|
|
assert webapp.rebuild_status()["running"] is False
|
|
assert webapp.rebuild_status()["done"] is not None
|
|
|
|
|
|
def test_status_exposes_tiles_from_events(tmp_path, monkeypatch):
|
|
"""/api/status agrège les événements en tuiles (nom court + étapes)."""
|
|
import lidar_pipeline.webapp as webapp
|
|
from lidar_pipeline.progress import report_event, reset_events
|
|
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
|
|
reset_events(tmp_path)
|
|
report_event(tmp_path, "LHD_FXX_1054_6882_PTS_LAMB93_IGN69", "viz",
|
|
"start", "aspect", res=0.5)
|
|
s = webapp.status()
|
|
match = [t for t in s["tiles"] if t["short"] == "1054-6882"]
|
|
assert len(match) == 1
|
|
assert match[0]["state"] == "running"
|
|
assert any(step["label"] == "Aspect" for step in match[0]["steps"])
|
|
|
|
|
|
def test_generate_resets_progress_events(tmp_path, monkeypatch):
|
|
"""Un nouveau run repart d'un journal d'événements vide."""
|
|
import lidar_pipeline.webapp as webapp
|
|
from lidar_pipeline.progress import read_events, report_event
|
|
|
|
def fake_popen(cmd, **kwargs):
|
|
class _P:
|
|
def wait(self):
|
|
return 0
|
|
|
|
def poll(self):
|
|
return 0
|
|
return _P()
|
|
|
|
import subprocess as _real_subprocess
|
|
|
|
class _SubprocessStub:
|
|
Popen = staticmethod(fake_popen)
|
|
run = staticmethod(_real_subprocess.run)
|
|
STDOUT = _real_subprocess.STDOUT
|
|
|
|
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
|
|
monkeypatch.setattr(webapp, "subprocess", _SubprocessStub)
|
|
report_event(tmp_path, "ancienne_tuile", "tile", "ok")
|
|
req = webapp.GenerateRequest(tiles=[[1054, 6882]])
|
|
assert webapp.generate(req)["demarré"] is True
|
|
assert read_events(tmp_path) == []
|
|
|
|
|
|
# ============================================================
|
|
# Mode webapp légère (Raspberry Pi) : génération déléguée à une
|
|
# machine de traitement distante via LIDAR_GENERATION_URL.
|
|
# ============================================================
|
|
|
|
def test_generate_forwards_to_remote(monkeypatch):
|
|
"""/api/generate transmet la demande à la machine de traitement."""
|
|
import lidar_pipeline.webapp as webapp
|
|
calls = []
|
|
|
|
def fake_proxy(method, path, payload=None, timeout=30):
|
|
calls.append((method, path, payload))
|
|
return {"demarré": True, "tuiles": 1}
|
|
|
|
monkeypatch.setattr(webapp, "GENERATION_URL", "http://distant:8973")
|
|
monkeypatch.setattr(webapp, "_proxy_api", fake_proxy)
|
|
req = webapp.GenerateRequest(tiles=[[1054, 6882]], viz=["aspect"])
|
|
result = webapp.generate(req)
|
|
assert result["demarré"] is True and result["distant"] is True
|
|
assert calls == [("POST", "/api/generate",
|
|
{"tiles": [[1054, 6882]], "regenerate": False,
|
|
"reclassify": False,
|
|
"ground_class": "ign", "ign_classes": "sol",
|
|
"bare_earth": False, "viz": ["aspect"],
|
|
"all_missing": False})]
|
|
|
|
|
|
def test_status_proxies_remote(monkeypatch):
|
|
"""/api/status rapporte l'état du job distant (file de génération)."""
|
|
import lidar_pipeline.webapp as webapp
|
|
calls = []
|
|
|
|
def fake_proxy(method, path, payload=None, timeout=30):
|
|
calls.append((method, path, timeout))
|
|
return {"running": True, "tiles": []}
|
|
|
|
monkeypatch.setattr(webapp, "GENERATION_URL", "http://distant:8973")
|
|
monkeypatch.setattr(webapp, "_proxy_api", fake_proxy)
|
|
data = webapp.status()
|
|
assert data["running"] is True and data["distant"] is True
|
|
assert calls == [("GET", "/api/status", 10)] # timeout court : sondage UI
|
|
|
|
|
|
def test_preview_forwards_to_remote(monkeypatch):
|
|
"""/api/preview interroge le distant (il seul connaît input/ complet)."""
|
|
import lidar_pipeline.webapp as webapp
|
|
calls = []
|
|
|
|
def fake_proxy(method, path, payload=None, timeout=30):
|
|
calls.append((method, path, payload))
|
|
return {"count": 0, "cells": []}
|
|
|
|
monkeypatch.setattr(webapp, "GENERATION_URL", "http://distant:8973")
|
|
monkeypatch.setattr(webapp, "_proxy_api", fake_proxy)
|
|
req = webapp.PreviewRequest(bbox=[7.85, 48.93, 7.86, 48.94], viz=["aspect"])
|
|
assert webapp.preview(req)["count"] == 0
|
|
assert calls[0][0] == "POST" and calls[0][1] == "/api/preview"
|
|
assert calls[0][2]["bbox"] == [7.85, 48.93, 7.86, 48.94]
|
|
|
|
|
|
def test_proxy_translates_unreachable_host(monkeypatch):
|
|
"""Une machine de traitement injoignable devient un 503 explicite."""
|
|
from fastapi import HTTPException
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "GENERATION_URL", "http://localhost:1")
|
|
try:
|
|
webapp._proxy_api("GET", "/api/status", timeout=1)
|
|
assert False, "une HTTPException était attendue"
|
|
except HTTPException as e:
|
|
assert e.status_code == 503
|
|
assert "injoignable" in e.detail
|
|
|
|
|
|
def test_fallback_viz_step_names_match_pipeline():
|
|
"""Le repli léger (sans pipeline) liste les mêmes étapes, même ordre."""
|
|
from lidar_pipeline.pipeline import VIZ_STEPS
|
|
from lidar_pipeline.webapp import _fallback_viz_step_names
|
|
assert _fallback_viz_step_names() == [name for name, _ in VIZ_STEPS]
|
|
|
|
|
|
def test_viz_step_labels_without_pipeline():
|
|
"""Les libellés d'étapes restent corrects même sans importer le pipeline."""
|
|
import lidar_pipeline.webapp as webapp
|
|
labels = webapp._viz_step_labels()
|
|
assert labels["hillshade"] == "Hillshade multidirectionnel"
|
|
assert labels["pos_open"] == "Openness positive"
|
|
|
|
|
|
def test_sync_is_rebuild_only(tmp_path, monkeypatch):
|
|
"""/api/sync régénère l'index en arrière-plan (plus de rsync)."""
|
|
import time as _time
|
|
import lidar_pipeline.webapp as webapp
|
|
import lidar_pipeline.index as index_mod
|
|
calls = []
|
|
monkeypatch.setattr(index_mod, "build_index", lambda out: calls.append(out))
|
|
webapp._rebuild["done"] = None # ignore l'état d'un test précédent
|
|
assert webapp.sync_and_rebuild()["demarré"] is True
|
|
for _ in range(200):
|
|
s = webapp.rebuild_status()
|
|
if s["done"] is not None and not s["running"]:
|
|
break
|
|
_time.sleep(0.05)
|
|
assert calls # l'index a été reconstruit
|
|
assert webapp.rebuild_status()["error"] is None
|
|
assert webapp.rebuild_status()["done"] is not None
|
|
|
|
|
|
def test_require_token():
|
|
"""Le jeton (si défini) bloque les appels sans ou avec mauvais mauvais header."""
|
|
from fastapi import HTTPException
|
|
import lidar_pipeline.webapp as webapp
|
|
old = webapp.API_TOKEN
|
|
webapp.API_TOKEN = "secret"
|
|
try:
|
|
for bad in (None, "faux"):
|
|
try:
|
|
webapp._require_token(bad)
|
|
assert False, "une HTTPException était attendue"
|
|
except HTTPException as e:
|
|
assert e.status_code == 401
|
|
assert webapp._require_token("secret") is None
|
|
finally:
|
|
webapp.API_TOKEN = old
|
|
# Sans jeton configuré : tout passe (LAN de confiance)
|
|
webapp.API_TOKEN = None
|
|
assert webapp._require_token(None) is None
|
|
|
|
|
|
class _FakeRequest:
|
|
"""Requête minimale pour tester la restriction par IP (peer + XFF)."""
|
|
|
|
def __init__(self, host=None, xff=None):
|
|
self.client = type("C", (), {"host": host})() if host else None
|
|
self.headers = {"x-forwarded-for": xff} if xff else {}
|
|
|
|
|
|
def test_generate_restricted_to_lan_cidr():
|
|
"""/api/generate n'est autorisé que depuis LIDAR_REGEN_CIDR (si défini)."""
|
|
from fastapi import HTTPException
|
|
import lidar_pipeline.webapp as webapp
|
|
old = webapp.REGEN_CIDR
|
|
webapp.REGEN_CIDR = "192.168.1.0/24"
|
|
try:
|
|
# IP du réseau autorisé
|
|
webapp._require_lan_for_generation(_FakeRequest(host="192.168.1.42"))
|
|
# Reverse proxy local (pair dans le réseau) : X-Forwarded-For désigne
|
|
# le client réel derrière lui — refusé s'il est hors réseau
|
|
webapp._require_lan_for_generation(
|
|
_FakeRequest(host="192.168.1.50", xff="192.168.1.10, 192.168.1.50"))
|
|
for host, xff in (("10.0.0.5", None), ("8.8.8.8", None),
|
|
("192.168.3.1", None), ("2001:db8::1", None),
|
|
# XFF forgé par un client externe : ignoré (pair hors réseau)
|
|
("8.8.8.8", "192.168.1.5"),
|
|
# Client réel externe derrière le proxy local
|
|
("192.168.1.50", "8.8.8.8"),
|
|
# IP absente
|
|
(None, None)):
|
|
try:
|
|
webapp._require_lan_for_generation(_FakeRequest(host=host, xff=xff))
|
|
assert False, f"403 attendu pour host={host!r} xff={xff!r}"
|
|
except HTTPException as e:
|
|
assert e.status_code == 403
|
|
assert "réseau local" in e.detail
|
|
finally:
|
|
webapp.REGEN_CIDR = old
|
|
|
|
|
|
def test_generate_cidr_disabled(monkeypatch):
|
|
"""LIDAR_REGEN_CIDR vide = restriction levée (tout client autorisé)."""
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "REGEN_CIDR", "")
|
|
webapp._require_lan_for_generation(_FakeRequest(host="8.8.8.8"))
|
|
webapp._require_lan_for_generation(_FakeRequest())
|
|
|
|
|
|
def test_status_exposes_regen_allowed():
|
|
"""/api/status indique à l'interface si le client peut générer."""
|
|
import lidar_pipeline.webapp as webapp
|
|
assert webapp.status(_FakeRequest(host="192.168.1.7"))["regen_allowed"] is True
|
|
assert webapp.status(_FakeRequest(host="8.8.8.8"))["regen_allowed"] is False
|
|
|
|
|
|
def test_regen_cidr_default_covers_local_and_private():
|
|
"""Défaut : localhost + plages privées (LAN, hôte Docker), pas Internet."""
|
|
import lidar_pipeline.webapp as webapp
|
|
# Miroir du défaut de webapp.REGEN_CIDR (multidiffusion interdite ici :
|
|
# on le fixe explicitement pour tester la valeur de liste multi-CIDR).
|
|
old = webapp.REGEN_CIDR
|
|
webapp.REGEN_CIDR = "127.0.0.0/8,::1,10.0.0.0/8,172.16.0.0/12,192.168.0.0/16"
|
|
try:
|
|
for host in ("127.0.0.1", "::1", # boucle locale
|
|
"172.17.0.1", "172.26.0.5", # ponts Docker
|
|
"192.168.1.42", "192.168.122.7", # LAN privé
|
|
"10.0.0.5"):
|
|
assert webapp._ip_in_regen_cidr(host), host
|
|
for host in ("8.8.8.8", "2001:db8::1", "169.254.1.1", None):
|
|
assert not webapp._ip_in_regen_cidr(host), host
|
|
finally:
|
|
webapp.REGEN_CIDR = old
|
|
|
|
|
|
def test_regen_cidr_accepts_comma_separated_list():
|
|
"""LIDAR_REGEN_CIDR accepte plusieurs réseaux (entrées invalides ignorées)."""
|
|
import lidar_pipeline.webapp as webapp
|
|
old = webapp.REGEN_CIDR
|
|
webapp.REGEN_CIDR = "192.168.1.0/24, 10.0.0.0/8, pas-un-cidr"
|
|
try:
|
|
assert webapp._ip_in_regen_cidr("192.168.1.9")
|
|
assert webapp._ip_in_regen_cidr("10.1.2.3")
|
|
assert not webapp._ip_in_regen_cidr("192.168.3.9")
|
|
assert not webapp._ip_in_regen_cidr("172.17.0.1")
|
|
finally:
|
|
webapp.REGEN_CIDR = old
|
|
|
|
|
|
|