path: root/tests/test_tiles.py

import io
import json
import os
import shutil
import tempfile
import threading
import time
import unittest
from pathlib import Path
from unittest import mock

import numpy as np
from PIL import Image

from mapgen.sphere import latlon_to_xyz

import serve
import tiles as T
from tests.test_serve import Running, built_world, request
from tests.helpers import big_river


def tile_at(lat, lon, z):
    span = 180.0 / 2 ** z
    return int((lon + 180) // span), int((90 - lat) // span)


class TilesTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.root = built_world()
        cls.cache = Path(tempfile.mkdtemp())
        cls.src = T.TileSource(serve.World(cls.root, 2), seed=7, cache_dir=cls.cache)

    @classmethod
    def tearDownClass(cls):
        shutil.rmtree(cls.cache)

    def test_tile_scheme(self):
        self.assertTrue(T.tile_ok(0, 1, 0))
        self.assertFalse(T.tile_ok(0, 2, 0))
        self.assertFalse(T.tile_ok(3, 0, 8))
        self.assertFalse(T.tile_ok(17, 0, 0))
        lat, lon = T.tile_lat_lon(0, 0, 0, np.array([0, 255]))
        self.assertAlmostEqual(lat[0], 90 - 0.5 / 256 * 180)
        self.assertAlmostEqual(lon[1], -180 + 255.5 / 256 * 180)

    def test_coarse_zoom_is_the_build_raster(self):
        f = self.src.fields(0, 0, 0, 0, ("z", "cat"))
        base = self.src.bilinear(self.src.surface(), f["lat"], f["lon"])
        dry = ~f["lake"].astype(bool)                        # (lakes lie flat at their level)
        np.testing.assert_array_equal(f["z"][dry], base[dry])   # the 8 km detail is below this zoom's pixel size

    def test_detail_appears_when_zoomed_in(self):
        f = self.src.fields(12, *tile_at(10.0, 20.0, 12), 0, ("z",))
        base = self.src.bilinear(self.src.raster("elevation"), f["lat"], f["lon"])
        self.assertGreater(float(np.std(f["z"] - base)), 1.0)
        self.assertTrue(np.all(np.isfinite(f["z"])))

    def test_seams_match(self):
        z, x, y = 7, *tile_at(5.0, 10.0, 7)
        a = self.src.fields(z, x, y, 1, ("z", "cat"))
        east = self.src.fields(z, x + 1, y, 1, ("z", "cat"))
        south = self.src.fields(z, x, y + 1, 1, ("z", "cat"))
        np.testing.assert_allclose(a["z"][:, -1], east["z"][:, 1], atol=1e-6)
        np.testing.assert_array_equal(a["holdridge"][:, -1], east["holdridge"][:, 1])
        np.testing.assert_allclose(a["z"][-1, :], south["z"][1, :], atol=1e-6)

    def test_antimeridian_seam(self):
        z = 6
        y = tile_at(0.0, 0.0, z)[1]
        west = self.src.fields(z, 2 ** (z + 1) - 1, y, 1, ("z",))
        east = self.src.fields(z, 0, y, 1, ("z",))
        np.testing.assert_allclose(west["z"][:, -1], east["z"][:, 1], atol=1e-6)

    def test_detail_amplitude_is_continuous(self):
        # blending landform amplitudes across cells must not jump where the set of nearest cells changes
        lon = np.linspace(-60.0, 60.0, 40001)
        amp, ridge, sea = self.src._amp(latlon_to_xyz(np.full_like(lon, 10.0), lon))
        for a in (amp, ridge, sea):
            span = a.max() - a.min()
            if span > 0:
                self.assertLess(np.abs(np.diff(a)).max(), 0.02 * span)

    def test_coast_rule_is_continuous(self):
        sea = np.linspace(0.0, 1.0, 1001)
        for z in (-3000.0, -100.0, 0.0, 100.0, 3000.0):
            w = T.coast_water(np.full_like(sea, z), sea)
            self.assertFalse(w[sea < 0.001].any())
            self.assertTrue(w[sea > 0.999].all())
            self.assertEqual(int(np.sum(np.diff(w.astype(int)) != 0)), 1, z)   # one clean switch, no hard cut at a threshold
            self.assertEqual(bool(T.coast_water(np.array([z]), np.array([0.5]))[0]), z < 0)
        # below sea level, water reaches further into the mixed band than a flat threshold would allow
        self.assertTrue(T.coast_water(np.array([-3000.0]), np.array([0.05]))[0])

    def test_coast_band(self):
        f = self.src.fields(4, *tile_at(10.0, -70.0, 4), 0, ("z", "water"))   # an 11° tile over the small world's west coast
        self.assertTrue(f["water"][f["sea"] > 0.999].all())
        self.assertFalse(f["water"][f["sea"] < 0.001].any())
        self.assertTrue(f["water"].any() and (~f["water"]).any(), "tile should hold both land and sea")

    def test_layers_render(self):
        for layer in T.LAYERS:
            im = Image.open(io.BytesIO(self.src.tile(layer, 5, *tile_at(10.0, 20.0, 5))))
            self.assertEqual((im.format, im.size), ("JPEG", (256, 256)), layer)
        h = Image.open(io.BytesIO(self.src.tile("height", 5, *tile_at(10.0, 20.0, 5))))
        self.assertEqual((h.format, h.size), ("PNG", (256, 256)))
        self.assertIn(h.mode, ("I;16", "I"))
        for y in (0, 2 ** 5 - 1):                                              # polar rows
            Image.open(io.BytesIO(self.src.tile("relief", 5, 3, y))).load()

    def test_deterministic_and_cached(self):
        args = ("relief", 9, *tile_at(12.0, 25.0, 9))
        a = self.src.tile(*args)
        other = T.TileSource(serve.World(self.root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp()))
        self.assertEqual(a, other.render(*args))
        self.assertTrue((self.cache / self.src.tag / "relief" / "9").is_dir())
        t = time.time()
        self.src.tile(*args)
        self.assertLess(time.time() - t, 0.05)

    def test_speed(self):
        t = time.time()
        self.src.render("relief", 14, *tile_at(3.0, 5.0, 14))
        self.assertLess(time.time() - t, 3.0)

    def test_fingerprint_follows_the_build(self):
        import os
        a = T.TileSource(serve.World(self.root, 2), seed=7)
        self.assertEqual(a.fingerprint, T.TileSource(serve.World(self.root, 2), seed=7).fingerprint)
        self.assertNotEqual(a.fingerprint, T.TileSource(serve.World(self.root, 2), seed=8).fingerprint)
        self.assertIn(a.fingerprint, str(a.cache_dir))
        self.assertIn(a.fingerprint, a.meta()["url"])
        elev = self.root / "out" / "r2" / "raster" / "elevation.png"
        st = elev.stat()
        os.utime(elev, ns=(st.st_atime_ns, st.st_mtime_ns + 10**9))      # a rebuild rewrites the rasters
        self.assertNotEqual(a.fingerprint, T.TileSource(serve.World(self.root, 2), seed=7).fingerprint)

    def test_z_at(self):
        z = self.src.z_at(10.0, 20.0)
        self.assertEqual(z.shape, (1,))
        self.assertTrue(np.isfinite(z[0]))


    def test_mesh_heights(self):
        z, (x, y) = 9, tile_at(40.0, 10.0, 9)
        h = self.src.mesh(z, x, y)[0]
        self.assertEqual(h.shape, (T.MESH_N, T.MESH_N))
        self.assertTrue(np.isfinite(h).all())
        np.testing.assert_array_equal(h[:, -1], self.src.mesh(z, x + 1, y)[0][:, 0])   # shared edges: same points, same heights
        np.testing.assert_array_equal(h[-1, :], self.src.mesh(z, x, y + 1)[0][0, :])
        n = 2 ** (z + 1)
        np.testing.assert_allclose(self.src.mesh(z, n - 1, y)[0][:, -1], self.src.mesh(z, 0, y)[0][:, 0], atol=1e-3)

    def test_mesh_true_heights_and_water(self):
        for x in range(64):                                                           # a coastal z5 tile
            h, water = self.src.mesh(5, x, 12)
            if water.any() and (~water).any():
                break
        else:
            self.fail("no coastal z5 tile on row 12")
        span = 180.0 / 32
        f = np.arange(T.MESH_N) / (T.MESH_N - 1)
        LA, LO = np.meshgrid(90.0 - (12 + f) * span, -180.0 + (x + f) * span, indexing="ij")
        z, wet = self.src.surface_at(LA.ravel(), LO.ravel(), span / 128 * np.pi / 180 * self.src.R)
        np.testing.assert_allclose(h, z.reshape(h.shape), atol=1e-6)
        self.assertLess(float(h[water].min()), -100.0, "the sea floor, not the water surface")
        self.assertTrue(np.all(h[water] < 60.0), "water only where the coast rule puts sea")

    def test_mesh_cached_and_fast(self):
        args = (11, *tile_at(-20.0, -35.0, 11))
        t = time.time()
        b = self.src.tile("mesh", *args)
        self.assertLess(time.time() - t, 0.5)
        self.assertEqual(len(b), T.MESH_BYTES)
        h, water = self.src.mesh(*args)
        n2 = T.MESH_N * T.MESH_N
        np.testing.assert_array_equal(np.frombuffer(b[:4 * n2], "<f4"), h.astype("<f4").ravel())
        np.testing.assert_array_equal(np.frombuffer(b[4 * n2:], np.uint8), water.astype(np.uint8).ravel())
        self.assertTrue((self.cache / self.src.tag / "mesh" / "11" / str(args[1]) / f"{args[2]}.bin").is_file())

    def test_a_saved_mesh_of_the_wrong_size_is_rendered_again(self):
        args = (10, *tile_at(-21.0, -36.0, 10))
        f = self.src.cache_path("mesh", *args)
        f.parent.mkdir(parents=True, exist_ok=True)
        f.write_bytes(b"\0" * (T.MESH_N * T.MESH_N * 4))
        self.assertEqual(len(self.src.tile("mesh", *args, remember=False)), T.MESH_BYTES)
        self.assertEqual(f.stat().st_size, T.MESH_BYTES)

    def test_surface_matches_the_tile_pixels(self):
        z, (x, y) = 8, tile_at(10.0, -70.0, 8)                                      # the test world's west coast
        f = self.src.fields(z, x, y, 0, ("z", "cat"))
        zz, water = self.src.surface_at(f["lat"].ravel(), f["lon"].ravel(), f["px_km"])
        tile_water = (f["water"] | f["lake"].astype(bool)).ravel()
        self.assertLess(np.mean(water != tile_water), 0.02)
        self.assertLess(np.mean(np.abs(zz - f["z"].ravel())), 10.0)


    def _river_tile(self, z):
        net = self.src.river_net
        a = self.src.w.arrays
        wet = a["lake"].astype(bool) | a["ocean"].astype(bool)
        ok = a["river"][net.seg_a].astype(bool) & ~wet[net.seg_a] & ~wet[a["recv"][net.seg_a]]   # on over land
        i = int(net.seg_a[np.argmax(np.where(ok, net.half_w, 0))])   # the widest river
        xyz = self.src.w.arrays["g_xyz"][i]
        lat, lon = np.degrees(np.arcsin(xyz[2])), np.degrees(np.arctan2(xyz[1], xyz[0]))
        return (z, *tile_at(lat, lon, z))

    def test_valleys_cut_never_raise_and_hold_water(self):
        args = self._river_tile(9)
        f = self.src.fields(*args, 0, ("z", "cat"))
        raw = self.src.fields(*args, 0, ("z", "cat"), valleys=False)
        rel = self.src.relief_at(f["lat"].ravel(), f["lon"].ravel(), f["px_km"]).reshape(f["z"].shape)
        self.assertTrue(np.all(f["z"] <= np.maximum(raw["z"], rel) + 1.0), "at most a micro-relief dip is kept dry")
        self.assertTrue(f["river"].any(), "the river channel is water in its tile")
        self.assertTrue((raw["z"] - f["z"]).max() > 1.0, "a valley was cut")

    def test_valley_seams_match(self):
        z, x, y = self._river_tile(10)
        a = self.src.fields(z, x, y, 1, ("z",))
        east = self.src.fields(z, x + 1, y, 1, ("z",))
        np.testing.assert_allclose(a["z"][:, -1], east["z"][:, 1], atol=1e-6)

    def test_valley_tiles_stay_fast_near_big_rivers(self):
        for z in (12, 16):
            args = self._river_tile(z)
            t = time.time()
            self.src.render("relief", *args)
            self.assertLess(time.time() - t, 1.5, f"z{z}")
        t = time.time()
        self.src.surface_at(np.linspace(-40, 40, 256), np.full(256, 20.0), 20.0)   # a 17,800 km profile
        self.assertLess(time.time() - t, 0.5)

    def test_valleys_fade_in(self):
        z, x, y = self._river_tile(5)
        f = self.src.fields(z, x, y, 0, ("z",))
        raw = self.src.fields(z, x, y, 0, ("z",), valleys=False)
        self.assertLess(np.abs(f["z"] - raw["z"]).max(), 2000.0)


    def test_relief_draws_the_river(self):
        args = self._river_tile(11)
        f = self.src.fields(*args, 1, ("z", "water", "cat"))
        img = self.src.rgb("relief", *args)
        lake = np.array([79, 143, 192])
        river = f["river_draw"][1:-1, 1:-1] & ~f["water"][1:-1, 1:-1]
        self.assertTrue(river.any())
        shade = img[river].astype(float) / lake
        self.assertTrue(np.all(np.abs(shade - shade[:, :1]) < 0.02), "river pixels are the (shaded) water colour")

    def test_tiles_draw_micro_relief_over_the_refine_surface(self):
        z, x, y = 15, *tile_at(12.0, 25.0, 15)
        f = self.src.fields(z, x, y, 0, ("z",), valleys=False)
        rel = self.src.relief_at(f["lat"].ravel(), f["lon"].ravel(), f["px_km"]).reshape(f["z"].shape)
        d = f["z"] - rel                                     # relief_at (refine's start) stays without it
        self.assertGreater(float(np.std(d)), 0.5)
        self.assertLess(float(np.abs(d).max()), 60.0)

    def test_micro_relief_never_digs_dry_ground_below_the_river(self):
        z, x, y = self._river_tile(14)
        f = self.src.fields(z, x, y, 0, ("z",))
        ch, zz = f["river"], f["z"]
        self.assertTrue(ch.any() and (~ch).any())
        for s in ((0, 1), (0, -1), (1, 0), (-1, 0)):         # dry pixels beside the channel: not below its water
            wet, h = np.roll(ch, s, (0, 1)), np.roll(zz, s, (0, 1))
            dry = wet & ~ch
            dry[[0, -1], :] = dry[:, [0, -1]] = False
            self.assertTrue(np.all(zz[dry] >= h[dry] - 1e-6), s)

    def test_land_colour_varies_within_one_zone(self):
        args = (13, *tile_at(12.0, 25.0, 13))
        f = self.src.fields(*args, 0, ("z", "water", "cat"))
        img = self.src.rgb("relief", *args).astype(float)
        land = ~f["water"] & ~f["lake"].astype(bool) & ~f["river_draw"] & (f["ice"] == 0)
        zone = f["holdridge"] == np.bincount(f["holdridge"][land]).argmax()
        hue = (img[..., 0] - img[..., 1]) / (img[..., 0] + img[..., 1] + 1.0)
        self.assertGreater(float(np.std(hue[land & zone])), 0.004, "not one flat colour under the hillshade")
        self.assertLess(float(np.std(hue[land & zone])), 0.05)


class RegionTilesTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        import refine as RF
        cls.root = built_world()
        from tests.test_refine import clear_regions
        clear_regions(cls.root)
        w = RF.WorldCells(cls.root / "out" / "r2")
        i = big_river(w.a)
        cls.lat, cls.lon = float(w.a["g_lat"][i]), float(w.a["g_lon"][i])
        reg = cls.root / "places" / "regions.json"
        reg.parent.mkdir(exist_ok=True)
        d = 3
        reg.write_text(json.dumps({"regions": [{"id": "r", "name": "t", "outline": [[cls.lat - d, cls.lon - d], [cls.lat - d, cls.lon + d], [cls.lat + d, cls.lon + d], [cls.lat + d, cls.lon - d]]}]}))
        from mapgen import config as C
        seed = int(C.load(cls.root)[0]["build"]["seed"])                  # the build's seed, as the server uses
        cls.plain = T.TileSource(serve.World(cls.root, 2), seed=seed, cache_dir=Path(tempfile.mkdtemp()))
        RF.build_areas(cls.root, 2, reg, log=lambda s: None)
        cls.src = T.TileSource(serve.World(cls.root, 2), seed=seed, cache_dir=Path(tempfile.mkdtemp()))

    @classmethod
    def tearDownClass(cls):
        from tests.test_refine import clear_regions
        clear_regions(cls.root)

    def test_tag_changes_with_regions(self):
        self.assertFalse(self.src.regions.empty)
        self.assertNotEqual(self.src.tag, self.plain.tag)
        self.assertRegex(self.src.tag, r"^v\d+\.\d+-[0-9a-f]+$")

    def test_region_tile_urls_follow_the_region_rendering_version(self):
        from unittest import mock
        with mock.patch.object(T, "REGION_VERSION", T.REGION_VERSION + 1):
            other = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()))
        self.assertNotEqual(other.tag, self.src.tag, "new region rendering: new tile URLs, no stale saved tiles")
        self.assertEqual(other.base_tag, self.src.base_tag, "the world's own tiles stay")
        self.assertRegex(other.tag, r"^v\d+\.\d+-[0-9a-f]+$")

    def test_inside_uses_refined_heights(self):
        z, (x, y) = 8, tile_at(self.lat, self.lon, 8)
        a, b = self.src.fields(z, x, y, 0, ("z",)), self.plain.fields(z, x, y, 0, ("z",))
        self.assertGreater(np.abs(a["z"] - b["z"]).mean(), 1.0)

    def test_region_edge_seams(self):
        for dx in (-3, -2, 2, 3):
            z = 8
            x, y = tile_at(self.lat, self.lon + dx, z)
            a, e = self.src.fields(z, x, y, 1, ("z",)), self.src.fields(z, x + 1, y, 1, ("z",))
            np.testing.assert_allclose(a["z"][:, -1], e["z"][:, 1], atol=1e-6)

    def test_blend_band_has_no_step(self):
        lat = np.full(400, self.lat)
        lon = self.lon + np.linspace(2.0, 4.0, 400)                     # across the east edge (lon + 3)
        z = self.src.z_at(lat, lon, 1.0)
        xyz = latlon_to_xyz(lat, lon).reshape(-1, 3)
        lake = self.src._flat_lakes(np.zeros(len(lat)), lat, lon, *self.src._warp_offsets(xyz, 4.0)) != 0
        shore = lake[1:] != lake[:-1]                                   # (a world lake's shore may be a bank)
        k = int(np.argmax(np.where(shore, 0.0, np.abs(np.diff(z)))))
        fine = np.linspace(lon[k], lon[k + 1], 41)                     # 10× finer around the worst step
        zf = self.src.z_at(np.full(41, self.lat), fine, 1.0)
        self.assertLess(np.abs(np.diff(zf)).max(), 0.3 * abs(z[k + 1] - z[k]) + 1.0,
                        "continuous: the worst step shrinks with the sample spacing (a cliff would not)")

    def test_outside_unchanged_and_no_regions_unchanged(self):
        z, (x, y) = 8, tile_at(self.lat + 20, self.lon, 8)
        np.testing.assert_array_equal(self.src.fields(z, x, y, 0, ("z",))["z"], self.plain.fields(z, x, y, 0, ("z",))["z"])
        self.assertTrue(self.plain.regions.empty)

    def test_inspector_answers_refined(self):
        w = serve.World(self.root, 2)
        w.regions = self.src.regions
        d = w.describe(self.lat, self.lon)
        self.assertTrue(d.get("refined"))
        far = w.describe(self.lat + 20, self.lon)
        self.assertFalse(far.get("refined", False))


    def _exit(self):
        import refine as RF
        (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint)
        a = dict(np.load(d / "cells.npz"))
        halo = a["halo"].astype(bool)
        feed = np.where(a["river"].astype(bool) & ~halo & halo[a["recv"]])[0]
        return a, feed

    def test_rivers_run_on_to_the_halo(self):
        a, feed = self._exit()
        net = self.src.regions.net
        self.assertTrue(len(feed), "the test area has rivers leaving it")
        self.assertTrue(np.all(net.seg_len > 0), "no zero-length segments")
        ends = {tuple(np.round(v, 9)) for v in net.b_xyz}
        for i in feed:
            self.assertIn(tuple(np.round(a["g_xyz"][a["recv"][i]], 9)), ends, "the river reaches its halo cell")

    def test_world_rivers_stay_drawn_across_the_blend_band(self):
        a, feed = self._exit()
        found = False
        for i in feed:
            lat, lon = float(a["g_lat"][a["recv"][i]]), float(a["g_lon"][a["recv"][i]])
            for dz in (7, 8):
                x, y = tile_at(lat, lon, dz)
                f, g = self.src.fields(dz, x, y, 0, ("z",)), self.plain.fields(dz, x, y, 0, ("z",))
                xyz = latlon_to_xyz(f["lat"], f["lon"]).reshape(-1, 3)
                w = self.src.regions.weight(xyz).reshape(f["z"].shape)
                band = g["river_draw"] & (w < 1)
                if band.any():
                    found = True
                    self.assertTrue(np.all(f["river_draw"][band]), "world rivers drawn wherever the world still shows")
        self.assertTrue(found)

    def test_region_valleys_reach_up_to_the_shoulders(self):
        import refine as RF
        import rivers as RV
        (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint)
        tmp = Path(tempfile.mkdtemp())
        dst = RF.results_dir(tmp, "k") / d.name
        shutil.copytree(d, dst)
        a = dict(np.load(dst / "cells.npz"))
        src = ~np.isin(np.arange(len(a["recv"])), a["recv"][a["recv"] != np.arange(len(a["recv"]))])
        dry = src & ~a["halo"].astype(bool) & ~a["ocean"].astype(bool) & ~a["lake"].astype(bool)   # (coarse test cells
        a["river"] = a["river"].astype(bool) & ~dry                                                 # are all rivers)
        a["z_surface_m"] = np.where(dry, a["z_surface_m"] + 500.0, a["z_surface_m"])               # canyons
        np.savez(dst / "cells.npz", **a)
        rs = RF.RegionSet(tmp, "k", self.src.w.legends, self.src.R)
        plain = RV.RiverNet(rs.arrays, self.src.w.legends, rs.R, levels=rs.arrays["z_surface_m"])
        self.assertTrue(np.any(rs.net.reach[: len(plain.reach)] > plain.reach + 1e-9), "walls reach up to the shoulders")

    def test_profile_inside_ignores_world_lakes(self):
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()))
        src.w.arrays = dict(src.w.arrays, lake=np.ones(len(src.w.arrays["g_ids"]), bool))   # lakes everywhere
        lat, lon = np.full(50, self.lat), self.lon + np.linspace(-0.5, 0.5, 50)
        np.testing.assert_array_equal(src.surface_at(lat, lon)[1], self.src.surface_at(lat, lon)[1])

    def test_inspector_band_answers_from_the_world(self):
        rs = self.src.regions
        a, feed = self._exit()
        pa, pb = a["g_xyz"][feed[0]], a["g_xyz"][a["recv"][feed[0]]]
        pts = pa + np.linspace(0.0, 0.6, 121)[:, None] * (pb - pa)             # from the border cell toward the halo
        pts /= np.linalg.norm(pts, axis=1, keepdims=True)
        lat, lon = np.degrees(np.arcsin(pts[:, 2])), np.degrees(np.arctan2(pts[:, 1], pts[:, 0]))
        band = [k for k in range(len(pts)) if rs.weight(pts[k:k + 1])[0] < 0.5 and rs.index_of(lat[k], lon[k]) is not None]
        self.assertTrue(band, "a point of a refined cell in the outer blend band")
        world = serve.World(self.root, 2)
        world.regions = rs
        d = world.describe(float(lat[band[0]]), float(lon[band[0]]))
        self.assertFalse(d["refined"])
        self.assertEqual(d["inner_km"], 0.0, "no reuse of an answer next to a refined area")

    def test_a_server_without_regions_leaves_region_tile_caches_alone(self):
        root = self.root / "out" / "r2" / "tiles"
        plain = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, regions_dir=Path(tempfile.mkdtemp()))   # no regions
        other_world, region = root / "v1-deadbeef00", root / (plain.base_tag + "abc123")
        for d in (other_world, region):
            d.mkdir(parents=True, exist_ok=True)
        plain.warm()
        self.assertFalse(other_world.exists(), "another world's tiles go")
        self.assertTrue(region.exists(), "a smoke server must not wipe the real server's region tiles")
        shutil.rmtree(region)

    def test_worker_processes_render_the_same_tiles(self):
        from concurrent.futures import ThreadPoolExecutor
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()))
        src.start_workers(2)
        try:
            args = [("relief", 8, *tile_at(self.lat, self.lon, 8)), ("mesh", 8, *tile_at(self.lat, self.lon, 8)),
                    ("relief", 6, *tile_at(self.lat + 20, self.lon, 6))]
            with ThreadPoolExecutor(3) as ex:
                got = list(ex.map(lambda a: src.tile(*a), args))
            self.assertEqual(got, [self.src.render(*a) for a in args])
            self.assertEqual(src.pool.alive, 2)
        finally:
            src.pool.close()

    def test_workers_exit_when_the_server_process_dies(self):
        import os
        import signal
        import subprocess
        import sys
        import textwrap
        code = textwrap.dedent(f"""
            import sys, socket, tempfile, pathlib; sys.path.insert(0, {str(Path(T.__file__).parent)!r})
            import tiles as T, serve
            lsock = socket.socket(); lsock.bind(("127.0.0.1", 0)); lsock.listen()     # a server's listening socket
            src = T.TileSource(serve.World(pathlib.Path({str(self.root)!r}), 2), seed=7, cache_dir=pathlib.Path(tempfile.mkdtemp()))
            src.start_workers(2)
            print(lsock.getsockname()[1], *[p.pid for p in src.pool.procs], flush=True)
            import time; time.sleep(60)
        """)
        p = subprocess.Popen([sys.executable, "-c", code], stdout=subprocess.PIPE, text=True)
        port, *pids = map(int, p.stdout.readline().split())
        os.kill(p.pid, signal.SIGKILL)                                   # no atexit: as a killed server
        p.wait()
        deadline = time.time() + 10
        while time.time() < deadline and any(os.path.exists(f"/proc/{q}") and open(f"/proc/{q}/stat").read().split()[2] != "Z"
                                             for q in pids):
            time.sleep(0.2)
        alive = [q for q in pids if os.path.exists(f"/proc/{q}") and open(f"/proc/{q}/stat").read().split()[2] != "Z"]
        for q in alive:
            os.kill(q, signal.SIGKILL)
        self.assertEqual(alive, [], "orphaned render workers linger")
        import socket
        s = socket.socket()
        self.assertNotEqual(s.connect_ex(("127.0.0.1", port)), 0, "nobody still holds the server's listening socket")
        s.close()

    def test_workers_follow_a_region_reload(self):
        import refine as RF
        empty = Path(tempfile.mkdtemp())
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()),
                           regions_dir=empty)
        src.start_workers(1)
        try:
            args = ("relief", 8, *tile_at(self.lat, self.lon, 8))
            self.assertEqual(src.tile(*args), self.plain.render(*args))
            (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint)
            shutil.copytree(d, RF.results_dir(empty, src.fingerprint) / d.name)
            src.reload_regions()
            self.assertEqual(src.tile(*args), self.src.render(*args))
        finally:
            src.pool.close()

    def test_prerender_saves_the_region_tiles(self):
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()))
        seen = []
        n = src.prerender(max_z=6, progress=lambda done, total: seen.append((done, total)))
        for z in (5, 6):
            for layer in ("relief", "mesh"):
                self.assertTrue(src.cache_path(layer, z, *tile_at(self.lat, self.lon, z)).exists(), (layer, z))
        self.assertFalse(src.cache_path("relief", 6, *tile_at(self.lat + 40, self.lon, 6)).exists())
        self.assertGreater(n, 0)
        self.assertEqual(seen[-1], (n, n))
        self.assertEqual(src.prerender(max_z=6), 0, "saved tiles are not rendered again")

    def _empty_src(self):
        return T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()),
                            regions_dir=Path(tempfile.mkdtemp()))

    def _add_area(self, src):
        import refine as RF
        (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint)
        dst = RF.results_dir(src.regions_dir, src.fingerprint) / d.name
        shutil.copytree(d, dst)
        return dst

    def test_a_reload_during_a_render_never_caches_mixed_tiles(self):
        src = self._empty_src()
        args = ("relief", 8, *tile_at(self.lat, self.lon, 8))
        orig, calls = src.render, []

        def render(*a):
            data = orig(*a)                                  # rendered with the old (no) regions …
            if not calls:
                calls.append(1)
                self._add_area(src)
                src.reload_regions()                         # … while a build finished
            return data
        src.render = render
        want = self.src.render(*args)
        self.assertEqual(src.tile(*args), want)
        self.assertEqual(src.tile(*args), want, "nor from the memory cache")
        base = src.cache_root / src.base_tag / "relief" / "8" / str(args[2]) / f"{args[3]}.jpg"
        self.assertFalse(base.exists(), "no pre-build bytes in the world's cache for a tile the region covers")

    def test_workers_never_render_a_half_built_region_set(self):
        src = self._empty_src()
        src.start_workers(1)                                 # forked without regions
        try:
            area = self._add_area(src)
            src.reload_regions()                             # the server has the region …
            shutil.rmtree(area)                              # … a new build has pruned it on disk meanwhile
            args = ("relief", 8, *tile_at(self.lat, self.lon, 8))
            self.assertEqual(src.tile(*args), self.src.render(*args))
        finally:
            src.pool.close()

    def test_dead_workers_never_hang_tile_requests(self):
        import os
        import signal
        src = self._empty_src()
        orig = src.render
        src.render = lambda *a: (time.sleep(1.0), orig(*a))[1]   # slow renders (the forked worker inherits this)
        src.start_workers(1)
        try:
            args = [("relief", 9, *tile_at(self.lat, self.lon, 9)), ("relief", 9, *tile_at(self.lat + 10, self.lon, 9))]
            got = [None, None]

            def run(k):
                got[k] = src.tile(*args[k])
            ts = [threading.Thread(target=run, args=(k,), daemon=True) for k in (0, 1)]
            ts[0].start()
            time.sleep(0.15)                                   # the first request is inside the worker …
            ts[1].start()
            time.sleep(0.15)                                   # … the second waits for it …
            os.kill(src.pool.procs[0].pid, signal.SIGKILL)     # … and the worker dies
            for t in ts:
                t.join(30)
            self.assertEqual(got, [self.plain.render(*a) for a in args], "both requests answered (rendered here)")
        finally:
            src.pool.close()

    def test_a_region_change_keeps_the_saved_tiles_of_unchanged_areas(self):
        import refine as RF
        from tests.test_refine import square
        tmp = Path(tempfile.mkdtemp())
        far_lat = self.lat - 50 if self.lat > 0 else self.lat + 50
        a, b = square(self.lat, self.lon, 3, "a"), square(far_lat, self.lon + 90, 3, "b")
        (tmp / "one.json").write_text(json.dumps({"regions": [a]}))
        (tmp / "two.json").write_text(json.dumps({"regions": [a, b]}))
        res = tmp / "results"
        RF.build_areas(self.root, 2, tmp / "one.json", log=lambda s: None, regions_root=res)
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=tmp / "cache", regions_dir=res)
        first = src.prerender(max_z=6, layers=("relief",))
        near_a = ("relief", 6, *tile_at(self.lat, self.lon, 6))
        old_tag = src.tag
        RF.build_areas(self.root, 2, tmp / "two.json", log=lambda s: None, regions_root=res)
        src.reload_regions()
        self.assertNotEqual(src.tag, old_tag)
        self.assertTrue(src.cache_path(*near_a).exists(), "area a's saved tiles moved to the new region set")
        self.assertFalse((src.cache_root / old_tag).exists())
        again = src.prerender(max_z=6, layers=("relief",))
        self.assertGreater(again, 0, "area b's tiles are new")
        self.assertLess(again, first + again - 1, "a's are not rendered again")
        self.assertLessEqual(again, first * 1.5)

    def _two_areas(self):
        import refine as RF
        from tests.test_refine import square
        tmp = Path(tempfile.mkdtemp())
        far_lat = self.lat - 50 if self.lat > 0 else self.lat + 50
        a, b = square(self.lat, self.lon, 3, "a"), square(far_lat, self.lon + 90, 3, "b")
        (tmp / "one.json").write_text(json.dumps({"regions": [a]}))
        (tmp / "two.json").write_text(json.dumps({"regions": [a, b]}))
        RF.build_areas(self.root, 2, tmp / "one.json", log=lambda s: None, regions_root=tmp / "results")
        return tmp

    def _source(self, tmp):
        return T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=tmp / "cache",
                            regions_dir=tmp / "results")

    def _rekey(self, src):
        """Area a's content changes (as after a new refine model or config) while its outline stays."""
        import refine as RF
        (link,) = RF.built_areas(src.regions_dir, src.fingerprint)
        old = Path(os.path.realpath(link))
        new = old.with_name(old.name + "x")
        shutil.copytree(old, new)
        os.utime(new / "cells.npz", ns=(time.time_ns(), time.time_ns() + 10 ** 9))
        link.unlink()
        link.symlink_to(os.path.relpath(new, link.parent))

    def test_a_region_change_while_stopped_keeps_the_saved_tiles_of_unchanged_areas(self):
        import refine as RF
        tmp = self._two_areas()
        src = self._source(tmp)
        src.carry_stale()                                            # a server start
        src.prerender(max_z=6, layers=("relief",))
        near_a = ("relief", 6, *tile_at(self.lat, self.lon, 6))
        old_tag = src.tag
        RF.build_areas(self.root, 2, tmp / "two.json", log=lambda s: None, regions_root=tmp / "results")
        src = self._source(tmp)                                      # the next start
        self.assertNotEqual(src.tag, old_tag)
        src.carry_stale()
        self.assertTrue(src.cache_path(*near_a).exists(), "area a's saved tiles moved to the new region set")
        self.assertFalse((src.cache_root / old_tag).exists())

    def test_saved_tiles_of_an_area_whose_content_changed_are_dropped(self):
        tmp = self._two_areas()
        src = self._source(tmp)
        src.carry_stale()
        src.prerender(max_z=6, layers=("relief",))
        near_a = ("relief", 6, *tile_at(self.lat, self.lon, 6))
        self._rekey(src)
        src.reload_regions()
        self.assertFalse(src.cache_path(*near_a).exists(), "a live reload drops them")
        src.carry_stale()
        src.prerender(max_z=6, layers=("relief",))
        self._rekey(src)
        src = self._source(tmp)
        src.carry_stale()
        self.assertFalse(src.cache_path(*near_a).exists(), "so does the next start")

    def test_saved_region_tiles_drawn_by_an_older_region_version_are_not_carried_over(self):
        tmp = self._two_areas()
        src = self._source(tmp)
        src.carry_stale()
        src.prerender(max_z=6, layers=("relief",))
        near_a = ("relief", 6, *tile_at(self.lat, self.lon, 6))
        with mock.patch.object(T, "REGION_VERSION", T.REGION_VERSION + 1):   # the next start draws refined areas anew
            src = self._source(tmp)
            src.carry_stale()
            self.assertFalse(src.cache_path(*near_a).exists(), "tiles drawn the old way re-render")

    def test_prerendering_never_delays_the_server_exit(self):
        import subprocess
        import sys
        import textwrap
        code = textwrap.dedent(f"""
            import sys, tempfile, threading, time, pathlib; sys.path.insert(0, {str(Path(T.__file__).parent)!r})
            import tiles as T, serve
            src = T.TileSource(serve.World(pathlib.Path({str(self.root)!r}), 2), seed={self.src.seed},
                               cache_dir=pathlib.Path(tempfile.mkdtemp()))
            threading.Thread(target=src.prerender, kwargs=dict(max_z=12, threads=2), daemon=True).start()
            time.sleep(1.0)
            print("exiting", flush=True)                       # as after Ctrl-C: main returns
        """)
        t = time.time()
        p = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, timeout=120)
        self.assertIn("exiting", p.stdout, p.stderr[-500:])
        self.assertLess(time.time() - t, 20, "the process ends without rendering the whole queue first")

    def test_a_reload_forks_fresh_workers_sharing_the_new_regions(self):
        src = self._empty_src()
        src.start_workers(2)
        try:
            before = [p.pid for p in src.pool.procs]
            self._add_area(src)
            src.reload_regions()
            after = [p.pid for p in src.pool.procs]
            self.assertTrue(set(before).isdisjoint(after), "new processes (they share the loaded regions)")
            self.assertEqual(src.pool.alive, 2)
            args = ("relief", 8, *tile_at(self.lat, self.lon, 8))
            self.assertEqual(src.tile(*args), self.src.render(*args))
            time.sleep(0.5)
            self.assertFalse(any(os.path.exists(f"/proc/{q}") and open(f"/proc/{q}/stat").read().split()[2] != "Z"
                                 for q in before), "the old workers are gone")
        finally:
            src.pool.close()

    def test_workers_forked_by_a_short_lived_thread_keep_working(self):
        src = self._empty_src()
        src.start_workers(2)
        try:
            self._add_area(src)
            t = threading.Thread(target=src.reload_regions)            # as a region build's thread does
            t.start()
            t.join()
            time.sleep(0.5)                                           # the forking thread is gone now
            self.assertEqual(src.pool.alive, 2)
            self.assertTrue(all(p.is_alive() for p in src.pool.procs), "workers outlive the thread that forked them")
            args = ("relief", 8, *tile_at(self.lat, self.lon, 8))
            data = src.pool.render(src.name, *args, src.regions.fingerprint)     # straight to a worker, no fallback
            self.assertEqual(data, self.src.render(*args))
        finally:
            src.pool.close()

    def test_prerender_keeps_to_a_tile_budget_by_zoom(self):
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()))
        full = src.prerender(max_z=6, layers=("relief",))
        src2 = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()))
        seen = []
        n = src2.prerender(max_z=6, layers=("relief",), max_tiles=full - 1, progress=lambda d, t: seen.append(t))
        self.assertLess(n, full, "the deepest zoom that would overrun the budget is left out")
        self.assertLessEqual(seen[-1], full - 1)
        self.assertTrue(src2.cache_path("relief", 5, *tile_at(self.lat, self.lon, 5)).exists(), "coarse zooms first")

    def test_no_water_walls_where_a_lake_sits_on_a_cliff(self):
        import refine as RF
        (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint)
        tmp = Path(tempfile.mkdtemp())
        dst = RF.results_dir(tmp, self.src.fingerprint) / d.name
        shutil.copytree(d, dst)
        a = dict(np.load(dst / "cells.npz"))
        halo, xyz = a["halo"].astype(bool), a["g_xyz"]
        land = ~halo & ~a["ocean"].astype(bool) & ~a["lake"].astype(bool)
        mid = latlon_to_xyz(self.lat, self.lon).reshape(3)                      # well inside the area: by its centre
        c = int(np.where(land)[0][np.argmin(np.linalg.norm(xyz[land] - mid, axis=1))])
        a["river"][c] = False                                   # (the coarse test cells are all rivers)
        nb = np.argsort(np.linalg.norm(xyz - xyz[c], axis=1))[1:7]          # its six neighbours: a cliff below it
        a["z_surface_m"][nb] = a["z_surface_m"][c] - 1500.0
        a["lake"][c], a["lake_id"][c] = True, int(a["lake_id"].max()) + 1
        a["z_filled_m"][c] = a["z_surface_m"][c] + 50.0
        a["depression_depth_m"][c] = 50.0
        np.savez(dst / "cells.npz", **a)
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), regions_dir=tmp)
        p = xyz[c] + np.linspace(0, 1, 400)[:, None] * (xyz[nb[0]] - xyz[c])  # from the lake over the cliff edge
        p /= np.linalg.norm(p, axis=1, keepdims=True)
        lat, lon = np.degrees(np.arcsin(p[:, 2])), np.degrees(np.arctan2(p[:, 1], p[:, 0]))
        z = src.z_at(lat, lon, 1.0)
        step = np.linalg.norm(p[1] - p[0]) * src.R
        self.assertLess(np.abs(np.diff(z)).max(), 50.0 + RF.LAKE_FLOOR_M + 60.0,
                        "the lake's edge stands at most its depth (+ margin, detail) over the drop, not 1.5 km")

    def test_no_dry_pits_inside_a_lake(self):
        import refine as RF
        (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint)
        tmp = Path(tempfile.mkdtemp())
        dst = RF.results_dir(tmp, self.src.fingerprint) / d.name
        shutil.copytree(d, dst)
        a = dict(np.load(dst / "cells.npz"))
        halo, xyz = a["halo"].astype(bool), a["g_xyz"]
        land = ~halo & ~a["ocean"].astype(bool)
        mid = latlon_to_xyz(self.lat, self.lon).reshape(3)                      # well inside the area: by its centre
        c = int(np.where(land)[0][np.argmin(np.linalg.norm(xyz[land] - mid, axis=1))])
        lake = np.argsort(np.linalg.norm(xyz - xyz[c], axis=1))[:7]              # a cell and its ring: one lake
        bed = float(a["z_surface_m"][c])
        a["z_surface_m"][lake] = bed
        a["lake"][lake], a["river"][lake] = True, False
        a["lake_id"][lake] = int(a["lake_id"].max()) + 1
        a["z_filled_m"][lake] = bed + 100.0
        a["depression_depth_m"][lake] = 100.0
        np.savez(dst / "cells.npz", **a)
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), regions_dir=tmp)
        rng = np.random.default_rng(3)
        sp = src.regions.spacing_km / src.R
        p = xyz[c] + rng.normal(size=(400, 3)) * sp * 0.12                     # well inside the lake (its middle
        p /= np.linalg.norm(p, axis=1, keepdims=True)                           # cell and ring: 90 km cells here)
        lat, lon = np.degrees(np.arcsin(p[:, 2])), np.degrees(np.arctan2(p[:, 1], p[:, 0]))
        z, water = src.surface_at(lat, lon, 1.0)
        below = z < bed + 100.0 - 1e-6
        self.assertFalse((below & ~water).any(), f"{(below & ~water).sum()} dry pits below the lake's surface")
        np.testing.assert_allclose(z[water], bed + 100.0, atol=1e-6)            # (detail may rise as islands)

    def test_no_dry_pits_where_a_lake_bed_deepens_between_its_cells(self):
        import refine as RF
        (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint)
        tmp = Path(tempfile.mkdtemp())
        dst = RF.results_dir(tmp, self.src.fingerprint) / d.name
        shutil.copytree(d, dst)
        a = dict(np.load(dst / "cells.npz"))
        halo, xyz = a["halo"].astype(bool), a["g_xyz"]
        land = ~halo & ~a["ocean"].astype(bool)
        mid = latlon_to_xyz(self.lat, self.lon).reshape(3)                      # well inside the area: by its centre
        c = int(np.where(land)[0][np.argmin(np.linalg.norm(xyz[land] - mid, axis=1))])
        lake = np.argsort(np.linalg.norm(xyz - xyz[c], axis=1))[:19]             # a cell and two rings: one lake
        bed = float(a["z_surface_m"][c])
        a["z_surface_m"][lake] = bed
        a["z_surface_m"][c] = bed - 200.0                                         # its middle much deeper
        a["lake"][lake], a["river"][lake] = True, False
        a["lake_id"][lake] = int(a["lake_id"].max()) + 1
        a["z_filled_m"][lake] = bed + 100.0
        a["depression_depth_m"][lake] = 100.0
        a["depression_depth_m"][c] = 300.0
        isle = lake[1]                                                            # and an island beside it
        a["lake"][isle], a["z_surface_m"][isle], a["z_filled_m"][isle] = False, bed + 150.0, bed + 150.0
        np.savez(dst / "cells.npz", **a)
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), regions_dir=tmp)
        p = np.concatenate([xyz[c] + np.linspace(0.2, 0.9, 100)[:, None] * (xyz[n] - xyz[c]) for n in lake[2:7]])
        p /= np.linalg.norm(p, axis=1, keepdims=True)                           # from the deep middle to the ring
        lat, lon = np.degrees(np.arcsin(p[:, 2])), np.degrees(np.arctan2(p[:, 1], p[:, 0]))
        z, water = src.surface_at(lat, lon, 1.0)
        below = z < bed + 100.0 - 1e-6
        self.assertFalse((below & ~water).any(), f"{(below & ~water).sum()} dry pits below the lake's surface")

    def test_ground_beside_a_lake_below_its_surface_is_under_water(self):
        import refine as RF
        (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint)
        tmp = Path(tempfile.mkdtemp())
        dst = RF.results_dir(tmp, self.src.fingerprint) / d.name
        shutil.copytree(d, dst)
        a = dict(np.load(dst / "cells.npz"))
        halo, xyz = a["halo"].astype(bool), a["g_xyz"]
        land = ~halo & ~a["ocean"].astype(bool)
        mid = latlon_to_xyz(self.lat, self.lon).reshape(3)                      # well inside the area: by its centre
        c = int(np.where(land)[0][np.argmin(np.linalg.norm(xyz[land] - mid, axis=1))])
        ring = np.argsort(np.linalg.norm(xyz - xyz[c], axis=1))[:7]
        lake, shore = ring[:6], ring[6]                                          # a lake and a dry cell beside it
        bed = float(a["z_surface_m"][c])
        a["z_surface_m"][ring] = bed
        a["lake"][lake], a["river"][ring] = True, False
        a["lake"][shore] = False
        a["lake_id"][lake] = int(a["lake_id"].max()) + 1
        a["z_filled_m"][lake] = bed + 100.0
        a["depression_depth_m"][lake] = 100.0
        a["z_surface_m"][shore] = a["z_filled_m"][shore] = bed + 60.0            # 40 m below the lake's surface
        np.savez(dst / "cells.npz", **a)
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), regions_dir=tmp)
        p = xyz[c] + np.linspace(0.55, 0.65, 50)[:, None] * (xyz[shore] - xyz[c])   # just past the cells' border
        p /= np.linalg.norm(p, axis=1, keepdims=True)
        lat, lon = np.degrees(np.arcsin(p[:, 2])), np.degrees(np.arctan2(p[:, 1], p[:, 0]))
        z, water = src.surface_at(lat, lon, 1.0)
        below = z < bed + 100.0 - 1e-6
        self.assertFalse((below & ~water).any(), f"{(below & ~water).sum()} dry pits below the lake's surface")

    def test_tiles_away_from_regions_share_the_world_cache(self):
        far, near = ("relief", 8, *tile_at(self.lat + 20, self.lon, 8)), ("relief", 8, *tile_at(self.lat, self.lon, 8))
        self.assertIn(self.plain.tag, self.src.cache_path(*far).parts, "a region change keeps the tiles away from it")
        self.assertIn(self.src.tag, self.src.cache_path(*near).parts)

    def test_a_region_set_lists_the_built_areas_once(self):
        import refine as RF
        real, calls = RF.built_areas, []
        with mock.patch.object(RF, "built_areas", side_effect=lambda *a: calls.append(1) or real(*a)):
            rs = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R)
        self.assertEqual(len(calls), 1, "names and fingerprint from one listing: a build finishing between is not mixed")
        self.assertEqual(rs.fingerprint, RF.regions_fingerprint(self.src.regions_dir, self.src.fingerprint))

    def test_cached_region_set_equals_the_uncached_one(self):
        import refine as RF
        fresh = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R, use_cache=False)
        cached = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R)
        self.assertIsNotNone(cached.cache)
        self.assertTrue((cached.cache / "done").exists())
        self.assertEqual(set(cached.arrays), set(fresh.arrays))
        for k in fresh.arrays:
            np.testing.assert_array_equal(np.asarray(cached.arrays[k]), fresh.arrays[k], err_msg=k)
        for k in ("z_env", "lake_level", "lake_floor"):
            np.testing.assert_array_equal(getattr(cached, k), getattr(fresh, k), err_msg=k)
        self.assertEqual(cached.spacing_km, fresh.spacing_km)
        self.assertEqual(cached.area_names, fresh.area_names)
        import rivers as RV
        for k in RV.STATE:
            np.testing.assert_array_equal(getattr(cached.net, k), getattr(fresh.net, k), err_msg=k)

    def test_low_memory_region_set_spills_to_disk_with_equal_arrays(self):
        import refine as RF
        dirs = RF.built_areas(self.src.regions_dir, self.src.fingerprint)
        want, wmeta = RF._compute_region_set(dirs, self.src.w.legends, self.src.R)    # oracle: the in-RAM build
        with tempfile.TemporaryDirectory() as spill:
            got, gmeta = RF._compute_region_set(dirs, self.src.w.legends, self.src.R, spill=Path(spill))
            self.assertEqual(set(got), set(want))
            self.assertEqual(gmeta, wmeta)
            for k in want:
                np.testing.assert_array_equal(np.asarray(got[k]), want[k], err_msg=k)
            big = [k for k in got if k.startswith("a.")]
            self.assertTrue(big and all(isinstance(got[k], np.memmap) for k in big), "fields kept on disk")
            del got

    def test_low_memory_region_set_leaves_no_spill_folder(self):
        import refine as RF
        with tempfile.TemporaryDirectory() as d, mock.patch.dict(os.environ, {"WORLDGEN_LOW_MEMORY": "1"}):
            regions = Path(d) / "regions"
            shutil.copytree(self.src.regions_dir, regions, symlinks=True)
            for p in regions.glob("*/serve-*"):
                shutil.rmtree(p)
            rs = RF.RegionSet(regions, self.src.fingerprint, self.src.w.legends, self.src.R)
            self.assertTrue((rs.cache / "done").exists())
            self.assertEqual(list(regions.rglob(".spill-*")), [])

    def test_cached_tiles_and_meshes_are_byte_identical(self):
        import refine as RF
        fresh = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()))
        fresh.regions = RF.RegionSet(fresh.regions_dir, fresh.fingerprint, fresh.w.legends, fresh.R, use_cache=False)
        fresh.tag = fresh.base_tag + fresh._region_tag(fresh.regions)
        for z in (6, 8):
            x, y = tile_at(self.lat, self.lon, z)
            self.assertEqual(self.src.render("relief", z, x, y), fresh.render("relief", z, x, y), f"relief z{z}")
            np.testing.assert_array_equal(self.src.mesh(z, x, y), fresh.mesh(z, x, y))

    def test_trees_are_not_built_at_load(self):
        import refine as RF
        rs = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R)
        self.assertIsNone(rs._tree)
        self.assertIsNone(rs._halo_tree)
        self.assertEqual(len(rs.area_trees._built), 0)
        rs.warm()
        self.assertIsNotNone(rs._tree)
        self.assertIsNotNone(rs._halo_tree)
        self.assertIsNotNone(rs.net._tree)

    def test_start_workers_warms_region_trees_before_forking(self):
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()))
        order = []
        with mock.patch.object(src.regions, "warm", side_effect=lambda: order.append("warm")), \
             mock.patch.object(T, "RenderPool", side_effect=lambda s, n: order.append("fork")):
            src.start_workers(2)
        self.assertEqual(order, ["warm", "fork"])

    def test_rebuilt_regions_get_a_new_cache_and_the_old_is_pruned(self):
        import refine as RF
        own = Path(tempfile.mkdtemp()) / "regions"
        shutil.copytree(self.src.regions_dir, own, symlinks=True)   # the class's areas keep their mtimes
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()),
                           regions_dir=own)
        old = src.regions.cache
        area = RF.built_areas(src.regions_dir, src.fingerprint)[0] / "cells.npz"
        os.utime(area)                                   # a rebuilt area: new mtime → new regions fingerprint
        src.reload_regions(cleanup=True)
        self.assertNotEqual(src.regions.cache, old)
        self.assertTrue(src.regions.cache.exists())
        self.assertFalse(old.exists())
        self.assertTrue(self.src.regions.cache.exists(), "the shared fixture's cache is untouched")

    def test_build_areas_keeps_serve_cache_folders(self):
        import refine as RF
        import servecache as SC
        base = RF.results_dir(self.src.regions_dir, self.src.fingerprint)
        marker = SC.cache_dir(base, "somefp")
        marker.mkdir(parents=True, exist_ok=True)
        try:
            RF.build_areas(self.root, 2, self.root / "places" / "regions.json", log=lambda s: None)
            self.assertTrue(marker.exists(), "serve caches are pruned by the server, not by builds")
        finally:
            shutil.rmtree(marker, ignore_errors=True)

    def test_region_cache_follows_the_code_that_computes_it(self):
        import refine as RF
        code = Path(tempfile.mkdtemp()) / "rivers.py"
        code.write_text("A = 1\n")
        with mock.patch.object(RF, "CODE_FILES", (code,)):
            a = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R).cache
            code.write_text("A = 2\n")                    # e.g. a river constant changed: the stored net is stale
            b = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R).cache
        self.assertNotEqual(a, b)
        self.assertTrue((b / "done").exists())

    def test_an_empty_region_set_prunes_the_last_cache(self):
        import refine as RF
        regions = Path(tempfile.mkdtemp()) / "regions"
        shutil.copytree(self.src.regions_dir, regions)     # (the shared fixture stays as it is)
        src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()),
                           regions_dir=regions)
        old = src.regions.cache
        self.assertTrue((old / "done").exists())
        for d in RF.built_areas(regions, src.fingerprint):   # the last region deleted
            shutil.rmtree(d)
        src.reload_regions(cleanup=True)
        self.assertTrue(src.regions.empty)
        self.assertFalse(old.exists())


