"""Tests for the XYZ tile pyramid (OpenStreetMap scheme)."""

import math
from pathlib import Path


# ---------------------------------------------------------------------------
# Fixtures: fake tiles following the pipeline conventions
# ---------------------------------------------------------------------------

def _basename(col, row):
    return f"LHD_FXX_{col:04d}_{row:04d}_PTS_LAMB93_IGN69"


def _make_dalle(output_dir, col, row, viz_keys, color=(200, 30, 30), px=64,
                thumbs=True):
    """Create a tile (image + thumbnails) as the pipeline would."""
    from PIL import Image
    base = _basename(col, row)
    vis = Path(output_dir) / "visualisations" / base
    vis.mkdir(parents=True, exist_ok=True)
    thumb_dir = Path(output_dir) / "index_thumbs"
    thumb_dir.mkdir(parents=True, exist_ok=True)
    for key in viz_keys:
        Image.new("RGB", (px, px), color).save(
            str(vis / f"{base}_{key}.webp"), format="WEBP", lossless=True)
        if thumbs:
            Image.new("RGB", (64, 64), color).save(
                str(thumb_dir / f"{base}_{key}.jpg"), format="JPEG", quality=90)
            Image.new("RGB", (64, 64), color).save(
                str(thumb_dir / f"{base}_{key}_mid.jpg"), format="JPEG", quality=90)
    return base


def _tile_of_cell(col, row, z):
    """Indices (x, y) of the level-z tile containing the centre of a source tile."""
    from lidar_pipeline.tiles import _transformer
    lon, lat = _transformer("EPSG:2154", "EPSG:4326").transform(
        col * 1000 + 500, (row - 1) * 1000 + 500)
    n = 2 ** z
    x = int((lon + 180.0) / 360.0 * n)
    rad = math.radians(lat)
    y = int((1.0 - math.log(math.tan(rad) + 1 / math.cos(rad)) / math.pi) / 2.0 * n)
    return x, y


# ---------------------------------------------------------------------------
# Grid geometry
# ---------------------------------------------------------------------------

def test_tile_bounds_3857_known_values():
    """z0 = the whole world; z1/x1/y0 = north-east quadrant."""
    from lidar_pipeline.tiles import ORIGIN, tile_bounds_3857
    w, s, e, n = tile_bounds_3857(0, 0, 0)
    assert (round(w), round(s), round(e), round(n)) == (
        round(-ORIGIN), round(-ORIGIN), round(ORIGIN), round(ORIGIN))
    w, s, e, n = tile_bounds_3857(1, 1, 0)
    assert abs(w) < 1e-6 and abs(s) < 1e-6
    assert abs(e - ORIGIN) < 1e-6 and abs(n - ORIGIN) < 1e-6


def test_tile_latitude_and_resolution():
    """Centre latitude and ground resolution (0.2 m/px ≈ z19 in France)."""
    from lidar_pipeline.tiles import (TILE_MAX_NATIVE_Z, target_resolution,
                                      tile_latitude)
    assert abs(tile_latitude(0, 0)) < 1e-9
    assert abs(tile_latitude(1, 0) - 66.51) < 0.05
    # Native-level tile at the latitude of metropolitan France
    z = TILE_MAX_NATIVE_Z
    y = int((1.0 - math.log(math.tan(math.radians(47)) + 1 / math.cos(math.radians(47)))
             / math.pi) / 2.0 * 2 ** z)
    res = target_resolution(z, y)
    assert 0.15 < res < 0.25, res
    # @2x: twice as fine for the same (z, x, y)
    assert abs(target_resolution(z, y, scale=2) - res / 2) < 1e-9


def test_tile_bounds_l93_covers_cell():
    """The L93 extent of a tile does contain the source tile it covers."""
    from lidar_pipeline.tiles import tile_bounds_l93
    col, row, z = 1054, 6882, 14
    x, y = _tile_of_cell(col, row, z)
    min_x, min_y, max_x, max_y = tile_bounds_l93(z, x, y)
    assert min_x < col * 1000 + 500 < max_x
    assert min_y < (row - 1) * 1000 + 500 < max_y


