"""Tests for pipeline orchestration.""" import pytest from pathlib import Path class TestVizSteps: def test_viz_steps_not_empty(self): from lidar_pipeline.pipeline import VIZ_STEPS assert len(VIZ_STEPS) > 0 def test_viz_steps_have_callable_functions(self): from lidar_pipeline.pipeline import VIZ_STEPS for name, func in VIZ_STEPS: assert callable(func), f"VIZ_STEPS entry '{name}' is not callable" def test_viz_steps_names_unique(self): from lidar_pipeline.pipeline import VIZ_STEPS names = [name for name, _ in VIZ_STEPS] assert len(names) == len(set(names)), "VIZ_STEPS has duplicate names" def test_expected_visualization_count(self): """17 visualisations : 14 terrain + densité de points + ortho + topo.""" from lidar_pipeline.pipeline import VIZ_STEPS assert len(VIZ_STEPS) == 17 def test_default_run_produces_only_panel_layers(self, tmp_path): """Sans --only : seules les couches affichées (relief orienté, densité de points) sont produites ; --only reste libre pour les autres.""" from lidar_pipeline.pipeline import LidarArchaeoPipeline p = LidarArchaeoPipeline(tmp_path, tmp_path / "out") assert [n for n, _ in p.viz_steps] == ["relief_oriente", "densite_sol"] p = LidarArchaeoPipeline(tmp_path, tmp_path / "out2", only_viz=["slope"]) assert [n for n, _ in p.viz_steps] == ["slope"] def test_incremental_index_on_by_default(self, tmp_path): """La carte suit le rendu en cours quel que soit le lanceur.""" from lidar_pipeline.pipeline import LidarArchaeoPipeline assert LidarArchaeoPipeline(tmp_path, tmp_path / "o").incremental_index assert not LidarArchaeoPipeline(tmp_path, tmp_path / "o2", no_index=True).incremental_index def test_debounced_tile_is_indexed_later(self, tmp_path, monkeypatch): """Une dalle terminée pendant l'anti-rebond est reprise par une passe différée, sans attendre la dalle suivante.""" import time import lidar_pipeline.index as index from lidar_pipeline.pipeline import LidarArchaeoPipeline calls = [] monkeypatch.setattr(index, "build_index", lambda *a, **k: calls.append(time.time())) p = LidarArchaeoPipeline(tmp_path, tmp_path / "o") p._rebuild_index_incremental() # passe immédiate p._last_index_rebuild = time.time() - 2.8 # anti-rebond presque écoulé p._rebuild_index_incremental() # différée (~0,2 s) p._rebuild_index_incremental() # déjà programmée : pas de doublon time.sleep(0.6) assert len(calls) == 2 def test_ortho_and_topo_present(self): from lidar_pipeline.pipeline import VIZ_STEPS names = [name for name, _ in VIZ_STEPS] assert "ortho" in names assert "topo" in names class TestFetchEdgeNeighbors: """Raccord des bords : pré-téléchargement des voisines manquantes.""" def test_downloads_missing_ring_dedup(self, tmp_path, monkeypatch): """Les 8 voisines manquantes sont demandées une seule fois, présente exclue.""" import lidar_pipeline.fetch_ign as fetch_ign from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() lazh = input_dir / "LHD_FXX_0999_6778_PTS_LAMB93_IGN69.copc.laz" lazh.touch() # Une voisine déjà présente ne doit pas être retéléchargée. (input_dir / "LHD_FXX_1000_6779_PTS_LAMB93_IGN69.copc.laz").touch() calls = [] def fake_fetch_tiles(input_dir_, specs, **kwargs): calls.extend(specs) return [input_dir / "fake" for _ in specs] monkeypatch.setattr(fetch_ign, "fetch_tiles", fake_fetch_tiles) pipeline = LidarArchaeoPipeline(input_dir, tmp_path / "output", edge_buffer=100.0) pipeline._fetch_edge_neighbors([lazh]) assert len(calls) == 7 # 8 voisines - 1 déjà présente assert (1000, 6779) not in calls assert sorted(set(calls)) == sorted(calls) # dédupliqué def test_no_download_without_edge_buffer(self, tmp_path, monkeypatch): """Raccord désactivé : aucun téléchargement de voisines.""" import lidar_pipeline.fetch_ign as fetch_ign from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() lazh = input_dir / "LHD_FXX_0999_6778_PTS_LAMB93_IGN69.copc.laz" lazh.touch() def boom(*args, **kwargs): raise AssertionError("fetch_tiles ne doit pas être appelé") monkeypatch.setattr(fetch_ign, "fetch_tiles", boom) pipeline = LidarArchaeoPipeline(input_dir, tmp_path / "output", edge_buffer=0.0) pipeline._fetch_edge_neighbors([lazh]) class TestCleanupEdgeNeighborDuplicates: """Nettoyage des doublons entre input/ et input/edge_neighbors/.""" def test_removes_true_duplicates_keeps_unique_and_part(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() edge_dir = input_dir / "edge_neighbors" edge_dir.mkdir() dup_name = "LHD_FXX_1000_6779_PTS_LAMB93_IGN69.copc.laz" unique_name = "LHD_FXX_1001_6779_PTS_LAMB93_IGN69.copc.laz" part_name = "LHD_FXX_1002_6779_PTS_LAMB93_IGN69.copc.laz.part" (input_dir / dup_name).write_bytes(b"authoritative") (edge_dir / dup_name).write_bytes(b"duplicata") (edge_dir / unique_name).write_bytes(b"voisine-unique") (edge_dir / part_name).write_bytes(b"en-cours") pipeline = LidarArchaeoPipeline(input_dir, tmp_path / "output", edge_buffer=100.0) pipeline._cleanup_edge_neighbor_duplicates(edge_dir) assert not (edge_dir / dup_name).exists() assert (input_dir / dup_name).read_bytes() == b"authoritative" assert (edge_dir / unique_name).exists() assert (edge_dir / part_name).exists() def test_noop_when_edge_dir_missing(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() pipeline = LidarArchaeoPipeline(input_dir, tmp_path / "output", edge_buffer=100.0) # Ne doit pas lever si edge_neighbors/ n'existe pas encore. pipeline._cleanup_edge_neighbor_duplicates(input_dir / "edge_neighbors") class TestLidarArchaeoPipeline: def test_init_creates_dirs(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() output_dir = tmp_path / "output" pipeline = LidarArchaeoPipeline(str(input_dir), str(output_dir)) assert (tmp_path / "output").exists() assert (tmp_path / "output" / "DTM").exists() assert (tmp_path / "output" / "visualisations").exists() assert (tmp_path / "output" / "temp").exists() def test_init_raises_on_missing_input(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline with pytest.raises(ValueError, match="introuvable"): LidarArchaeoPipeline("/nonexistent/path", str(tmp_path / "output")) def test_incremental_index_rebuild(self, tmp_path, monkeypatch): """Mode incrémental : l'index est régénéré après une tuile, avec anti-rebond.""" from lidar_pipeline.pipeline import LidarArchaeoPipeline import lidar_pipeline.index as index_mod input_dir = tmp_path / "input" input_dir.mkdir() calls = [] monkeypatch.setattr(index_mod, "build_index", lambda *a, **k: calls.append(a) or None) pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"), incremental_index=True) pipeline._rebuild_index_incremental() pipeline._rebuild_index_incremental() # < 3 s : anti-rebond, ignoré assert len(calls) == 1 # --no-index : jamais de rebuild incrémental pipeline._last_index_rebuild = 0.0 pipeline.no_index = True pipeline._rebuild_index_incremental() assert len(calls) == 1 def test_find_laz_files_empty(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output")) files = pipeline.find_laz_files() assert files == [] def test_find_laz_files(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() (input_dir / "test.laz").touch() (input_dir / "other.las").touch() (input_dir / "readme.txt").touch() pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output")) files = pipeline.find_laz_files() names = [f.name for f in files] assert "test.laz" in names assert "other.las" in names assert "readme.txt" not in names def test_find_laz_files_sorted_north_to_south(self, tmp_path): """Lignes LHD triées du nord au sud (row décroissante, col croissante).""" from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() for name in ("LHD_FXX_1054_6880_PTS_LAMB93_IGN69.copc.laz", "LHD_FXX_1056_6882_PTS_LAMB93_IGN69.copc.laz", "LHD_FXX_1053_6882_PTS_LAMB93_IGN69.copc.laz", "zz_autre.laz"): (input_dir / name).touch() pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output")) names = [f.name for f in pipeline.find_laz_files()] assert names == [ "LHD_FXX_1053_6882_PTS_LAMB93_IGN69.copc.laz", # ligne nord, col mini "LHD_FXX_1056_6882_PTS_LAMB93_IGN69.copc.laz", # ligne nord, col maxi "LHD_FXX_1054_6880_PTS_LAMB93_IGN69.copc.laz", # ligne sud "zz_autre.laz", # hors pattern : en fin ] class TestDtmMethodSidecar: """Méthode de classification enregistrée à côté du DTM (invalidation du cache).""" def test_missing_sidecar_matches(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"), ground_method='csf') # Aucun sidecar écrit → cache conservé (considéré compatible). assert pipeline._dtm_method_matches("tileA", "") is True def test_matching_method(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"), ground_method='csf') pipeline._write_dtm_method("tileA", "") assert pipeline._dtm_method_matches("tileA", "") is True def test_different_method_invalidates_cache(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() out = str(tmp_path / "output") LidarArchaeoPipeline(str(input_dir), out, ground_method='ign')._write_dtm_method("tileA", "") csf = LidarArchaeoPipeline(str(input_dir), out, ground_method='csf') assert csf._dtm_method_matches("tileA", "") is False def test_write_dtm_method(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"), ground_method='smrf') pipeline._write_dtm_method("tileA", "_r0p2") sidecar = tmp_path / "output" / "DTM" / "tileA_dtm_r0p2_method.txt" assert sidecar.exists() assert sidecar.read_text(encoding="utf-8").strip() == "smrf" assert pipeline._dtm_method_name("tileA", "_r0p2") == "smrf" # Le sidecar est un fichier .txt : il ne gêne pas la recherche des DTM .tif. dtm = tmp_path / "output" / "DTM" / "tileA_dtm_r0p2.tif" dtm.touch() assert [p.name for p in (tmp_path / "output" / "DTM").glob("*.tif")] == ["tileA_dtm_r0p2.tif"] def test_force_images_regenerates_existing(self, tmp_path): from lidar_pipeline.pipeline import LidarArchaeoPipeline input_dir = tmp_path / "input" input_dir.mkdir() pipeline = LidarArchaeoPipeline(str(input_dir), str(tmp_path / "output"), output_format='avif') calls = [] def fake_ortho(dem_file, basename, vis_dir, resolution): calls.append(basename) return vis_dir / f"{basename}_ortho.avif" pipeline.viz_steps = [('ortho', fake_ortho)] vis_dir = tmp_path / "output" / "visualisations" / "tileA" vis_dir.mkdir(parents=True) (vis_dir / "tileA_ortho.avif").touch() dtm = tmp_path / "dtm.tif" # Image existante, pas de force → ignorée (pas de régénération). pipeline.generate_all_visualizations(dtm, "tileA", resolution=0.5, vis_dir=vis_dir, force_images=False) assert calls == [] # Image existante, force_images=True → régénérée. calls.clear() pipeline.generate_all_visualizations(dtm, "tileA", resolution=0.5, vis_dir=vis_dir, force_images=True) assert calls == ["tileA"] class TestEffectiveGroundMethod: def test_ign_label_encodes_classes(self): """Les classes IGN sont encodées dans l'étiquette de cache (reclassification).""" from lidar_pipeline.pipeline import LidarArchaeoPipeline import tempfile with tempfile.TemporaryDirectory() as tmpdir: p = LidarArchaeoPipeline(tmpdir, tmpdir, ground_method='ign', ign_classes="sol,unclassified") assert p._effective_ground_method() == "ign_1_2" def test_ign_default_label(self): from lidar_pipeline.pipeline import LidarArchaeoPipeline import tempfile with tempfile.TemporaryDirectory() as tmpdir: p = LidarArchaeoPipeline(tmpdir, tmpdir, ground_method='ign') assert p._effective_ground_method() == "ign" def test_other_methods_unchanged(self): from lidar_pipeline.pipeline import LidarArchaeoPipeline import tempfile with tempfile.TemporaryDirectory() as tmpdir: p = LidarArchaeoPipeline(tmpdir, tmpdir, ground_method='smrf', ign_classes="sol,unclassified") assert p._effective_ground_method() == "smrf" class TestResolveWorkers: def test_auto_scales_with_cpus(self): from lidar_pipeline.pipeline import resolve_workers import os w = resolve_workers('auto') assert w == max(2, min((os.cpu_count() or 4) - 2, 16)) def test_auto_bounded(self): from lidar_pipeline.pipeline import resolve_workers # Bornes : jamais sous 2, jamais au-dessus de 16 assert 2 <= resolve_workers('auto') <= 16 def test_explicit_int(self): from lidar_pipeline.pipeline import resolve_workers assert resolve_workers('4') == 4 assert resolve_workers(7) == 7 def test_invalid_falls_back_to_one(self): from lidar_pipeline.pipeline import resolve_workers assert resolve_workers('abc') == 1 assert resolve_workers(None) == 1 def test_worker_slot_initializer(self, monkeypatch): """Chaque processus du pool prend UNE place à sa création : GPU (set_active_gpu) ou CPU forcé (-1) ; None ou file vide = libre.""" import queue from lidar_pipeline import gpu, pipeline calls = [] monkeypatch.setattr(gpu, "set_active_gpu", lambda i: calls.append(("gpu", i))) monkeypatch.setattr(gpu, "force_cpu", lambda: calls.append(("cpu",))) q = queue.Queue() for slot in (1, -1, None): q.put(slot) for _ in range(4): # 4e appel : file vide pipeline._init_worker_slot(q) assert calls == [("gpu", 1), ("cpu",)]