class ProfileSurfaceTest(unittest.TestCase):
    def test_profile_water_matches_the_tiles(self):
        server = serve.make_server(built_world(), 2, 0, pins_path=Path(tempfile.mkdtemp()) / "p.json")
        with Running(server) as base:
            prof = json.loads(request(base + "/api/profile?path=10,-85;10,-55&n=64")[1])   # across the west coast
        S = prof["samples"]
        step = S[1]["d_km"] - S[0]["d_km"]
        z, water = server.tiles.surface_at([x["lat"] for x in S], [x["lon"] for x in S], step / 2)
        np.testing.assert_allclose([x["z_m"] for x in S], z, atol=0.1)
        self.assertEqual([x["water"] for x in S], [bool(w) for w in water])
        self.assertTrue(any(water) and not all(water))


class TileEndpointTest(unittest.TestCase):
    def test_endpoints(self):
        v = T.VERSION
        with Running(serve.make_server(built_world(), 2, 0, pins_path=Path(tempfile.mkdtemp()) / "p.json")) as base:
            meta = json.loads(request(base + "/api/meta")[1])
            tag = meta["tiles"]["url"].split("/")[2]                          # v1-<build fingerprint>
            s1, b1 = request(base + f"/tiles/{tag}/relief/5/40/10.jpg")
            s2, _ = request(base + f"/tiles/{tag}/height/5/40/10.png")
            s3, b3 = request(base + f"/tiles/{tag}/mesh/5/40/10.bin")
            s4 = request(base + f"/tiles/{tag}/mesh/5/40/10.jpg")[0]
            bad = [request(base + p)[0] for p in (f"/tiles/{tag}/nope/5/1/1.jpg", f"/tiles/{tag}/relief/5/99/1.jpg",
                   f"/tiles/{tag}/relief/17/0/0.jpg", f"/tiles/v{v}-0000000000/relief/5/1/1.jpg", f"/tiles/{tag}/relief/5/1/1.png",
                   f"/tiles/{tag}/height/5/1/1.jpg", f"/tiles/v{v}/relief/5/1/1.jpg")]
            cell = json.loads(request(base + "/api/cell?lat=10&lon=20")[1])
            prof = json.loads(request(base + "/api/profile?path=10,10;10,30&n=32")[1])
        self.assertRegex(meta["tiles"]["url"], rf"^/tiles/v{v}\.{T.DRAW}-[0-9a-f]{{10}}/{{layer}}/{{z}}/{{x}}/{{y}}\.jpg$")
        self.assertEqual((meta["tiles"]["min_z"], meta["tiles"]["max_z"]), (5, 16))
        self.assertEqual((s1, Image.open(io.BytesIO(b1)).size), (200, (256, 256)))
        self.assertEqual(s2, 200)
        self.assertEqual(bad, [404] * 7)
        self.assertRegex(meta["tiles"]["mesh_url"], rf"^/tiles/v{v}\.{T.DRAW}-[0-9a-f]{{10}}/mesh/{{z}}/{{x}}/{{y}}\.bin$")
        self.assertEqual(T.MESH_N, 129, "3D heights: 129 × 129 per tile (author 2026-09-25)")
        self.assertEqual((s3, len(b3), s4), (200, T.MESH_BYTES, 404))
        self.assertEqual((T.VERSION, T.DRAW), (6, 5))
        f = {x["key"]: x for x in cell["fields"]}
        self.assertEqual(f["z_detail_m"]["label"], "Local elevation (procedural)")
        self.assertIn("spans", f["z_detail_m"]["alt"])
        self.assertTrue(all("z_cell_m" in s for s in prof["samples"]))