def test_perspective_coeffs_identity_and_scale():
    """Identity → neutral coefficients; scaling → exact factor."""
    from lidar_pipeline.tiles import perspective_coeffs
    quad = [(0, 0), (256, 0), (256, 256), (0, 256)]
    c = perspective_coeffs(quad, quad)
    assert [round(v, 9) for v in c] == [1, 0, 0, 0, 1, 0, 0, 0]
    # Output twice as large as the source: Pillow samples at x/2
    c = perspective_coeffs([(0, 0), (512, 0), (512, 512), (0, 512)], quad)
    assert abs(c[0] - 0.5) < 1e-9 and abs(c[4] - 0.5) < 1e-9


def test_perspective_coeffs_degenerate():
    """Degenerate quadrilateral (tile shrunk to a point) → None, no exception."""
    from lidar_pipeline.tiles import perspective_coeffs
    flat = [(0, 0), (0, 0), (0, 0), (0, 0)]
    assert perspective_coeffs(flat, [(0, 0), (1, 0), (1, 1), (0, 1)]) is None


def test_zoom_supported_bounds():
    """Served zoom range; @2x stops one level earlier (512 px)."""
    from lidar_pipeline.tiles import (TILE_MAX_NATIVE_Z, TILE_MIN_Z,
                                      zoom_supported)
    assert not zoom_supported(TILE_MIN_Z - 1)
    assert zoom_supported(TILE_MIN_Z)
    assert zoom_supported(TILE_MAX_NATIVE_Z)
    assert not zoom_supported(TILE_MAX_NATIVE_Z + 1)
    assert not zoom_supported(TILE_MAX_NATIVE_Z, scale=2)
    assert zoom_supported(TILE_MAX_NATIVE_Z - 1, scale=2)


# ---------------------------------------------------------------------------
# Source index
# ---------------------------------------------------------------------------

def test_source_index_and_layers(tmp_path, monkeypatch):
    """The index lists the layers present and their tiers (coarse → fine)."""
    from lidar_pipeline import index, tiles
    monkeypatch.setattr(index, "PANEL_VIZ", None)
    _make_dalle(tmp_path, 1054, 6882, ["aspect", "slope"])
    layers = tiles.source_index(tmp_path, force=True)
    assert set(layers) == {"aspect", "slope"}
    tiers = layers["aspect"][(1054, 6882)]
    # thumbnail (3.9 m/px) → intermediate (1.56) → tile (0.5)
    assert [round(t[0].res, 2) for t in tiers] == [3.91, 1.56, 0.5]
    assert tiles.available_layers(tmp_path) == ["slope", "aspect"] or \
        set(tiles.available_layers(tmp_path)) == {"slope", "aspect"}


def test_available_layers_follow_panel(tmp_path, monkeypatch):
    """Layers on disk outside PANEL_VIZ: neither displayed nor served."""
    from lidar_pipeline import index, tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect", "relief_oriente"])
    tiles.source_index(tmp_path, force=True)
    monkeypatch.setattr(index, "PANEL_VIZ", ("relief_oriente",))
    assert tiles.available_layers(tmp_path) == ["relief_oriente"]


def test_grid_bounds(tmp_path):
    """L93 and WGS84 extent of the available grid."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    _make_dalle(tmp_path, 1055, 6883, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    assert tiles.grid_bounds_l93(tmp_path) == (1054000.0, 6881000.0,
                                               1056000.0, 6883000.0)
    w, s, e, n = tiles.grid_bounds_wgs84(tmp_path)
    assert 7.0 < w < 8.5 and 47.5 < s < 49.5 and e > w and n > s


def test_pick_tier_by_resolution(tmp_path):
    """Chosen tier: the coarsest whose resolution is sufficient for the tile."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiers = tiles.source_index(tmp_path, force=True)["aspect"][(1054, 6882)]
    assert tiles._pick_tier(tiers, 10.0)[0].res == tiers[0][0].res      # thumbnail
    assert tiles._pick_tier(tiers, 2.0)[0].res == tiers[1][0].res       # intermediate
    assert tiles._pick_tier(tiers, 0.2)[0].res == tiers[-1][0].res      # tile
    # Target finer than anything available: the finest tier is kept
    assert tiles._pick_tier(tiers, 0.01)[0].res == tiers[-1][0].res


