Merge branch 'fix-edge-neighbor-duplicates'

lidar_pipeline/tests/test_pipeline.py

@@ -107,6 +107,52 @@ class TestFetchEdgeNeighbors:
         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"
+        mismatch_name = "LHD_FXX_1003_6779_PTS_LAMB93_IGN69.copc.laz"
+
+        (input_dir / dup_name).write_bytes(b"authoritative")
+        (edge_dir / dup_name).write_bytes(b"authoritative")
+        # input/ tronqué (taille différente) : la voisine peut être la seule
+        # copie saine, elle doit rester.
+        (input_dir / mismatch_name).write_bytes(b"")
+        (edge_dir / mismatch_name).write_bytes(b"complete")
+        (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()
+        assert (edge_dir / mismatch_name).read_bytes() == b"complete"
+
+    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