class SharedPoolTest(unittest.TestCase):
    def test_two_sources_share_one_pool(self):
        a_root, b_root = built_world(), built_world()
        a = T.TileSource(serve.World(a_root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp()), name="a")
        b = T.TileSource(serve.World(b_root, 2), seed=8, cache_dir=Path(tempfile.mkdtemp()), name="b")
        T.join([a, b])
        self.assertIs(a.peers, b.peers)
        a.start_workers(1)
        try:
            self.assertIs(a.pool, b.pool)
            args = (6, *tile_at(10.0, 20.0, 6))
            self.assertEqual(b.pool.render("b", "relief", *args, b.regions.fingerprint), b.render("relief", *args))
            self.assertEqual(a.pool.render("a", "relief", *args, a.regions.fingerprint), a.render("relief", *args))
            self.assertNotEqual(a.render("relief", *args), b.render("relief", *args), "seeds differ: two worlds")
            old = a.pool
            b.reload_regions()
            self.assertIsNot(a.pool, old, "a reload forks fresh workers for every peer")
            self.assertIs(a.pool, b.pool)
            made = []
            real = T.RenderPool
            with mock.patch.object(T, "RenderPool", side_effect=lambda *x: made.append(1) or real(*x)):
                old = a.pool
                T.reload_all([a, b])
            self.assertEqual(len(made), 1, "every era reloaded, then one fork")
            self.assertIsNot(a.pool, old)
            self.assertIs(a.pool, b.pool)
        finally:
            a.pool.close()


