Split webapp for Raspberry Pi deployment, remote generation API and sync
La webapp (carte + vignettes) et la génération de tuiles se déploient sur deux machines : image légère Dockerfile.webapp (FastAPI + Pillow AVIF natif + pyproj) sur Raspberry Pi, pipeline complet sur la machine de traitement. LIDAR_GENERATION_URL délègue /api/generate, /api/preview et /api/status ; /api/sync ramène les tuiles par rsync puis régénère vignettes et index localement. Token partagé optionnel (LIDAR_API_TOKEN/LIDAR_REMOTE_TOKEN). Retire du dépôt les journaux internes (.swival, audit-findings) et les données (data/, notebooks/). Doc : docs/DEPLOY_WEBAPP.md.
This commit is contained in:
Antoine Jacquin
@ -26,7 +26,7 @@ def test_processed_cells(tmp_path):
|
||||
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_r0p2_aspect.avif").write_bytes(b"x")
|
||||
(vis / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_aspect.avif").write_bytes(b"x")
|
||||
assert processed_cells(tmp_path) == {(1054, 6882)}
|
||||
|
||||
|
||||
@ -53,6 +53,174 @@ def test_missing_cells_include_done(tmp_path):
|
||||
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_both_resolutions(tmp_path):
|
||||
"""Une tuile complète à 0.5 mais sans 0.2 reste à traiter."""
|
||||
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=True ajoute --force --force-classification à la commande."""
|
||||
from lidar_pipeline.webapp import _build_command
|
||||
@ -92,9 +260,37 @@ def test_build_command_default_viz_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", "aniso_open"])
|
||||
cmd = _build_command([(1054, 6882)], viz=["aspect", "wavelet", "slope"])
|
||||
i = cmd.index("--only")
|
||||
assert cmd[i + 1:i + 4] == ["aspect", "wavelet", "aniso_open"]
|
||||
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_viz_step_names_match_pipeline():
|
||||
@ -117,9 +313,8 @@ def test_generate_rejects_unknown_viz():
|
||||
assert "inconnu" in e.detail
|
||||
|
||||
|
||||
def test_generate_accepts_wavelet_and_aniso():
|
||||
"""L'API accepte wavelet + aniso_open et les transmet à --only."""
|
||||
import subprocess
|
||||
def test_generate_accepts_wavelet_and_slope(monkeypatch):
|
||||
"""L'API accepte wavelet + slope et les transmet à --only."""
|
||||
import lidar_pipeline.webapp as webapp
|
||||
captured = {}
|
||||
|
||||
@ -134,16 +329,22 @@ def test_generate_accepts_wavelet_and_aniso():
|
||||
return 0
|
||||
return _P()
|
||||
|
||||
webapp.subprocess.Popen = fake_popen
|
||||
try:
|
||||
req = webapp.GenerateRequest(tiles=[[1054, 6882]],
|
||||
viz=["wavelet", "aniso_open"])
|
||||
result = webapp.generate(req)
|
||||
assert result["demarré"] is True
|
||||
i = captured["cmd"].index("--only")
|
||||
assert captured["cmd"][i + 1:i + 3] == ["wavelet", "aniso_open"]
|
||||
finally:
|
||||
webapp.subprocess.Popen = subprocess.Popen
|
||||
# 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):
|
||||
@ -173,3 +374,212 @@ def test_rebuild_index_background(tmp_path, monkeypatch):
|
||||
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,
|
||||
"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_runs_command_then_rebuild(tmp_path, monkeypatch):
|
||||
"""POST /api/sync exécute LIDAR_SYNC_CMD puis build_index en arrière-plan."""
|
||||
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))
|
||||
monkeypatch.setattr(webapp, "SYNC_CMD", "exit 0")
|
||||
webapp._rebuild["done"] = None # ignore l'état d'un test précédent
|
||||
assert webapp.sync_and_rebuild()["sync"] 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 après la synchronisation
|
||||
assert webapp.rebuild_status()["error"] is None
|
||||
assert webapp.rebuild_status()["done"] is not None
|
||||
|
||||
|
||||
def test_sync_failure_reported_but_index_rebuilt(tmp_path, monkeypatch):
|
||||
"""Un rsync en échec est remonté dans l'état sans bloquer le rebuild."""
|
||||
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))
|
||||
monkeypatch.setattr(webapp, "SYNC_CMD", "exit 3")
|
||||
webapp._rebuild["done"] = None
|
||||
webapp.sync_and_rebuild()
|
||||
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 # rebuild quand même (données locales éventuelles)
|
||||
assert "rc=3" in webapp.rebuild_status()["error"]
|
||||
|
||||
|
||||
def test_sync_without_cmd_is_rebuild_only(monkeypatch):
|
||||
"""Sans LIDAR_SYNC_CMD (machine locale), /api/sync = simple rebuild."""
|
||||
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))
|
||||
monkeypatch.setattr(webapp, "SYNC_CMD", None)
|
||||
webapp._rebuild["done"] = None
|
||||
assert webapp.sync_and_rebuild()["sync"] is False
|
||||
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 and webapp.rebuild_status()["error"] is 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
|
||||
|
||||
Reference in New Issue
Block a user