def test_sources_in_bbox_picks_tier(tmp_path):
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"], px=64)
    tiles.source_index(tmp_path, force=True)
    bbox = (1054000.0, 6881000.0, 1055000.0, 6882000.0)
    fine = tiles.sources_in_bbox(tmp_path, "aspect", bbox, 0.2)
    coarse = tiles.sources_in_bbox(tmp_path, "aspect", bbox, 50.0)
    assert fine and coarse and fine[0][0] == (1054, 6882)
    assert min(s.res for _c, s in fine) <= min(s.res for _c, s in coarse)
    assert tiles.sources_in_bbox(tmp_path, "aspect", (0, 0, 10, 10), 0.2) == []
    img = tiles.load_source(fine[0][1])
    assert img is not None and img.mode in ("RGB", "RGBA")


def test_subtiles_preferred_over_full_dalle(tmp_path):
    """index_subtiles quadrants serve as the fine tier (4× less to decode)."""
    import pytest
    from PIL import Image
    from lidar_pipeline import tiles
    base = _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    sub = tmp_path / "index_subtiles"
    sub.mkdir(parents=True, exist_ok=True)
    try:
        for i in range(2):
            for j in range(2):
                Image.new("RGB", (32, 32), (10, 10, 10)).save(
                    str(sub / f"{base}_aspect_{i}_{j}.avif"), format="AVIF")
    except Exception:
        pytest.skip("AVIF encoder unavailable")
    tiers = tiles.source_index(tmp_path, force=True)["aspect"][(1054, 6882)]
    quads = next(t for t in tiers if len(t) == 4)
    # At equal resolution, quadrants come BEFORE the whole tile
    assert tiers.index(quads) < tiers.index(next(t for t in tiers if len(t) == 1
                                                 and t[0].path.suffix == ".webp"))
    assert len(quads) == 4
    # Quadrant extents: four adjoining half-kilometre squares
    assert {q.bounds for q in quads} == {
        (1054000.0, 6881000.0, 1054500.0, 6881500.0),
        (1054500.0, 6881000.0, 1055000.0, 6881500.0),
        (1054000.0, 6881500.0, 1054500.0, 6882000.0),
        (1054500.0, 6881500.0, 1055000.0, 6882000.0)}


# ---------------------------------------------------------------------------
# Rendering and cache
# ---------------------------------------------------------------------------

def test_render_tile_paints_cell(tmp_path):
    """A tile over the source tile is painted; elsewhere it is empty."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"], color=(200, 30, 30))
    tiles.source_index(tmp_path, force=True)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    img = tiles.render_tile(tmp_path, "aspect", z, x, y)
    assert img is not None and img.size == (256, 256)
    r, g, b, a = img.getpixel((128, 128))
    assert a == 255 and r > 150 and g < 90 and b < 90
    # Distant tile (another continent): no source
    assert tiles.render_tile(tmp_path, "aspect", z, 1, 1) is None


def test_render_tile_scale2(tmp_path):
    """`scale=2` renders the same extent at 512 px (@2x convention)."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    img = tiles.render_tile(tmp_path, "aspect", z, x, y, scale=2)
    assert img is not None and img.size == (512, 512)


