lidar_rendu/lidar_pipeline/tests/test_progress.py

"""Tests for the progress event log (generation queue)."""


def test_report_and_read_events(tmp_path):
    """Events are written as JSONL and read back in order."""
    from lidar_pipeline.progress import read_events, report_event, reset_events
    reset_events(tmp_path)
    report_event(tmp_path, "LHD_FXX_1054_6882_PTS_LAMB93_IGN69", "download", "start")
    report_event(tmp_path, "LHD_FXX_1054_6882_PTS_LAMB93_IGN69", "download", "ok")
    events = read_events(tmp_path)
    assert len(events) == 2
    assert events[0]["phase"] == "download" and events[0]["state"] == "start"
    assert events[1]["state"] == "ok"
    assert events[1]["tile"] == "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"


def test_reset_clears_previous_run(tmp_path):
    """reset_events truncates the log: a new run starts from scratch."""
    from lidar_pipeline.progress import read_events, report_event, reset_events
    reset_events(tmp_path)
    report_event(tmp_path, "X", "tile", "ok")
    reset_events(tmp_path)
    assert read_events(tmp_path) == []


def test_read_events_missing_or_partial(tmp_path):
    """Directory without a log → empty list; partial line → skipped."""
    from lidar_pipeline.progress import events_path, read_events
    assert read_events(tmp_path) == []
    events_path(tmp_path).write_text(
        '{"tile": "X", "phase": "tile", "state": "ok"}\n'
        '{"tile": "X", "phase": "vi', encoding="utf-8")
    events = read_events(tmp_path)
    assert len(events) == 1 and events[0]["state"] == "ok"


def test_short_name():
    """The short name extracts col-row from the IGN tile name."""
    from lidar_pipeline.progress import tile_short_name
    assert tile_short_name("LHD_FXX_1054_6882_PTS_LAMB93_IGN69") == "1054-6882"
    assert tile_short_name("LHD_FXX_1054_6882_PTS_LAMB93_IGN69.copc.laz") == "1054-6882"
    assert tile_short_name("other.tif") == "other.tif"
    assert tile_short_name("") == ""


def test_aggregate_full_run(tmp_path):
    """Full run: ordered steps with their labels, tile done."""
    from lidar_pipeline.progress import aggregate_tiles, read_events, report_event, reset_events
    reset_events(tmp_path)
    b = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
    report_event(tmp_path, b, "download", "skip", "already in input/")
    report_event(tmp_path, b, "classif", "start")
    report_event(tmp_path, b, "classif", "ok")
    report_event(tmp_path, b, "dtm", "ok", res=0.5)
    report_event(tmp_path, b, "viz", "start", "aspect", res=0.5)
    report_event(tmp_path, b, "viz", "ok", "aspect", res=0.5)
    report_event(tmp_path, b, "tile", "ok", "123s")
    tiles = aggregate_tiles(read_events(tmp_path), viz_labels={"aspect": "Aspect"})
    assert len(tiles) == 1
    t = tiles[0]
    assert t["short"] == "1054-6882" and t["state"] == "done"
    assert [s["label"] for s in t["steps"]] == [
        "Download", "Classification", "DTM 0.5 m", "Aspect"]
    assert [s["state"] for s in t["steps"]] == ["skip", "ok", "ok", "ok"]


def test_aggregate_running_step(tmp_path):
    """A started but unfinished step leaves the tile "running"."""
    from lidar_pipeline.progress import aggregate_tiles, read_events, report_event, reset_events
    reset_events(tmp_path)
    b = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
    report_event(tmp_path, b, "download", "ok")
    report_event(tmp_path, b, "classif", "ok")
    report_event(tmp_path, b, "dtm", "ok", res=0.5)
    report_event(tmp_path, b, "viz", "start", "aspect", res=0.5)
    tiles = aggregate_tiles(read_events(tmp_path))
    assert tiles[0]["state"] == "running"
    assert tiles[0]["steps"][-1]["state"] == "running"


def test_aggregate_failure_marks_tile_failed():
    """A failed viz marks the tile failed even without a tile event."""
    from lidar_pipeline.progress import aggregate_tiles
    b = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
    tiles = aggregate_tiles([
        {"tile": b, "phase": "classif", "state": "ok"},
        {"tile": b, "phase": "dtm", "state": "ok", "res": 0.5},
        {"tile": b, "phase": "viz", "state": "start", "detail": "aspect", "res": 0.5},
        {"tile": b, "phase": "viz", "state": "fail", "detail": "aspect", "res": 0.5},
    ])
    assert tiles[0]["state"] == "failed"
    aspect = [s for s in tiles[0]["steps"] if s["detail"] == "aspect"][0]
    assert aspect["state"] == "fail"


def test_aggregate_multi_resolution_and_bad_input():
    """Two resolutions → distinct DTM/viz steps; invalid entries skipped."""
    from lidar_pipeline.progress import aggregate_tiles
    b = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
    tiles = aggregate_tiles([
        "not json at all",
        None,
        {"tile": b, "phase": "unknown", "state": "ok"},
        {"tile": b, "phase": "dtm", "state": "ok", "res": 0.5},
        {"tile": b, "phase": "dtm", "state": "ok", "res": 0.2},
        {"tile": b, "phase": "viz", "state": "ok", "detail": "aspect", "res": 0.2},
        {"tile": b, "phase": "tile", "state": "ok"},
    ], viz_labels={"aspect": "Aspect"})
    assert len(tiles) == 1
    labels = [s["label"] for s in tiles[0]["steps"]]
    assert labels == ["DTM 0.5 m", "DTM 0.2 m", "Aspect 0.2 m"]
    # The secondary resolution (0.2 m) gets a suffix, the primary one (0.5 m) does not


def test_aggregate_multiple_tiles_ordered():
    """One entry per tile, in order of first appearance."""
    from lidar_pipeline.progress import aggregate_tiles
    a = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
    b = "LHD_FXX_1055_6882_PTS_LAMB93_IGN69"
    tiles = aggregate_tiles([
        {"tile": a, "phase": "download", "state": "ok"},
        {"tile": b, "phase": "download", "state": "start", "detail": "IGN catalogue"},
        {"tile": a, "phase": "tile", "state": "ok"},
    ])
    assert [t["short"] for t in tiles] == ["1054-6882", "1055-6882"]
    assert tiles[0]["state"] == "done"
    assert tiles[1]["state"] == "running"