lidar_rendu/lidar_pipeline/tests/test_webapp.py

"""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_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
    cmd = " ".join(_build_command([(1054, 6882)], regenerate=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, avec ou sans régénération
    for method in GROUND_CLASS_METHODS:
        for regenerate in (False, True):
            cmd = _build_command([(1054, 6882)], regenerate=regenerate, ground_class=method)
            i = cmd.index("--ground-classification")
            assert cmd[i + 1] == method
            assert ("--force" in cmd) == regenerate
            assert ("--force-classification" in cmd) == regenerate


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_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,
                       "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


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 (défaut 192.168.1.0/24)."""
    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_auto_sync_once_skips_when_busy(monkeypatch):
    """Le cycle de cache ne lance rien si un rebuild tourne déjà."""
    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")
    # Un cycle démarre bien (sync + rebuild)
    assert webapp._auto_sync_once() is True
    for _ in range(200):
        if not webapp.rebuild_status()["running"]:
            break
        _time.sleep(0.05)
    assert calls
    # Rebuild artificiellement occupé : le cycle suivant est sauté sans erreur
    webapp._rebuild["running"] = True
    try:
        assert webapp._auto_sync_once() is False
        assert len(calls) == 1
    finally:
        webapp._rebuild["running"] = False