def test_tile_edges_transparent_outside_data(tmp_path):
    """Outside the source tiles' extent, the tile stays transparent (overlayable)."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    # Zoom where a tile is much larger than the source tile: edges are empty
    z = 11
    x, y = _tile_of_cell(1054, 6882, z)
    img = tiles.render_tile(tmp_path, "aspect", z, x, y)
    assert img is not None
    assert img.getpixel((0, 0))[3] == 0
    assert img.getpixel((255, 255))[3] == 0


def test_get_tile_cache_and_staleness(tmp_path, monkeypatch):
    """The disk cache is reused, then invalidated by a regenerated source tile."""
    from lidar_pipeline import tiles as _t
    monkeypatch.setattr(_t, "TILE_EVEN_LEVELS", False)   # historical rule: everything is stored
    monkeypatch.setattr(_t, "TILE_CACHE_MAX_Z", 99)
    import os
    import time
    from lidar_pipeline import tiles
    base = _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    data = tiles.get_tile(tmp_path, "aspect", z, x, y)
    assert data and data[:8] == b"\x89PNG\r\n\x1a\n"
    cache = tiles.tile_cache_path(tmp_path, "aspect", z, x, y)
    assert cache.is_file()
    first = cache.stat().st_mtime

    # Unchanged: the cached tile is served again as-is
    time.sleep(0.02)
    assert tiles.get_tile(tmp_path, "aspect", z, x, y) == data
    assert cache.stat().st_mtime == first

    # Regenerated source tile (newer mtime): the tile is recomputed
    src = tmp_path / "visualisations" / base / f"{base}_aspect.webp"
    newer = first + 10
    os.utime(src, (newer, newer))
    for f in (tmp_path / "index_thumbs").iterdir():
        os.utime(f, (newer, newer))
    tiles.source_index(tmp_path, force=True)
    tiles.get_tile(tmp_path, "aspect", z, x, y)
    assert cache.stat().st_mtime > first


def test_get_tile_empty_marker(tmp_path, monkeypatch):
    """Tile without data: None + remembered .empty marker."""
    from lidar_pipeline import tiles as _t
    monkeypatch.setattr(_t, "TILE_EVEN_LEVELS", False)   # historical rule: everything is stored
    monkeypatch.setattr(_t, "TILE_CACHE_MAX_Z", 99)
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    assert tiles.get_tile(tmp_path, "aspect", 15, 1, 1) is None
    assert tiles.empty_marker_exists(tmp_path, "aspect", 15, 1, 1)


def test_get_tile_webp_scale2(tmp_path):
    """@2x tier in WebP: 512 px, cache path distinct from the 256 px one."""
    from PIL import Image
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    data = tiles.get_tile(tmp_path, "aspect", z, x, y, scale=2, fmt="webp")
    assert data and data[:4] == b"RIFF"
    path = tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 2, "webp")
    assert path.name.endswith("@2x.webp") and path.is_file()
    import io
    assert Image.open(io.BytesIO(data)).size == (512, 512)


def test_transparent_tile_is_fully_transparent():
    """The fallback tile (empty area) is fully transparent."""
    import io
    from PIL import Image
    from lidar_pipeline import tiles
    img = Image.open(io.BytesIO(tiles.transparent_tile()))
    assert img.size == (256, 256)
    assert img.convert("RGBA").getextrema()[3] == (0, 0)


def test_tiles_in_bounds_and_warm(tmp_path):
    """Warm-up: the tiles of the extent are computed and cached."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    bounds = tiles.grid_bounds_wgs84(tmp_path)
    assert len(tiles.tiles_in_bounds(bounds, 14)) >= 1
    report = tiles.warm(tmp_path, ["aspect"], 12, 13)
    assert report["rendues"] >= 1 and report["limite"] is False
    assert any((tmp_path / tiles.TILE_DIRNAME / "aspect").rglob("*.png"))


def test_tiles_stamp_follows_sources(tmp_path):
    """The global version follows the most recent source-tile mtime."""
    import os
    from lidar_pipeline import tiles
    base = _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    first = tiles.tiles_stamp(tmp_path)
    src = tmp_path / "visualisations" / base / f"{base}_aspect.webp"
    os.utime(src, (first / 1000 + 60, first / 1000 + 60))
    tiles.source_index(tmp_path, force=True)
    assert tiles.tiles_stamp(tmp_path) > first


def test_source_cache_respects_memory_budget(tmp_path, monkeypatch):
    """The source image cache evicts by a byte budget, not by a count.

    A decoded 5000² tile weighs ~100 MB (4 bytes/px): an "N entries" cache
    would overflow the memory of a small machine (Raspberry Pi).
    """
    from PIL import Image
    from lidar_pipeline import tiles
    tiles.clear_source_cache()
    # Deliberately tiny budget: only one image fits at a time
    monkeypatch.setattr(tiles, "SOURCE_CACHE_BYTES", 40 * 40 * 3 * 2 - 1)
    paths = []
    for i in range(3):
        f = tmp_path / f"src{i}.png"
        Image.new("RGB", (40, 40), (i * 40, 0, 0)).save(str(f))
        paths.append(f)
    for f in paths:
        tiles._open_source(str(f), f.stat().st_mtime)
    assert len(tiles._source_cache) == 1
    # The last source used is the one that remains
    assert str(paths[-1]) == list(tiles._source_cache)[0][0]
    tiles.clear_source_cache()
    assert tiles._source_cache == {}