class _Hangs:
    name = "h"

    class regions:
        fingerprint = "fp"

    def render(self, layer, z, x, y):
        time.sleep(3600)


class PoolTimeoutTest(unittest.TestCase):
    def test_a_hung_worker_is_stopped_and_the_request_fails(self):
        pool = T.RenderPool({"h": _Hangs()}, 1)
        try:
            with mock.patch.object(T, "RENDER_TIMEOUT_S", 0.5):
                t = time.time()
                with self.assertRaisesRegex(RuntimeError, "timed out"):
                    pool.render("h", "relief", 5, 1, 1, "fp")
            self.assertLess(time.time() - t, 5)
            self.assertEqual(pool.alive, 0)
            pool.procs[0].join(timeout=5)
            self.assertFalse(pool.procs[0].is_alive(), "the stuck process is gone")
        finally:
            pool.close()


class EraShareTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        import refine as RF
        from mapgen import config as C
        from mapgen import eras as ER
        from tests.test_refine import era_world, square, write_regions
        cls.root, cls.far, cls.cut, _ = era_world()
        tmp = Path(tempfile.mkdtemp())
        reg = write_regions(tmp / "regions.json", square(*cls.far, 2, "far"), square(*cls.cut, 2, "cut"))
        cls.rdir = tmp / "res"
        RF.build_areas(cls.root, 2, reg, log=lambda s: None, regions_root=cls.rdir)
        seed = int(C.load(cls.root)[0]["build"]["seed"])            # the areas were built for the world's own seed
        cls.base = T.TileSource(serve.World(cls.root, 2), seed=seed, cache_dir=tmp / "tb", regions_dir=cls.rdir)
        cls.era = T.TileSource(serve.World(cls.root, 2, out=ER.era_dir(cls.root, 2, "after")), seed=seed,
                               cache_dir=tmp / "te", regions_dir=cls.rdir, name="after", share=cls.base)
        T.join([cls.base, cls.era])

    def test_area_keys_name_the_content(self):
        rs_b, rs_e = self.base.regions, self.era.regions
        self.assertEqual(sorted(rs_b.area_keys), sorted(rs_b.area_names))
        same = [h for h in rs_e.area_names if rs_e.area_keys[h] == rs_b.area_keys.get(h)]
        self.assertEqual(len(same), 1, "the far area is shared, the cut one is the era's own")

    def test_tiles_away_from_the_era_are_the_bases(self):
        args = (7, *tile_at(*self.far, 7))
        self.assertTrue(self.era.shared(*args))
        self.assertEqual(self.era.tile("relief", *args), self.base.tile("relief", *args))
        self.assertEqual(self.era.cache_path("relief", *args), self.base.cache_path("relief", *args))
        self.assertFalse(any((self.era.cache_root).rglob("*.jpg")), "nothing saved twice")

    def test_a_shared_tile_keeps_the_render_limit(self):
        import threading
        args = (8, *tile_at(*self.far, 8))
        self.assertTrue(self.era.shared(*args))
        gate = threading.BoundedSemaphore(1)
        gate.acquire()                                              # no slot free
        with self.assertRaises(T.Busy):
            self.era.tile("seafloor", *args, remember=False, gate=gate)

    def test_tiles_near_the_mask_are_the_eras_own(self):
        args = (7, *tile_at(*self.cut, 7))
        self.assertFalse(self.era.shared(*args))
        self.assertNotEqual(self.era.tile("height", *args), self.base.tile("height", *args))

    def test_the_mask_edge_alone_decides_without_refined_areas(self):
        from mapgen import eras as ER
        from mapgen.sphere import gc_dist_km
        tmp = Path(tempfile.mkdtemp())
        era = T.TileSource(serve.World(self.root, 2, out=ER.era_dir(self.root, 2, "after")), seed=self.base.seed,
                           cache_dir=tmp / "te", regions_dir=tmp / "none", name="after", share=self.base)
        with np.load(ER.era_dir(self.root, 2, "after") / "cells.npz") as a:
            xyz, m = np.asarray(a["g_xyz"], np.float64), a["era_mask"].astype(bool)
        z, span = 9, 180.0 / 2 ** 9
        near = far = None
        for k in range(1, 4000):
            lon = self.cut[1] + k * span
            x, y = tile_at(self.cut[0], lon, z)
            c = latlon_to_xyz(90 - (y + 0.5) * span, -180 + (x + 0.5) * span)
            d = gc_dist_km(xyz, c, era.R)
            to_mask = float(d[m].min())
            if near is None and not m[int(np.argmin(d))] and to_mask < 2 * era._spacing_km:
                near = (x, y)
            if to_mask > 2 * era._spacing_km + T.CARRY_KM + 2 * span * np.pi / 180 * era.R:
                far = (x, y)
                break
        self.assertIsNotNone(near)
        self.assertIsNotNone(far)
        self.assertFalse(era.shared(z, *near), "just outside the mask: the era's own")
        self.assertTrue(era.shared(z, *far), "beyond the mask's reach: the base's")

    def test_the_base_shares_nothing(self):
        self.assertFalse(self.base.shared(7, *tile_at(*self.far, 7)))


class SeaFloorTileTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.root = built_world()
        cls.src = T.TileSource(serve.World(cls.root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp()))

    def test_depth_shading_darkens_with_depth(self):
        z = np.tile(np.linspace(-6000.0, -10.0, 64), (64, 1))
        rgb = T.seafloor_rgb(z, np.full(z.shape, 0.7), np.full(z.shape, 3), np.zeros(z.shape, bool)).astype(float)
        self.assertLess(rgb[:, :8].mean(), rgb[:, -8:].mean(), "deep is darker")

    def test_type_tints_the_sea_and_land_keeps_relief(self):
        z = np.full((8, 8), -3000.0)
        land = np.zeros((8, 8), bool)
        land[:, :4] = True
        a = T.seafloor_rgb(np.where(land, 500.0, z), np.full(z.shape, 0.7), np.full(z.shape, 3), land)
        b = T.seafloor_rgb(np.where(land, 500.0, z), np.full(z.shape, 0.7), np.full(z.shape, 7), land)
        self.assertFalse(np.array_equal(a[:, 4:], b[:, 4:]), "clay vs vent field")
        np.testing.assert_array_equal(a[:, :4], b[:, :4])

    def test_seafloor_tile(self):
        self.assertIn("seafloor", T.LAYERS)
        args = (5, *tile_at(-30.0, -150.0, 5))
        im = Image.open(io.BytesIO(self.src.tile("seafloor", *args)))
        self.assertEqual(im.size, (256, 256))
        f = self.src.fields(*args, 0, ("cat",))
        self.assertIn("seabed_type", f)

    def test_world_holds_the_new_fields(self):
        for k in ("seabed_type", "bottom_temp_c", "pressure_bar", "fire_reactivity", "o2_fraction"):
            self.assertIn(k, self.src.w.arrays, k)

    def test_new_data_layers_render(self):
        args = (6, *tile_at(10.0, 20.0, 6))
        for layer in ("seabed", "minerals", "bottom_temp", "sediment", "vent_potential", "pressure", "fire"):
            self.assertIn(layer, T.LAYERS, layer)
            self.assertEqual(Image.open(io.BytesIO(self.src.tile(layer, *args))).size, (256, 256), layer)

    def test_ocean_layers(self):
        for layer in ("currents", "sst", "productivity"):
            self.assertIn(layer, T.LAYERS, layer)
        self.assertEqual(T.CONT["currents"], "current_speed")


