f024514427
- Clic sur la carte pour choisir des dalles précises, même sans données existantes (via /api/cell) ; les zones dessinées s'ajoutent à la sélection au lieu de la remplacer - Une demande lancée pendant un run part en file d'attente côté serveur et démarre à la fin du travail en cours (plus de refus 409), file vidable depuis l'interface - Tuiles et interface toujours fraîches : images servies sans cache navigateur (revalidation 304), rechargement de la carte seulement une fois le rebuild de l'index terminé - serve-webapp.sh : sous-commande update retirée, documentation de déploiement corrigée en conséquence
1048 lines
44 KiB
Python
1048 lines
44 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_preset_sanitize_and_upsert(tmp_path, monkeypatch):
|
|
"""Presets de couches : valeurs assainies, remplacées par identifiant."""
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "PRESETS_FILE", tmp_path / ".presets.json")
|
|
req = webapp.PresetRequest(id="Relief v2!", label=" Reliefs ",
|
|
on=["slope", "aspect"],
|
|
opacity={"slope": 0.5, "aspect": "0.8", "lrm": "x"},
|
|
osm={"on": True, "opacity": 1.5, "dark": True})
|
|
p = webapp._sanitize_preset(req)
|
|
assert p["id"] == "reliefv2"
|
|
assert p["label"] == "Reliefs"
|
|
assert p["opacity"] == {"slope": 0.5, "aspect": 0.8}
|
|
assert p["osm"] == {"on": True, "opacity": 1.0, "dark": True}
|
|
webapp.upsert_preset(req)
|
|
assert [x["label"] for x in webapp.list_presets()["presets"]] == ["Reliefs"]
|
|
# Même identifiant : remplacement, pas de doublon
|
|
webapp.upsert_preset(webapp.PresetRequest(id="reliefv2", label="Reliefs v2",
|
|
on=["slope"], opacity={}))
|
|
got = webapp.list_presets()["presets"]
|
|
assert len(got) == 1 and got[0]["label"] == "Reliefs v2"
|
|
# Suppression inconnue : 404
|
|
from fastapi import HTTPException
|
|
webapp.delete_preset(got[0]["id"])
|
|
assert webapp.list_presets()["presets"] == []
|
|
try:
|
|
webapp.delete_preset("reliefv2")
|
|
assert False, "une HTTPException était attendue"
|
|
except HTTPException as e:
|
|
assert e.status_code == 404
|
|
|
|
|
|
def test_preset_file_corrupted_is_ignored(tmp_path, monkeypatch):
|
|
"""Un fichier de presets corrompu est ignoré (liste vide, pas d'exception)."""
|
|
import lidar_pipeline.webapp as webapp
|
|
monkeypatch.setattr(webapp, "PRESETS_FILE", tmp_path / ".presets.json")
|
|
webapp.PRESETS_FILE.write_text("{pas du json", encoding="utf-8")
|
|
assert webapp._load_presets() == []
|
|
|
|
|
|
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_rebuild_flag_synchronous(monkeypatch):
|
|
"""Le rebuild est marqué running dès le POST (pas après coup).
|
|
|
|
Un GET /api/sync juste après le POST doit voir le rebuild en cours :
|
|
sinon il lirait le done du rebuild précédent et l'interface
|
|
rechargerait une carte périmée (bug de rafraîchissement).
|
|
"""
|
|
import threading
|
|
import time
|
|
import lidar_pipeline.webapp as webapp
|
|
import lidar_pipeline.index as index_mod
|
|
|
|
go = threading.Event()
|
|
|
|
def slow_build(out):
|
|
go.wait(2) # rebuild artificiellement long
|
|
|
|
monkeypatch.setattr(index_mod, "build_index", slow_build)
|
|
saved = dict(webapp._rebuild)
|
|
try:
|
|
d = webapp._start_rebuild()
|
|
assert d["demarré"] is True
|
|
assert d["started"] <= time.time()
|
|
# Déjà « running » au retour du POST, avant la fin du fil
|
|
assert webapp.rebuild_status()["running"] is True
|
|
assert webapp.rebuild_status()["done"] is None
|
|
go.set()
|
|
for _ in range(100):
|
|
if not webapp.rebuild_status()["running"]:
|
|
break
|
|
time.sleep(0.05)
|
|
s = webapp.rebuild_status()
|
|
assert s["running"] is False
|
|
assert s["done"] is not None and s["done"] >= d["started"]
|
|
finally:
|
|
webapp._rebuild.update(saved)
|
|
|
|
|
|
def test_static_mounts_served_no_cache(tmp_path):
|
|
"""Les montages d'images servent Cache-Control: no-cache.
|
|
|
|
Le ?v= des URLs suit la mtime de la source, pas celle du fichier servi
|
|
(vignette recalculée, cache webapp rapatrié) : sans revalidation imposée,
|
|
le cache heuristique du navigateur peut afficher l'ancien rendu.
|
|
"""
|
|
import asyncio
|
|
from starlette.routing import Mount
|
|
import lidar_pipeline.webapp as webapp
|
|
|
|
assert issubclass(webapp._OnDemandStaticFiles, webapp._NoCacheStaticFiles)
|
|
mounted = {r.path: r for r in webapp.app.routes if isinstance(r, Mount)}
|
|
for name in ("index_thumbs", "index_subtiles", "visualisations", "DTM"):
|
|
route = mounted.get(f"/{name}")
|
|
assert route is not None, f"montage /{name} absent"
|
|
assert isinstance(route.app, webapp._NoCacheStaticFiles)
|
|
|
|
# L'en-tête est bien posé sur la réponse servie
|
|
(tmp_path / "x.jpg").write_bytes(b"1")
|
|
srv = webapp._NoCacheStaticFiles(directory=str(tmp_path))
|
|
|
|
async def fetch():
|
|
scope = {"type": "http", "method": "GET", "path": "/x.jpg",
|
|
"headers": [], "query_string": b""}
|
|
return await srv.get_response("x.jpg", scope)
|
|
|
|
resp = asyncio.run(fetch())
|
|
assert resp.headers["cache-control"] == "no-cache, must-revalidate"
|
|
|
|
|
|
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_point_to_cell_matches_bbox_cells():
|
|
"""point_to_cell désigne la cellule contenant le point (cohérent bbox_to_cells)."""
|
|
from lidar_pipeline.webapp import point_to_cell, bbox_to_cells
|
|
from rasterio.warp import transform as warp_transform
|
|
# Centre de la dalle 1054,6882 en L93 → WGS84
|
|
lons, lats = warp_transform('EPSG:2154', 'EPSG:4326', [1054500], [6881500])
|
|
lon, lat = lons[0], lats[0]
|
|
assert point_to_cell(lon, lat) == (1054, 6882)
|
|
# Une petite bbox autour du même point donne la même cellule, elle seule
|
|
cells = bbox_to_cells(lon - 1e-4, lat - 1e-4, lon + 1e-4, lat + 1e-4)
|
|
assert cells == [(1054, 6882)]
|
|
# Point vers le bord est de la dalle suivante
|
|
lons, lats = warp_transform('EPSG:2154', 'EPSG:4326', [1055800], [6881500])
|
|
assert point_to_cell(lons[0], lats[0]) == (1055, 6882)
|
|
|
|
|
|
def test_api_cell_returns_corners():
|
|
"""/api/cell (clic carte) renvoie la dalle L93 et ses coins GPS."""
|
|
from lidar_pipeline.webapp import cell_at_point
|
|
from rasterio.warp import transform as warp_transform
|
|
lons, lats = warp_transform('EPSG:2154', 'EPSG:4326', [1054500], [6881500])
|
|
d = cell_at_point(lat=lats[0], lng=lons[0])
|
|
assert d["col"] == 1054 and d["row"] == 6882
|
|
assert len(d["corners"]) == 4 # SW, SE, NE, NW
|
|
|
|
|
|
def test_generate_enqueues_while_running(tmp_path, monkeypatch):
|
|
"""Une demande pendant un run part en file (plus de 409, rien n'est coupé)."""
|
|
import time
|
|
import lidar_pipeline.webapp as webapp
|
|
|
|
class FakeProc:
|
|
pid = 4242
|
|
def poll(self):
|
|
return None # en cours
|
|
def wait(self, timeout=None):
|
|
return None
|
|
|
|
saved_job = dict(webapp._job)
|
|
saved_queue = list(webapp._queue)
|
|
monkeypatch.setattr(webapp, "QUEUE_FILE", tmp_path / "queue.json")
|
|
try:
|
|
webapp._job.update({"proc": FakeProc(), "started": time.time(),
|
|
"returncode": None, "cmd": ["x"], "finished": None,
|
|
"qid": None})
|
|
webapp._queue.clear()
|
|
req = webapp.GenerateRequest(tiles=[[1054, 6882]], viz=["aspect"])
|
|
d = webapp.generate(req)
|
|
assert d["demarré"] is False
|
|
assert d["en_file"] == 1 and d["tuiles"] == 1 and d["qid"] is not None
|
|
assert len(webapp._queue) == 1
|
|
assert webapp._queue[0]["req"]["tiles"] == [[1054, 6882]]
|
|
# La file survit à un « redémarrage » du serveur
|
|
webapp._queue.clear()
|
|
webapp._load_queue()
|
|
assert len(webapp._queue) == 1
|
|
# Retrait des demandes en attente (le run en cours n'est pas touché)
|
|
assert webapp.queue_clear() == {"retirées": 1}
|
|
assert webapp._queue == []
|
|
finally:
|
|
webapp._job.update(saved_job)
|
|
webapp._queue[:] = saved_queue
|
|
|
|
|
|
def test_start_next_queued_launches_after_run(tmp_path, monkeypatch):
|
|
"""Serveur libre + file non vide : la demande suivante démarre (fidélité qid)."""
|
|
import time
|
|
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()
|
|
|
|
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)
|
|
monkeypatch.setattr(webapp, "INPUT_DIR", tmp_path)
|
|
monkeypatch.setattr(webapp, "OUTPUT_DIR", tmp_path)
|
|
monkeypatch.setattr(webapp, "LOG_FILE", tmp_path / "gen.log")
|
|
monkeypatch.setattr(webapp, "QUEUE_FILE", tmp_path / "queue.json")
|
|
saved_job = dict(webapp._job)
|
|
saved_queue = list(webapp._queue)
|
|
try:
|
|
webapp._job.update({"proc": None, "started": None, "returncode": 0,
|
|
"cmd": None, "finished": time.time(), "qid": None})
|
|
webapp._queue[:] = [{"id": 7, "queued_at": 1.0,
|
|
"req": {"tiles": [[1054, 6882]], "viz": ["aspect"]}}]
|
|
assert webapp._start_next_queued() is True
|
|
assert webapp._queue == [] # la demande est consommée
|
|
assert webapp._job["qid"] == 7 # le run porte l'identifiant de file
|
|
assert "--fetch-tiles" in captured["cmd"]
|
|
assert "1054,6882" in captured["cmd"]
|
|
finally:
|
|
webapp._job.update(saved_job)
|
|
webapp._queue[:] = saved_queue
|
|
|
|
|
|
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
|
|
|
|
|
|
|