def test_source_cache_holds_plain_decoded_images(tmp_path):
    """The cache holds pixels only: no open file, no decoder.

    An open AVIF image keeps its decoder (libavif/dav1d buffers): ~43 MB
    retained per 2500² quadrant instead of 25 — the Pi's container (1 GB)
    was killed by the OOM killer when browsing at high zoom. The budget
    counts 4 bytes/pixel: PIL stores RGB on 32 bits.
    """
    from PIL import Image
    from lidar_pipeline import tiles
    tiles.clear_source_cache()
    f = tmp_path / "src.png"
    Image.new("RGB", (40, 30), (10, 20, 30)).save(str(f))
    img = tiles._open_source(str(f), f.stat().st_mtime)
    assert type(img) is Image.Image
    assert getattr(img, "fp", None) is None
    assert img.info["_bytes"] == 40 * 30 * 4
    assert img.getpixel((0, 0))[:3] == (10, 20, 30)
    tiles.clear_source_cache()


def test_source_cache_keyed_by_mtime(tmp_path):
    """A rewritten source is not served again from the cache."""
    import os
    from PIL import Image
    from lidar_pipeline import tiles
    tiles.clear_source_cache()
    f = tmp_path / "src.png"
    Image.new("RGB", (8, 8), (10, 10, 10)).save(str(f))
    first = tiles._open_source(str(f), f.stat().st_mtime)
    assert first.getpixel((0, 0))[:3] == (10, 10, 10)
    Image.new("RGB", (8, 8), (200, 200, 200)).save(str(f))
    os.utime(f, (f.stat().st_mtime + 5, f.stat().st_mtime + 5))
    second = tiles._open_source(str(f), f.stat().st_mtime)
    assert second.getpixel((0, 0))[:3] == (200, 200, 200)
    tiles.clear_source_cache()


def test_png_palette_option_shrinks_tiles(tmp_path, monkeypatch):
    """`LIDAR_TILE_PNG_PALETTE=1` shrinks the canonical PNG (palette + alpha).

    Measured on a realistic rendering (noisy color ramp): a flat-colored tile
    already compresses better in RGBA than with a palette, it would prove
    nothing.
    """
    import io
    import random
    from PIL import Image
    from lidar_pipeline import tiles

    base = _basename(1054, 6882)
    vis = tmp_path / "visualisations" / base
    vis.mkdir(parents=True, exist_ok=True)
    rng = random.Random(7)
    img = Image.new("RGB", (256, 256))
    px = img.load()
    for j in range(256):
        for i in range(256):
            px[i, j] = (min(255, i + rng.randint(0, 12)),
                        min(255, j + rng.randint(0, 12)),
                        rng.randint(40, 90))
    img.save(str(vis / f"{base}_aspect.webp"), format="WEBP", lossless=True)
    tiles.source_index(tmp_path, force=True)

    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    rendered = tiles.render_tile(tmp_path, "aspect", z, x, y)
    lossless = tiles._encode(rendered, "png")
    monkeypatch.setattr(tiles, "PNG_PALETTE", True)
    palette = tiles._encode(rendered, "png")
    assert len(palette) < len(lossless)
    # The palettized PNG remains a readable PNG, at the right size, with alpha
    out = Image.open(io.BytesIO(palette))
    assert out.size == (256, 256)
    assert out.convert("RGBA").getextrema()[3][1] == 255


# ---------------------------------------------------------------------------
# Upstream tile server (LIDAR_SOURCE_URL)
# ---------------------------------------------------------------------------

def _remote_payload():
    """/api/tiles payload as served by mapserve (tile server)."""
    base = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69_r0p2"
    tiles = []
    for i in range(2):
        for j in range(2):
            tiles.append({
                "col": 1054, "row": 6882, "resolution": 0.2,
                "dir_name": base, "sub_i": i, "sub_j": j, "sub_k": 2,
                "viz": {"aspect": {
                    "thumb": f"index_subtiles/{base}_aspect_{i}_{j}_thumb160.webp?v=1700000000000",
                    "mid": f"index_subtiles/{base}_aspect_{i}_{j}_mid.webp?v=1700000000000",
                    "full": f"index_subtiles/{base}_aspect_{i}_{j}.avif?v=1700000000000",
                }},
            })
    return {"tiles": tiles, "viz_meta": {"aspect": {"label": "Aspect"}}}


