lidar_rendu/lidar_pipeline/tests/test_fetch_ign.py

"""Tests du téléchargement des dalles LiDAR HD de l'IGN (fetch_ign)."""


def test_parse_tile_specs():
    """Accepte 'col,row', 'col:row' et ignore les espaces."""
    from lidar_pipeline.fetch_ign import parse_tile_specs
    assert parse_tile_specs(["1055,6882"]) == [(1055, 6882)]
    assert parse_tile_specs(["1055:6883", " 651 , 6630 "]) == [(1055, 6883), (651, 6630)]


def test_parse_tile_specs_rejects_invalid():
    """Les spécifications mal formées lèvent une erreur explicite."""
    import pytest
    from lidar_pipeline.fetch_ign import parse_tile_specs
    with pytest.raises(ValueError):
        parse_tile_specs(["1055"])
    with pytest.raises(ValueError):
        parse_tile_specs(["abc,def"])
    with pytest.raises(ValueError):
        parse_tile_specs(["1055,6882,9999"])


def test_tile_filename_pads_coordinates():
    """Le nom de fichier DALLE utilise des coordonnées à 4 chiffres."""
    from lidar_pipeline.fetch_ign import tile_filename
    assert tile_filename(1055, 6882) == "LHD_FXX_1055_6882_PTS_LAMB93_IGN69.copc.laz"
    assert tile_filename(651, 6630) == "LHD_FXX_0651_6630_PTS_LAMB93_IGN69.copc.laz"


def test_match_feature_uses_coordonnees_nw():
    """La correspondance se fait sur lidarhd:coordonnees_NW (format 0816-6847)."""
    from lidar_pipeline.fetch_ign import match_feature
    features = [
        {"id": "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_NE",
         "properties": {"lidarhd:coordonnees_NW": "1054-6882"},
         "assets": {"data": {"href": "https://example.org/a.copc.laz"}}},
        {"id": "LHD_FXX_1055_6882_PTS_LAMB93_IGN69_NE",
         "properties": {"lidarhd:coordonnees_NW": "1055-6882"},
         "assets": {"data": {"href": "https://example.org/b.copc.laz"}}},
    ]
    assert match_feature(features, 1055, 6882) is features[1]
    assert match_feature(features, 9999, 9999) is None


def test_find_tile_url_matches_and_returns_href(monkeypatch):
    """find_tile_url interroge le STAC et retourne l'href de l'asset data."""
    from lidar_pipeline import fetch_ign

    class FakeResponse:
        def __init__(self, payload):
            self._payload = payload.encode("utf-8")

        def read(self):
            return self._payload

        def __enter__(self):
            return self

        def __exit__(self, *args):
            return False

    payload = ('{"features": [{"properties": {"lidarhd:coordonnees_NW": "1055-6882"},'
               '"assets": {"data": {"href": "https://data.geopf.fr/x.copc.laz"}}}]}')
    captured = {}

    def fake_urlopen(req, timeout=None):
        captured["url"] = req.full_url
        return FakeResponse(payload)

    monkeypatch.setattr(fetch_ign.urllib.request, "urlopen", fake_urlopen)
    url = fetch_ign.find_tile_url(1055, 6882)
    assert url == "https://data.geopf.fr/x.copc.laz"
    assert "api.stac.teledetection.fr" in captured["url"]
    assert "bbox=" in captured["url"]


def test_fetch_tiles_skips_existing_and_generated(tmp_path, monkeypatch):
    """Pas de téléchargement si le LAZ existe ou si les visualisations existent."""
    from lidar_pipeline import fetch_ign

    input_dir = tmp_path / "input"
    input_dir.mkdir()
    output_dir = tmp_path / "output"
    vis_dir = output_dir / "visualisations" / "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
    vis_dir.mkdir(parents=True)

    # 1054,6882 : visualisations déjà générées
    # 1055,6882 : LAZ déjà présent
    existing = input_dir / fetch_ign.tile_filename(1055, 6882)
    existing.write_bytes(b"naze")

    def fail_download(*args, **kwargs):
        raise AssertionError("ne doit pas être appelé")

    monkeypatch.setattr(fetch_ign, "find_tile_url", fail_download)
    result = fetch_ign.fetch_tiles(input_dir, [(1054, 6882), (1055, 6882)],
                                   output_dir=output_dir)
    assert result == []


def _make_viz_dirs(output_dir, 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 = output_dir / "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_fetch_tiles_downloads_incomplete_tile(tmp_path, monkeypatch):
    """Une tuile incomplète (visualisation manquante) est téléchargée, une complète non."""
    from pathlib import Path
    from lidar_pipeline import fetch_ign

    input_dir = tmp_path / "input"
    input_dir.mkdir()
    output_dir = tmp_path / "output"
    _make_viz_dirs(output_dir, 1054, 6882, ["aspect"])            # slope manquant
    _make_viz_dirs(output_dir, 1055, 6882, ["aspect", "slope"])   # complète

    calls = []

    def fake_find_tile_url(col, row, timeout=20):
        calls.append((col, row))
        return "https://example.org/dalle.copc.laz"

    def fake_download(url, dest, timeout=120, chunk=1024 * 1024):
        Path(dest).write_bytes(b"laz")
        return 3

    monkeypatch.setattr(fetch_ign, "find_tile_url", fake_find_tile_url)
    monkeypatch.setattr(fetch_ign, "download_file", fake_download)

    result = fetch_ign.fetch_tiles(input_dir, [(1054, 6882), (1055, 6882)],
                                   output_dir=output_dir,
                                   only_viz=["aspect", "slope"],
                                   resolutions=(0.5, 0.2))
    # Seule la tuile incomplète est retéléchargée
    assert calls == [(1054, 6882)]
    assert [p.name for p in result] == [fetch_ign.tile_filename(1054, 6882)]


def test_fetch_tiles_force_downloads_complete(tmp_path, monkeypatch):
    """force=True télécharge même une tuile complète (régénération avec --force)."""
    from pathlib import Path
    from lidar_pipeline import fetch_ign

    input_dir = tmp_path / "input"
    input_dir.mkdir()
    output_dir = tmp_path / "output"
    _make_viz_dirs(output_dir, 1054, 6882, ["aspect", "slope"])

    calls = []

    def fake_find_tile_url(col, row, timeout=20):
        calls.append((col, row))
        return "https://example.org/dalle.copc.laz"

    def fake_download(url, dest, timeout=120, chunk=1024 * 1024):
        Path(dest).write_bytes(b"laz")
        return 3

    monkeypatch.setattr(fetch_ign, "find_tile_url", fake_find_tile_url)
    monkeypatch.setattr(fetch_ign, "download_file", fake_download)

    result = fetch_ign.fetch_tiles(input_dir, [(1054, 6882)],
                                   output_dir=output_dir,
                                   only_viz=["aspect", "slope"],
                                   resolutions=(0.5, 0.2), force=True)
    assert calls == [(1054, 6882)]
    assert len(result) == 1