class LandColourTest(unittest.TestCase):
    """Land colour = terrain shading + a vegetation mosaic (woods vs open ground; author 2026-10-03, spike E)."""
    @classmethod
    def setUpClass(cls):
        cls.root = built_world()
        cls.src = T.TileSource(serve.World(cls.root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp()))
        a = cls.src.w.arrays
        i = int(np.flatnonzero(~np.asarray(a["ocean"]).astype(bool))[0])
        q = np.asarray(a["g_xyz"][i], dtype=float)
        cls.land = float(np.degrees(np.arcsin(q[2] / np.linalg.norm(q)))), float(np.degrees(np.arctan2(q[1], q[0])))
        names = list(cls.src.w.legends["holdridge"])
        cls.zone = {n: names.index(n) for n in names}

    def synthetic(self, z, x, y, zone, height=None, water=None):
        B = T.LAND_BORDER
        f = dict(self.src.fields(z, x, y, B, ("z",), valleys=False))
        n = f["lat"].shape[0]
        f["z"] = f["base"] + (0.0 if height is None else height)        # height: above the smooth base surface
        f["water"] = np.zeros((n, n), bool) if water is None else water
        f["lake"] = np.zeros((n, n), np.int64)
        f["river_draw"] = np.zeros((n, n), bool)
        f["holdridge"] = np.full((n, n), self.zone[zone])
        f["ground"] = np.zeros((n, n), np.int64)
        return f

    def wood_share(self, zone, z=9):
        x0, y0 = tile_at(*self.land, z)
        return float(np.mean([self.src._veg(self.synthetic(z, x0 + i, y0 + j, zone), T.LAND_BORDER)["wood"].mean()
                              for i in range(3) for j in range(3)]))

    def test_wide_border_keeps_the_interior(self):   # a wider border must not change what the tile itself holds
        z, x, y = 10, *tile_at(*self.land, 10)
        a = self.src.fields(z, x, y, 1, ("z", "water", "cat"))
        b = self.src.fields(z, x, y, 9, ("z", "water", "cat"))
        for k in ("z", "water", "holdridge", "river_draw", "lake"):
            np.testing.assert_array_equal(a[k], b[k][8:-8, 8:-8], k)

    def test_border_is_the_neighbours_own_pixels(self):   # a wide border holds what the next tile draws there
        B = T.LAND_BORDER
        z, x, y = 10, *tile_at(*self.land, 10)
        f = self.src.fields(z, x, y, B, ("z", "water", "cat"))
        west = self.src.fields(z, x - 1, y, 0, ("z", "water", "cat"))
        np.testing.assert_allclose(f["z"][B:-B, :B], west["z"][:, -B:], atol=1e-6)
        np.testing.assert_array_equal(f["holdridge"][B:-B, :B], west["holdridge"][:, -B:])

    def test_colour_does_not_depend_on_the_border(self):   # seamless: nothing reaches past LAND_BORDER
        B = T.LAND_BORDER
        z, x, y = 10, *tile_at(*self.land, 10)
        a = self.src._veg(self.src.fields(z, x, y, B, ("z", "water", "cat")), B)
        b = self.src._veg(self.src.fields(z, x, y, 2 * B, ("z", "water", "cat")), 2 * B)
        for k in ("wood", "bright", "tint"):
            np.testing.assert_allclose(a[k], b[k], atol=1e-9, err_msg=k)

    def test_cover_by_zone(self):
        c = T.veg_cover(list(self.src.w.legends["holdridge"]))
        z = self.zone
        self.assertEqual(c[z["polar desert"]], 0.0)
        self.assertLess(c[z["tropical desert"]], c[z["tropical thorn woodland"]])
        self.assertLess(c[z["tropical thorn woodland"]], c[z["tropical dry forest"]])
        self.assertLess(c[z["tropical dry forest"]], c[z["tropical rain forest"]])
        self.assertLess(c[z["cool temperate steppe"]], c[z["cool temperate moist forest"]])
        self.assertLess(c[z["subpolar dry tundra"]], c[z["subpolar wet tundra"]])

    def test_wetter_zones_are_more_wooded(self):
        rain, dry, desert = (self.wood_share(n) for n in ("tropical rain forest", "tropical dry forest", "tropical desert"))
        self.assertGreater(rain, 0.6)
        self.assertLess(desert, 0.12)
        self.assertLess(desert, dry)
        self.assertLess(dry, rain)

    def test_far_out_the_mosaic_is_the_zone_mean(self):   # patches of a few px read as noise: none at z4
        f = self.synthetic(4, *tile_at(*self.land, 4), "tropical dry forest")
        c = T.veg_cover(list(self.src.w.legends["holdridge"]))[self.zone["tropical dry forest"]]
        np.testing.assert_allclose(self.src._veg(f, T.LAND_BORDER)["wood"], c, atol=1e-12)

    def test_salt_flats_bare(self):
        f = self.synthetic(9, *tile_at(*self.land, 9), "tropical rain forest")
        f["ground"][:] = list(self.src.w.legends["ground"]).index("salt flat")
        self.assertEqual(float(self.src._veg(f, T.LAND_BORDER)["wood"].max()), 0.0)

    def test_woods_in_hollows_not_on_crests(self):   # the same pixels, flat vs sine ridges (the patches cancel)
        z, (x, y) = 11, tile_at(*self.land, 11)
        n, B = T.TILE + 2 * T.LAND_BORDER, T.LAND_BORDER
        s = np.sin(2 * np.pi * np.arange(n) / 48.0)
        zone = "warm temperate dry forest"                   # cover ½: room to gain and to lose
        d = np.hstack([self.src._veg(self.synthetic(z, x + i, y, zone, height=np.tile(120.0 * s, (n, 1))), B)["wood"]
                       - self.src._veg(self.synthetic(z, x + i, y, zone), B)["wood"] for i in range(3)])
        s = np.tile(s[B:-B], 3)
        self.assertLess(d[:, s > 0.97].mean(), -0.03, "fewer woods on crests")   # flat tops and bottoms: no slope term
        self.assertGreater(d[:, s < -0.97].mean(), 0.03, "more in hollows")

    def test_woods_along_water(self):
        z, x, y = 12, *tile_at(*self.land, 12)
        n, B = T.TILE + 2 * T.LAND_BORDER, T.LAND_BORDER
        wet = np.zeros((n, n), bool)
        wet[:, n // 2] = True
        zone = "warm temperate dry forest"
        d = self.src._veg(self.synthetic(z, x, y, zone, water=wet), B)["wood"] - \
            self.src._veg(self.synthetic(z, x, y, zone), B)["wood"]
        mid = T.TILE // 2
        self.assertGreater(d[:, np.r_[mid - 3:mid, mid + 1:mid + 4]].mean(), 0.15, "woods along the water")
        self.assertAlmostEqual(float(np.abs(d[:, :mid - 20]).max()), 0.0, 9, msg="none far away")

    def test_relief_tile_renders_and_repeats(self):
        z, x, y = 9, *tile_at(*self.land, 9)
        a, b = self.src.rgb("relief", z, x, y), self.src.rgb("relief", z, x, y)
        self.assertEqual(a.shape, (T.TILE, T.TILE, 3))
        np.testing.assert_array_equal(a, b)


class GullyTest(unittest.TestCase):
    """Relief detail with erosion gullies on slopes (author 2026-10-03: spike C; full rebuild accepted)."""
    @classmethod
    def setUpClass(cls):
        cls.root = built_world()
        cls.src = T.TileSource(serve.World(cls.root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp()))

    def line(self, lat0, lon0, bearing, km, n=400):   # points along a straight east (90°) or north (0°) line
        d = np.linspace(0.0, km, n) / self.src.R
        if bearing == 90:
            return np.full(n, lat0), lon0 + np.degrees(d) / np.cos(np.radians(lat0))
        return lat0 + np.degrees(d), np.full(n, lon0)

    def detail(self, lat, lon, px_km, slope):   # on a plane rising east at `slope` m/m, landform amplitude 450 m
        xyz = latlon_to_xyz(lat, lon)
        base = lambda la, lo: np.radians(lo) * self.src.R * 1000.0 * np.cos(np.radians(10.0)) * slope
        n = len(lat)
        return self.src._detail(xyz, px_km, np.full(n, 450.0), np.full(n, 0.5), lat, lon, base=base)

    def test_gullies_run_downhill(self):   # the first (4 km) octave alone (px 0.6 km): stripes along the slope
        px = 0.6                           # (finer octaves follow the gully walls: that is the branching)
        down = self.detail(*self.line(10.0, 20.0, 90, 40.0), px, 0.4)
        across = self.detail(*self.line(10.0, 20.0, 0, 40.0), px, 0.4)
        self.assertGreater(np.std(np.diff(across)), 2.0 * np.std(np.diff(down)))

    def test_flat_ground_keeps_its_detail(self):   # no slope: (almost) the plain fractal detail
        lat, lon = self.line(10.0, 20.0, 0, 3.0)
        xyz = latlon_to_xyz(lat, lon)
        n = len(lat)
        flat = self.src._detail(xyz, 0.03, np.full(n, 25.0), np.full(n, 0.0), lat, lon, base=lambda la, lo: 0.0 * la)
        plain = 25.0 * self.src._octaves(xyz, 0.03, T.TOP_KM, T.MIN_KM, T.GAIN, self.src.seed + 7001, np.zeros(n))
        self.assertLess(np.abs(flat - plain).max(), 0.1 * 25.0)   # ≤ 10 % of the landform amplitude

    def test_far_out_no_gullies(self):   # gullies need ≥ 4 px per wavelength: none at z6
        lat, lon = self.line(10.0, 20.0, 0, 300.0)
        xyz, n = latlon_to_xyz(lat, lon), len(lat)
        px = 2.0
        d = self.detail(lat, lon, px, 0.4)
        plain = 450.0 * self.src._octaves(xyz, px, T.TOP_KM, T.MIN_KM, T.GAIN, self.src.seed + 7001, np.full(n, 0.5))
        np.testing.assert_array_equal(d, plain)

    def test_heights_agree_everywhere(self):   # tiles, profile/3D points and the refinement start share one detail
        z, x, y = 11, *tile_at(10.0, 20.0, 11)
        f = self.src.fields(z, x, y, 0, ("z",), valleys=False)
        lat, lon = f["lat"][::37, ::41].ravel(), f["lon"][::37, ::41].ravel()
        ra = self.src.relief_at(lat, lon, f["px_km"])
        xyz = latlon_to_xyz(lat, lon)
        amp, ridge, _ = self.src._amp(xyz)
        np.testing.assert_allclose(ra - self.src.bilinear(self.src.raster("elevation"), lat, lon),
                                   self.src._detail(xyz, f["px_km"], amp, ridge, lat, lon), atol=1e-9)

    def test_compiled_stripes_match_numpy(self):
        if T._phacelle_jit is None:
            self.skipTest("numba not installed")
        rng = np.random.default_rng(3)
        p = rng.normal(size=(5000, 3))
        p /= np.linalg.norm(p, axis=1, keepdims=True)
        e, n = T.east_north(p)
        ang = rng.uniform(0, 2 * np.pi, 5000)[:, None]
        a = np.cos(ang) * e + np.sin(ang) * n
        jit = self.src._phacelle(p, a, 0.5, 123)
        with mock.patch.object(T, "_phacelle_jit", None):
            ref = self.src._phacelle(p, a, 0.5, 123)
        np.testing.assert_allclose(jit[0], ref[0], atol=1e-9)
        np.testing.assert_allclose(jit[1], ref[1], atol=1e-9)