def test_remote_index_lists_tiles_without_local_data(tmp_path, monkeypatch):
    """Without any local data, the index comes from the upstream (quadrants placed)."""
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
    monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: _remote_payload())
    layers = tiles.source_index(tmp_path, force=True)
    assert set(layers) == {"aspect"}
    tiers = layers["aspect"][(1054, 6882)]
    # Three tiers (thumbnail, intermediate, quadrant) × 4 quadrants each
    assert [len(t) for t in tiers] == [4, 4, 4]
    assert [round(t[0].res, 2) for t in tiers] == [3.12, 0.78, 0.2]
    fine = tiers[-1][0]
    assert fine.url.startswith("http://amont:8973/index_subtiles/")
    assert fine.path == tmp_path / fine.url.split("8973/")[1].split("?")[0]
    # The reference date comes from the announced ?v=, without downloading anything
    assert fine.mtime() == 1700000000.0
    assert tiles.grid_bounds_l93(tmp_path) == (1054000.0, 6881000.0,
                                               1055000.0, 6882000.0)


def test_remote_source_downloaded_on_demand(tmp_path, monkeypatch):
    """A remote source is only fetched by the first rendering that needs it."""
    from PIL import Image
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
    monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: _remote_payload())
    tiles.clear_source_cache()

    import io
    buf = io.BytesIO()
    Image.new("RGB", (64, 64), (12, 200, 90)).save(buf, format="WEBP", lossless=True)
    body = buf.getvalue()
    fetched = []

    def fake_fetch(url, dest):
        fetched.append(url)
        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_bytes(body)
        return True

    monkeypatch.setattr(tiles, "_fetch_source", fake_fetch)
    z = 15
    x, y = _tile_of_cell(1054, 6882, z)
    img = tiles.render_tile(tmp_path, "aspect", z, x, y)
    assert img is not None
    assert fetched, "no source fetched"
    r, g, b, a = img.getpixel((128, 128))
    assert a == 255 and g > 150 and r < 80
    # Fetched files land in the local cache, at the same path
    assert list(tmp_path.rglob("*.webp")) or list(tmp_path.rglob("*.avif"))
    # Second rendering: no more downloads (local cache)
    before = len(fetched)
    tiles.render_tile(tmp_path, "aspect", z, x, y)
    assert len(fetched) == before
    tiles.clear_source_cache()


def test_remote_index_failure_keeps_local(tmp_path, monkeypatch):
    """Upstream unreachable: the local index is still served, without exception."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["slope"])
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
    monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: None)
    layers = tiles.source_index(tmp_path, force=True)
    assert set(layers) == {"slope"}


def test_cached_tile_states(tmp_path):
    """`cached_tile`: cache-only read — fresh / pending / empty, no rendering."""
    from lidar_pipeline import tiles
    _make_dalle(tmp_path, 1054, 6882, ["slope"])
    x, y = _tile_of_cell(1054, 6882, 14)
    data, state = tiles.cached_tile(tmp_path, "slope", 14, x, y)
    assert state == "pending" and data is None
    assert not tiles.tile_cache_path(tmp_path, "slope", 14, x, y).exists()
    tiles.get_tile(tmp_path, "slope", 14, x, y)  # rendering (maintenance, warm…)
    data, state = tiles.cached_tile(tmp_path, "slope", 14, x, y)
    assert state == "fresh" and data
    # Outside the data: empty state + marker set, still without rendering
    data, state = tiles.cached_tile(tmp_path, "slope", 14, 0, 0)
    assert state == "empty" and data is None
    assert tiles.empty_marker_exists(tmp_path, "slope", 14, 0, 0)


def test_storage_policy_even_levels_up_to_max(monkeypatch):
    """Reduced storage: even standard levels ≤ LIDAR_TILE_CACHE_MAX_Z (an
    @2x tile at level z equals a 256 px tile at z+1)."""
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", True)
    monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 16)
    assert tiles.zoom_cached(16, 1) and not tiles.zoom_cached(15, 1)
    assert tiles.zoom_cached(15, 2) and not tiles.zoom_cached(16, 2)      # @2x z15 = standard z16
    assert not tiles.zoom_cached(18, 1) and not tiles.zoom_cached(17, 2)  # fine level: on the fly


def test_unstored_level_rendered_on_the_fly_without_disk(tmp_path, monkeypatch):
    """Unstored level: tile rendered, nothing written to disk, second
    request served by the memory cache."""
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "TILE_EVEN_LEVELS", True)
    monkeypatch.setattr(tiles, "TILE_CACHE_MAX_Z", 16)
    _make_dalle(tmp_path, 1054, 6882, ["aspect"])
    tiles.source_index(tmp_path, force=True)
    z = 17
    x, y = _tile_of_cell(1054, 6882, z)
    calls = []
    real = tiles.render_tile
    monkeypatch.setattr(tiles, "render_tile", lambda *a, **k: calls.append(1) or real(*a, **k))
    data = tiles.get_tile(tmp_path, "aspect", z, x, y, 1, "png")
    assert data and data[:4] == b"\x89PNG"
    assert not tiles.tile_cache_path(tmp_path, "aspect", z, x, y, 1, "png").exists()
    assert tiles.get_tile(tmp_path, "aspect", z, x, y, 1, "png") == data and len(calls) == 1


def test_remote_payload_failure_not_retried_each_call(monkeypatch):
    """Upstream unreachable with no known inventory: the failure is remembered (TTL).

    Otherwise every call (several per /api/map/meta) pays the connection
    timeout again: the interface no longer loaded when the worker was down.
    """
    import urllib.request
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont:8973")
    monkeypatch.setattr(tiles, "_remote_cache",
                        {"payload": None, "at": 0.0, "index": None, "root": None})
    calls = []

    def boom(*a, **k):
        calls.append(1)
        raise OSError("host unreachable")

    monkeypatch.setattr(urllib.request, "urlopen", boom)
    for _ in range(3):
        assert tiles._remote_payload() is None
    assert len(calls) == 1


def test_fetch_source_offline_breaker(tmp_path, monkeypatch):
    """Failed upstream source: the following ones fail without waiting for the
    network timeout during the suspension (the maintenance falls back to local
    rendering)."""
    import urllib.request
    from lidar_pipeline import tiles
    monkeypatch.setattr(tiles, "_SOURCE_OFFLINE", {"until": 0.0})
    calls = []

    def boom(*a, **k):
        calls.append(1)
        raise OSError("host unreachable")

    monkeypatch.setattr(urllib.request, "urlopen", boom)
    assert tiles._fetch_source("http://amont/a", tmp_path / "a.avif") is False
    assert tiles._fetch_source("http://amont/b", tmp_path / "b.avif") is False
    assert len(calls) == 1


def test_fetched_source_dated_to_upstream_version(tmp_path, monkeypatch):
    """A fetched source carries the upstream version date: when the upstream
    goes down, the local index finds the same dates and already-rendered tiles
    stay fresh (no whole pyramid to redo)."""
    from lidar_pipeline import tiles

    def fake_fetch(url, dest):
        dest.write_bytes(b"x")
        return True

    monkeypatch.setattr(tiles, "_fetch_source", fake_fetch)
    src = tiles._Source(tmp_path / "q.avif", (0, 0, 1, 1), 0.2,
                        url="http://amont/q.avif", version=1700000000000)
    assert src.ensure() is True
    assert (tmp_path / "q.avif").stat().st_mtime == 1700000000.0


def test_tile_refreshed_when_older_dalle_appears(tmp_path):
    """Source tile entering the inventory AFTER an XYZ tile covering it was
    rendered, but with an older version date (written before, inventoried
    after): the XYZ tile must become stale — otherwise a permanent hole at
    that level, on disk, in memory and in the browser (unchanged stamp)."""
    import io
    import os
    from PIL import Image
    from lidar_pipeline import tiles
    tiles.clear_source_cache()
    tiles._mem_tiles.clear()
    z = 10
    assert _tile_of_cell(1054, 6882, z) == _tile_of_cell(1055, 6882, z)
    x, y = _tile_of_cell(1054, 6882, z)
    _make_dalle(tmp_path, 1054, 6882, ["slope"], color=(200, 30, 30))
    tiles.source_index(tmp_path, force=True)
    first = tiles.get_tile(tmp_path, "slope", z, x, y)
    assert first is not None
    stamp = tiles.tiles_stamp(tmp_path)

    _make_dalle(tmp_path, 1055, 6882, ["slope"], color=(30, 200, 30))
    for f in tmp_path.rglob("*1055_6882*"):
        os.utime(f, (1_000_000, 1_000_000))   # version older than the XYZ tile
    tiles.source_index(tmp_path, force=True)
    assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "pending"
    second = tiles.get_tile(tmp_path, "slope", z, x, y)
    img = Image.open(io.BytesIO(second)).convert("RGBA")
    assert any(g > 150 and r < 80 and a == 255
               for r, g, b, a in img.getdata()), "new source tile missing from the XYZ tile"
    assert tiles.tiles_stamp(tmp_path) > stamp
    # Persistent registry: a restart invalidates nothing again
    tiles._index_cache.clear()
    tiles.source_index(tmp_path, force=True)
    assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "fresh"
    tiles.clear_source_cache()


def test_existing_cache_without_registry_is_refreshed_once(tmp_path):
    """Upgrade: tile cache present but no registry — the existing tiles
    (possibly with holes) are made stale exactly once."""
    from lidar_pipeline import tiles
    tiles._seen_cache.clear()
    tiles._mem_tiles.clear()
    z = 10
    x, y = _tile_of_cell(1054, 6882, z)
    _make_dalle(tmp_path, 1054, 6882, ["slope"])
    tiles.source_index(tmp_path, force=True)
    assert tiles.get_tile(tmp_path, "slope", z, x, y) is not None
    (tmp_path / tiles.TILE_DIRNAME / tiles._SEEN_FILE).unlink()   # previous version
    tiles._seen_cache.clear()
    tiles.source_index(tmp_path, force=True)
    assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "pending"
    tiles.get_tile(tmp_path, "slope", z, x, y)
    tiles._seen_cache.clear()
    tiles.source_index(tmp_path, force=True)
    assert tiles.cached_tile(tmp_path, "slope", z, x, y)[1] == "fresh"


def test_webp_subtiles_indexed_and_rendered_nearest(tmp_path):
    """Density layer: lossless .webp quadrants recognised as the fine tier,
    and rendered with nearest neighbour (16 exact greys even when upscaled)."""
    from PIL import Image
    from lidar_pipeline import tiles
    base = _make_dalle(tmp_path, 1054, 6882, ["densite_sol"])
    sub = tmp_path / "index_subtiles"
    sub.mkdir(parents=True, exist_ok=True)
    for i in range(2):
        for j in range(2):
            im = Image.new("L", (8, 8), 0)
            im.paste(255, (0, 0, 4, 8))  # half white, half black
            im.save(str(sub / f"{base}_densite_sol_{i}_{j}.webp"), format="WEBP", lossless=True)
    tiers = tiles.source_index(tmp_path, force=True)["densite_sol"][(1054, 6882)]
    quads = next(t for t in tiers if len(t) == 4)
    assert all(q.path.suffix == ".webp" for q in quads)
    assert "densite_sol" in tiles.NEAREST_LAYERS
    # z17: target resolution (~0.8 m) reached by the quadrants (0.5 m here)
    x, y = _tile_of_cell(1054, 6882, 17)
    img = tiles.render_tile(tmp_path, "densite_sol", 17, x, y)
    assert img is not None
    grays = {p[0] for p in img.getdata() if p[3] == 255}
    assert grays <= {0, 255}  # no invented intermediate grey


def test_remote_quality_persisted_locally(tmp_path, monkeypatch):
    """The upstream quality table is copied into local sidecars (self-sufficient Pi)."""
    from lidar_pipeline import tiles
    from lidar_pipeline.quality import read_quality
    payload = _remote_payload()
    base = "LHD_FXX_1054_6882_PTS_LAMB93_IGN69"
    payload["quality"] = {base: {"version": 1, "ground_density": 3.0},
                          "../evasion": {"version": 1}}
    monkeypatch.setattr(tiles, "REMOTE_SOURCE_URL", "http://amont")
    monkeypatch.setattr(tiles, "_remote_payload", lambda force=False: payload)
    monkeypatch.setattr(tiles, "_remote_cache", dict(tiles._remote_cache, index=None, built=None))
    tiles._remote_index(tmp_path, force=True)
    assert read_quality(tmp_path, base) == {"version": 1, "ground_density": 3.0}
    assert not (tmp_path / "evasion.json").exists()
    assert list((tmp_path / "quality").iterdir()) == [tmp_path / "quality" / f"{base}.json"]