diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/__init__.py | 15 | ||||
| -rw-r--r-- | tests/helpers.py | 39 | ||||
| -rw-r--r-- | tests/test_dedupe.py | 95 | ||||
| -rw-r--r-- | tests/test_deploy.py | 102 | ||||
| -rw-r--r-- | tests/test_export.py | 249 | ||||
| -rw-r--r-- | tests/test_refine.py | 805 | ||||
| -rw-r--r-- | tests/test_rivers.py | 266 | ||||
| -rw-r--r-- | tests/test_seafloor.py | 195 | ||||
| -rw-r--r-- | tests/test_serve.py | 1346 | ||||
| -rw-r--r-- | tests/test_servecache.py | 182 | ||||
| -rw-r--r-- | tests/test_tilecap.py | 81 | ||||
| -rw-r--r-- | tests/test_tiles.py | 1383 | ||||
| -rw-r--r-- | tests/test_viewer_fixtures.py | 23 |
13 files changed, 4781 insertions, 0 deletions
diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..c17e6de --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,15 @@ +import atexit +import os +import shutil +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +import worldgen_path # noqa: E402,F401 (mapgen on sys.path) + +if "WORLDMAP_TEST_TMP" not in os.environ: + _root = tempfile.mkdtemp(prefix="worldmap-viewer-tests-") + os.environ["WORLDMAP_TEST_TMP"] = os.environ["TMPDIR"] = tempfile.tempdir = _root + _pid = os.getpid() + atexit.register(lambda: os.getpid() == _pid and shutil.rmtree(_root, True)) diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 0000000..1bb865b --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,39 @@ +import copy +import functools +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] + +BASE_CFG = { + "planet": {"radius_km": 12742.0, "gravity_g": 1.05, "day_hours": 31.149, "year_days": 216, + "tilt_deg": 20.0, "sea_level_pressure_bar": 1.0, "scale_height_km": 8.0, "o2_fraction": 0.21}, + "build": {"seed": 7, "res_dev": 2, "res_final": 2, "raster_width": 256, "preview_width": 128, + "land_fraction": 0.22}, +} + + +@functools.lru_cache(maxsize=4) +def small_grid(res: int = 2): + from mapgen.grid import build_grid + return build_grid(res, 12742.0) + + +def make_ctx(res=2, data=None, cfg=None, tect=None, root=None): + from mapgen.pipeline import Ctx + c = copy.deepcopy(BASE_CFG) + for sec, vals in (cfg or {}).items(): + c.setdefault(sec, {}).update(vals) + ctx = Ctx(root or ROOT, c, tect or {"plate": []}, res) + ctx.grid = small_grid(res) + ctx.data = dict(data or {}) + return ctx + + +def big_river(a) -> int: + """The cell of the biggest river that flows on over land (not into a lake at its own level, which cuts no + valley, nor off a coast, like a basin lake's spillway): the fixtures' river.""" + import numpy as np + river, lake = np.asarray(a["river"]).astype(bool), np.asarray(a["lake"]).astype(bool) + recv = np.asarray(a["recv"]) + wet = lake | np.asarray(a["ocean"]).astype(bool) + return int(np.argmax(np.asarray(a["discharge_km3_yr"]) * (river & ~wet & ~wet[recv]))) diff --git a/tests/test_dedupe.py b/tests/test_dedupe.py new file mode 100644 index 0000000..2521f95 --- /dev/null +++ b/tests/test_dedupe.py @@ -0,0 +1,95 @@ +import os +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import numpy as np + +import dedupe +import servecache as SC + +B = dedupe.BLOCK + + +class SameRunsTest(unittest.TestCase): + def test_runs_of_identical_whole_blocks(self): + rng = np.random.default_rng(0) + x = rng.integers(0, 256, 10 * B + 100, dtype=np.uint8) + y = x.copy() + y[2 * B + 5] ^= 1 # block 2 differs + y[3 * B] ^= 1 # block 3 differs + y[7 * B + B - 1] ^= 1 # block 7 differs (its last byte) + with tempfile.TemporaryDirectory() as t: + a, b = Path(t) / "a", Path(t) / "b" + a.write_bytes(x.tobytes()) + b.write_bytes(y.tobytes()) + fa, fb = os.open(a, os.O_RDONLY), os.open(b, os.O_RDONLY) + try: + runs = dedupe.same_runs(fa, fb, len(x)) + finally: + os.close(fa) + os.close(fb) + self.assertEqual(runs, [(0, 2 * B), (4 * B, 3 * B), (8 * B, 2 * B)]) # the 100-byte tail is never shared + + def test_dedupe_asks_for_each_identical_run_in_bounded_calls(self): + x = np.zeros(40 * B, np.uint8) + y = x.copy() + y[5 * B] = 1 + calls = [] + with tempfile.TemporaryDirectory() as t, \ + mock.patch.object(dedupe, "_ioctl", lambda a, b, o, n: calls.append((o, n)) or n), \ + mock.patch.object(dedupe, "MAX_CALL", 8 * B): + (Path(t) / "a").write_bytes(x.tobytes()) + (Path(t) / "b").write_bytes(y.tobytes()) + shared = dedupe.dedupe_file(Path(t) / "a", Path(t) / "b") + self.assertEqual(calls, [(0, 5 * B), (6 * B, 8 * B), (14 * B, 8 * B), (22 * B, 8 * B), (30 * B, 8 * B), + (38 * B, 2 * B)]) + self.assertEqual(shared, 39 * B) + + def test_unsupported_file_system_changes_nothing(self): + with tempfile.TemporaryDirectory() as t: + data = os.urandom(3 * B) + for n in "ab": + (Path(t) / n).write_bytes(data) + with mock.patch.object(dedupe, "_ioctl", side_effect=OSError(95, "Operation not supported")): + self.assertEqual(dedupe.dedupe_file(Path(t) / "a", Path(t) / "b"), 0) + self.assertEqual((Path(t) / "b").read_bytes(), data) + + def test_different_sizes_are_skipped(self): + with tempfile.TemporaryDirectory() as t: + (Path(t) / "a").write_bytes(bytes(2 * B)) + (Path(t) / "b").write_bytes(bytes(3 * B)) + with mock.patch.object(dedupe, "_ioctl", side_effect=AssertionError): + self.assertEqual(dedupe.dedupe_file(Path(t) / "a", Path(t) / "b"), 0) + + +class PeersTest(unittest.TestCase): + def test_a_peer_vanishing_midway_is_skipped(self): + with tempfile.TemporaryDirectory() as t: + new = Path(t) / "new" + new.mkdir() + (new / "x.npy").write_bytes(bytes(B)) + with mock.patch.object(dedupe, "dedupe_file", side_effect=FileNotFoundError): + self.assertEqual(dedupe.dedupe_dirs(new, [Path(t) / "gone"]), 0) + with mock.patch.object(Path, "is_file", return_value=True), \ + mock.patch.object(dedupe, "dedupe_file", side_effect=FileNotFoundError): + self.assertEqual(dedupe.dedupe_dirs(new, [Path(t) / "gone"]), 0) + + + def test_a_new_era_cache_is_shared_with_the_other_eras_complete_caches(self): + with tempfile.TemporaryDirectory() as t: + regions = Path(t) / "regions" + arrays = {"x": np.arange(10.0)} + old = SC.cache_dir(regions / "era-a", "k1") + SC.write(old, arrays, {}) + (regions / "era-c" / f"{SC.PREFIX}half").mkdir(parents=True) # incomplete: not a peer + new = SC.cache_dir(regions / "era-b", "k2") + with mock.patch.object(SC.dedupe, "dedupe_dirs") as dd: + SC.write(new, arrays, {}) + dd.assert_called_once_with(new, [old]) + self.assertTrue(np.array_equal(SC.read(new)[0]["x"], np.arange(10.0))) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_deploy.py b/tests/test_deploy.py new file mode 100644 index 0000000..679e621 --- /dev/null +++ b/tests/test_deploy.py @@ -0,0 +1,102 @@ +import contextlib +import io +import shutil +import tempfile +import unittest +from pathlib import Path + +from mapgen import config as C +from mapgen.testing import built_world + +import deploy +import refine +import servecache +import tiles + + +class BundleTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.tmp = Path(tempfile.mkdtemp()) + cls.root = cls.tmp / "world" + shutil.copytree(built_world(), cls.root) # copy2: keeps mtimes (ns) + out = cls.root / "out" / "r2" + seed = int(C.load(cls.root)[0]["build"]["seed"]) + cls.fp = tiles.fingerprint(out, seed) + tag = f"v{tiles.VERSION}.{tiles.DRAW}-{cls.fp}" + for rel, text in { + f"out/r2/regions/{cls.fp}-m{refine.MODEL}/a1/cells.npz": "current area", + "out/r2/regions/0000000000-m7/a1/cells.npz": "an earlier build's area", + "out/r2/regions/areas/k1/cells.npz": "shared area results", + f"out/r2/tiles/{tag}/relief/5/0/0.jpg": "current tile", + f"out/r2/tiles/{tag}+abc/relief/5/0/0.jpg": "current tile, refined set", + "out/r2/tiles/v1.1-0000000000/relief/5/0/0.jpg": "old tile", + f"out/r2/tiles/{tag}/relief/5/0/1.jpg.part": "half written", + "places/pins.json": "{\"pins\": []}", + "places/regions.json": "{\"regions\": []}", + "places/views.md": "notes", + "exports/x.zip": "export", + "lore/true-gods.md": "secret", + "sketch/a.png": "input", + }.items(): + f = cls.root / rel + f.parent.mkdir(parents=True, exist_ok=True) + f.write_text(text) + cls.dest = cls.tmp / "dest" + with contextlib.redirect_stdout(io.StringIO()): + cls.rc = deploy.bundle(cls.root, str(cls.dest), 2) + + @classmethod + def tearDownClass(cls): + shutil.rmtree(cls.tmp) + + def test_copies_what_the_public_server_reads(self): + self.assertEqual(self.rc, 0) + tag = f"v{tiles.VERSION}.{tiles.DRAW}-{self.fp}" + for rel in ("config/world.toml", "places/regions.json", "out/r2/cells.npz", "out/r2/fields.json", + f"out/r2/regions/{self.fp}-m{refine.MODEL}/a1/cells.npz", "out/r2/regions/areas/k1/cells.npz", + f"out/r2/tiles/{tag}/relief/5/0/0.jpg", f"out/r2/tiles/{tag}+abc/relief/5/0/0.jpg"): + self.assertTrue((self.dest / rel).is_file(), rel) + + def test_leaves_private_stale_and_input_files_behind(self): + for rel in ("places/pins.json", "places/views.md", "exports", "lore", "sketch", + "out/r2/regions/0000000000-m7", "out/r2/tiles/v1.1-0000000000"): + self.assertFalse((self.dest / rel).exists(), rel) + self.assertFalse(list(self.dest.rglob("*.part"))) + + def test_the_copy_keeps_the_build_fingerprint(self): + seed = int(C.load(self.dest)[0]["build"]["seed"]) + self.assertEqual(tiles.fingerprint(self.dest / "out" / "r2", seed), self.fp) + + def test_a_second_bundle_keeps_the_servers_own_tiles(self): + mine = self.dest / "out/r2/tiles/v9.9-ffffffffff/relief/9/1/1.jpg" + mine.parent.mkdir(parents=True, exist_ok=True) + mine.write_text("rendered on the server") + with contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(deploy.bundle(self.root, str(self.dest), 2), 0) + self.assertTrue(mine.exists()) + + +class DedupeWorldTest(unittest.TestCase): + def test_pairs_each_era_cache_with_earlier_peers(self): + tmp = Path(tempfile.mkdtemp()) + try: + out = tmp / "out" / "r2" + dirs = [out / "regions" / k / f"{servecache.PREFIX}abc-v1" for k in ("aaa-m7", "bbb-m7", "ccc-m7")] + for d in dirs: + d.mkdir(parents=True) + (d / servecache.DONE).write_text("ok") + calls = [] + orig = deploy.dedupe.dedupe_dirs + deploy.dedupe.dedupe_dirs = lambda new, peers: calls.append((new, sorted(peers))) or 7 + try: + self.assertEqual(deploy.dedupe_world(tmp, 2), 21) + finally: + deploy.dedupe.dedupe_dirs = orig + self.assertEqual(calls, [(dirs[0], []), (dirs[1], [dirs[0]]), (dirs[2], [dirs[0], dirs[1]])]) + finally: + shutil.rmtree(tmp) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_export.py b/tests/test_export.py new file mode 100644 index 0000000..f5e3cb6 --- /dev/null +++ b/tests/test_export.py @@ -0,0 +1,249 @@ +import json +import math +import tempfile +import unittest +from pathlib import Path + +import numpy as np +from PIL import Image + +import export as EX +import serve +import tiles as T +from mapgen import config as C +from tests.test_serve import built_world + + +def _spot(): + """A river mouth on the test world: land, sea and a river in one small square.""" + a = np.load(built_world() / "out" / "r2" / "cells.npz") + mouths = np.where(a["river"].astype(bool) & a["ocean"][a["recv"]].astype(bool))[0] + i = int(mouths[np.argmax(a["discharge_km3_yr"][mouths])]) + t = np.linspace(0, 1, 101)[:, None] # from the mouth toward its sea cell … + p = a["g_xyz"][i] * (1 - t) + a["g_xyz"][a["recv"][i]] * t + 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])) + src = T.TileSource(serve.World(built_world(), 2), seed=int(C.load(built_world())[0]["build"]["seed"]), + cache_dir=Path(tempfile.mkdtemp()), regions_dir=Path(tempfile.mkdtemp())) + wet = src.surface_at(lat, lon, 1.0)[1] + k = int(np.argmax(wet)) if wet.any() else 50 # … to where the map shows the sea: the coast + return float(lat[k]), float(lon[k]) + + +class ExportTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.root = built_world() + cls.lat, cls.lon = _spot() + cls.pins = Path(tempfile.mkdtemp()) / "pins.json" + cls.pins.write_text(json.dumps({"pins": [{"id": "pin-1", "name": "Here", "lat": cls.lat, "lon": cls.lon, + "lore": "somewhere", "note": "", "epoch": None}, + {"id": "pin-2", "name": "Far", "lat": -cls.lat, "lon": cls.lon + 90}]})) + cls.out = Path(tempfile.mkdtemp()) + cls.d = EX.export(cls.root, 2, cls.lat, cls.lon, size_km=200, res_m=2000, out_dir=cls.out, name="t", + pins_path=cls.pins, regions_dir=Path(tempfile.mkdtemp())) + cls.meta = json.loads((cls.d / "meta.json").read_text()) + cls.n = cls.meta["samples"] + cls.h = np.fromfile(cls.d / "height.f32", dtype="<f4").reshape(cls.n, cls.n) + + def test_files_and_grid(self): + self.assertEqual(self.d, self.out / "t") + for f in ("height.f32", "height.png", "height.r16", "water.png", "biome.png", "ground.png", "landform.png", + "preview.png", "rivers.json", "pins.json", "legend.json", "meta.json", "README.txt"): + self.assertTrue((self.d / f).exists(), f) + self.assertEqual(self.n, 129, "2^n + 1 samples, the smallest covering 200 km at 2 km") + self.assertEqual(self.meta["res_m"], 2000) + self.assertEqual(self.meta["size_m"], 128 * 2000) + for f in ("water.png", "biome.png", "preview.png"): + self.assertEqual(Image.open(self.d / f).size, (self.n, self.n), f) + self.assertEqual(self.meta["sea_level_m"], 0) + self.assertIn("world_build", self.meta["sources"]) + + def test_samples_are_res_apart_on_the_ground(self): + lat, lon = EX.grid_latlon(self.lat, self.lon, self.n, 2000.0, 12742.0) + def gc(i1, j1, i2, j2): + p1, p2 = np.radians([lat[i1, j1], lon[i1, j1]]), np.radians([lat[i2, j2], lon[i2, j2]]) + return 12742e3 * 2 * math.asin(math.sqrt(math.sin((p2[0] - p1[0]) / 2) ** 2 + math.cos(p1[0]) * math.cos(p2[0]) + * math.sin((p2[1] - p1[1]) / 2) ** 2)) + c = self.n // 2 + for a, b in (((c, c), (c, c + 1)), ((0, 0), (0, 1)), ((0, 0), (1, 0)), ((self.n - 1, 5), (self.n - 2, 5))): + self.assertAlmostEqual(gc(*a, *b), 2000.0, delta=4.0) + self.assertAlmostEqual(lat[c, c], self.lat, places=9) + self.assertGreater(lat[0, c], lat[-1, c], "row 0 is the north edge") + + def test_heights_are_the_viewers_terrain(self): + src = T.TileSource(serve.World(self.root, 2), seed=int(C.load(self.root)[0]["build"]["seed"]), + cache_dir=Path(tempfile.mkdtemp()), regions_dir=Path(tempfile.mkdtemp())) + lat, lon = EX.grid_latlon(self.lat, self.lon, self.n, 2000.0, 12742.0) + k = np.s_[::8, ::8] + ref = src.z_at(lat[k].ravel(), lon[k].ravel(), self.meta["tile_px_m"] / 1000.0).reshape(lat[k].shape) + self.assertGreater(np.corrcoef(ref.ravel(), self.h[k].ravel())[0, 1], 0.99) + self.assertLess(np.median(np.abs(ref - self.h[k])), 0.1 * ref.std() + 1.0) + + def test_16_bit_heights_decode_to_the_float_heights(self): + lo, hi = self.meta["height"]["min"], self.meta["height"]["max"] + v = np.asarray(Image.open(self.d / "height.png"), dtype=np.float64) + raw = np.fromfile(self.d / "height.r16", dtype="<u2").reshape(self.n, self.n) + np.testing.assert_array_equal(v, raw) + np.testing.assert_allclose(lo + v / 65535 * (hi - lo), self.h, atol=(hi - lo) / 65535 + 1e-3) + + def test_water_classes(self): + w = np.asarray(Image.open(self.d / "water.png")) + self.assertTrue(set(np.unique(w)) <= {0, 1, 2, 3}) + self.assertTrue((w == 1).any() and (w == 0).any(), "the square holds sea and land") + self.assertLess(np.median(self.h[w == 1]), 0.0, "the sea lies below sea level") + self.assertEqual(json.loads((self.d / "legend.json").read_text())["water"], ["land", "sea", "lake", "river"]) + + def test_rivers_and_pins_in_local_metres(self): + rivers = json.loads((self.d / "rivers.json").read_text()) + self.assertTrue(rivers, "the river mouth's river") + half = self.meta["size_m"] / 2 + for r in rivers: + self.assertGreater(r["width_m"], 0) + p = np.asarray(r["points"]) + self.assertTrue(np.all(np.abs(p) <= half + 1e-6)) + pins = json.loads((self.d / "pins.json").read_text()) + self.assertEqual([p["name"] for p in pins], ["Here"]) + self.assertLess(abs(pins[0]["x"]) + abs(pins[0]["y"]), 1.0) + self.assertEqual(pins[0]["lore"], "somewhere") + + def test_preview_shades_the_relief_so_gentle_land_still_shows(self): + pv = self.meta["preview"] + self.assertGreaterEqual(pv["exaggeration"], 1.0) + rgb = np.asarray(Image.open(self.d / "preview.png"), dtype=np.float64) + w = np.asarray(Image.open(self.d / "water.png")) + lum = rgb.mean(axis=2)[w == 0] + self.assertGreater(lum.std(), 2.0, "not a flat colour field") + + def test_deterministic_in_parallel_too(self): + d2 = EX.export(self.root, 2, self.lat, self.lon, size_km=200, res_m=2000, out_dir=self.out, name="t2", + pins_path=self.pins, regions_dir=Path(tempfile.mkdtemp()), workers=2) + for f in ("height.f32", "water.png", "biome.png", "preview.png", "rivers.json", "pins.json"): + self.assertEqual((self.d / f).read_bytes(), (d2 / f).read_bytes(), f) + + def test_exports_into_a_new_folder_leave_no_work_files(self): + out = Path(tempfile.mkdtemp()) / "new" / "exports" + d = EX.export(self.root, 2, self.lat, self.lon, size_km=50, res_m=2000, out_dir=out, name="n", + regions_dir=Path(tempfile.mkdtemp())) + self.assertEqual([p.name for p in out.iterdir()], ["n"]) + self.assertTrue((d / "meta.json").exists()) + + def test_a_region_change_during_an_export_stops_it_cleanly(self): + import refine as RF + import shutil + from tests.test_refine import square + regions = Path(tempfile.mkdtemp()) + reg = regions / "r.json" + reg.write_text(json.dumps({"regions": [square(self.lat, self.lon, 3)]})) + built = Path(tempfile.mkdtemp()) + RF.build_areas(self.root, 2, reg, log=lambda s: None, regions_root=built) + src = T.TileSource(serve.World(self.root, 2), seed=int(C.load(self.root)[0]["build"]["seed"]), + cache_dir=Path(tempfile.mkdtemp()), regions_dir=regions / "live") + real, calls = src.export_tile, [] + + def tile(*a): + calls.append(a) + if len(calls) == 2: # a region build finishes mid-export + shutil.copytree(built, regions / "live") + src.reload_regions() + return real(*a) + src.export_tile = tile + out = Path(tempfile.mkdtemp()) + with self.assertRaises(RuntimeError): + EX.export(self.root, 2, self.lat, self.lon, size_km=200, res_m=2000, out_dir=out, name="mixed", src=src) + self.assertEqual(list(out.iterdir()), [], "nothing half-mixed written") + + def test_an_existing_export_is_not_overwritten(self): + with self.assertRaises(FileExistsError): + EX.export(self.root, 2, self.lat, self.lon, size_km=200, res_m=2000, out_dir=self.out, name="t", + pins_path=self.pins, regions_dir=Path(tempfile.mkdtemp())) + + +class CliTest(unittest.TestCase): + def test_export_command(self): + import importlib.util + spec = importlib.util.spec_from_file_location("mapview_cli", Path(EX.__file__).parent / "mapview.py") + cli = importlib.util.module_from_spec(spec) + spec.loader.exec_module(cli) + lat, lon = _spot() + out = Path(tempfile.mkdtemp()) + code = cli.main(["export", "--res", "2", "--lat", str(lat), "--lon", str(lon), "--size-km", "100", + "--res-m", "2000", "--name", "cli", "--out", str(out), "--workers", "0"], root=built_world()) + self.assertEqual(code, 0) + self.assertEqual(json.loads((out / "cli" / "meta.json").read_text())["samples"], 65) + + def test_exact_grid_512_m_at_1_m(self): + import importlib.util + spec = importlib.util.spec_from_file_location("mapview_cli", Path(EX.__file__).parent / "mapview.py") + cli = importlib.util.module_from_spec(spec) + spec.loader.exec_module(cli) + lat, lon = _spot() + out = Path(tempfile.mkdtemp()) + self.assertEqual(cli.main(["export", "--res", "2", "--lat", str(lat), "--lon", str(lon), "--size-km", "0.512", "--res-m", "1", "--pixels", "513", + "--name", "c", "--out", str(out), "--workers", "0"], root=built_world()), 0) + meta = json.loads((out / "c" / "meta.json").read_text()) + self.assertEqual((meta["samples"], meta["res_m"], meta["size_m"]), (513, 1, 512)) + h = np.fromfile(out / "c" / "height.f32", dtype="<f4").reshape(513, 513) + self.assertAlmostEqual(meta["center_height_m"], float(h[256, 256]), places=3) + +class ExportForkTest(unittest.TestCase): + def test_region_trees_are_built_before_export_workers_fork(self): + import refine + from unittest import mock + order = [] + + class FakePool: # stands in for the forked pool: records when it forks + def __init__(self, n): + order.append("fork") + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def imap(self, f, jobs, chunksize=1): + return map(f, jobs) + warm = refine.RegionSet.warm + a = np.load(built_world() / "out" / "r2" / "cells.npz") + i = int(np.argmax(a["z_surface_m"])) + with mock.patch.object(refine.RegionSet, "warm", autospec=True, + side_effect=lambda rs: (order.append("warm"), warm(rs))[1]), \ + mock.patch.object(EX.multiprocessing, "get_context", return_value=mock.Mock(Pool=FakePool)): + EX.export(built_world(), 2, float(a["g_lat"][i]), float(a["g_lon"][i]), size_km=50, res_m=2000, + out_dir=Path(tempfile.mkdtemp()), name="w", workers=2) + self.assertIn("warm", order) + self.assertLess(order.index("warm"), order.index("fork")) + + +if __name__ == "__main__": + unittest.main() + + +class EraExportTest(unittest.TestCase): + def test_export_of_an_era_with_sea_and_zone_metadata(self): + import json + import export + from tests.test_refine import era_world + root, _, cut, _ = era_world() + out = Path(tempfile.mkdtemp()) + a = export.export(root, 2, cut[0], cut[1], size_km=20.0, res_m=500.0, out_dir=out, name="after-x") + b = export.export(root, 2, cut[0], cut[1], size_km=20.0, res_m=500.0, out_dir=out, name="before-x", + era="before") + ma, mb = (json.loads((d / "meta.json").read_text()) for d in (a, b)) + self.assertEqual((ma["era"]["name"], mb["era"]["name"]), ("after", "before")) + self.assertLess(ma["center_height_m"], mb["center_height_m"], "the cut lowered the ground") + for m in (ma, mb): + self.assertEqual(set(m["water"]), {"sea_fraction", "max_depth_m", "center_depth_m"}) + self.assertEqual(set(m["zones"]), {"gravity_g", "o2_fraction", "po2_bar", "pressure_bar", "fire_reactivity"}) + self.assertIsInstance(m["vents"], list) + self.assertEqual(m["format"], "worldmap-terrain-export") + self.assertAlmostEqual(ma["zones"]["fire_reactivity"], 0.4, places=3) + with self.assertRaises(SystemExit): + export.export(root, 2, cut[0], cut[1], size_km=20.0, res_m=500.0, out_dir=out, name="n", era="nope") + from unittest import mock + import serve + real, made = serve.World, [] + with mock.patch.object(serve, "World", side_effect=lambda *a, **k: made.append(k.get("era")) or real(*a, **k)): + export.export(root, 2, cut[0], cut[1], size_km=20.0, res_m=500.0, out_dir=out, name="one", era="before") + self.assertEqual([e["name"] for e in made], ["before"], "only the chosen era's world is loaded") diff --git a/tests/test_refine.py b/tests/test_refine.py new file mode 100644 index 0000000..3284dc5 --- /dev/null +++ b/tests/test_refine.py @@ -0,0 +1,805 @@ +import json +import tempfile +import unittest +from pathlib import Path + +import h3.api.basic_int as h3 +import numpy as np + +import refine as RF +from mapgen import config as C + + +def square(lat, lon, d, rid="r1", name="test"): + return {"id": rid, "name": name, "outline": [[lat - d, lon - d], [lat - d, lon + d], [lat + d, lon + d], [lat + d, lon - d]]} + + +class AreasTest(unittest.TestCase): + def test_regions_file(self): + self.assertEqual(RF.load_regions(Path(tempfile.mkdtemp()) / "none.json"), []) + p = Path(tempfile.mkdtemp()) / "regions.json" + p.write_text(json.dumps({"regions": [square(10, 20, 1)]})) + self.assertEqual(RF.load_regions(p)[0]["name"], "test") + + def test_cells_inside_the_outline(self): + (cells,) = RF.area_cells([square(10, 20, 1)], 4) + lat = np.array([h3.cell_to_latlng(int(c))[0] for c in cells]) + lon = np.array([h3.cell_to_latlng(int(c))[1] for c in cells]) + self.assertTrue(len(cells) > 10) + self.assertTrue(np.all(np.abs(lat - 10) <= 1.1) and np.all(np.abs(lon - 20) <= 1.1)) + self.assertTrue(np.all(np.diff(cells.astype(np.int64)) > 0), "sorted, unique") + + def test_touching_outlines_merge_separate_ones_dont(self): + a, b = square(10, 20, 1, "a"), square(10, 22, 1, "b") # share the lon 21 edge + c = square(-30, 100, 1, "c") + areas = RF.area_cells([a, b, c], 4) + self.assertEqual(len(areas), 2) + self.assertEqual(sorted(len(x) for x in areas)[1], len(RF.area_cells([a], 4)[0]) + len(RF.area_cells([b], 4)[0]) - len(np.intersect1d(RF.area_cells([a], 4)[0], RF.area_cells([b], 4)[0]))) + + def test_hash_follows_the_cells_only(self): + (x,) = RF.area_cells([square(10, 20, 1, "a", "one")], 4) + (y,) = RF.area_cells([square(10, 20, 1, "b", "renamed")], 4) + (w,) = RF.area_cells([square(10, 20, 1.5)], 4) + self.assertEqual(RF.area_hash(x), RF.area_hash(y)) + self.assertNotEqual(RF.area_hash(x), RF.area_hash(w)) + + def test_sea_only_where_it_reaches_the_open_ocean(self): + """World ocean cut off by a refined land strip is no sea (a lagoon: the lakes take it); as the world's own rule, + a separate basin counts as sea only when it is that big.""" + (cells,) = RF.area_cells([square(10, 20, 3)], 4) + g, halo = RF.subgrid(cells, 12742.0) + d = np.hypot(g.lat - 10, g.lon - 20) + z = np.where((d > 0.9) & (d < 1.6), 5.0, -10.0) # a ring of land around a wet middle + ocean_world = np.ones(g.n, bool) + inner, outer = d <= 0.9, d >= 1.6 + self.assertTrue(inner.sum() > 3 and (outer & halo).any()) + sea = RF._sea(g, z, ocean_world, halo) + self.assertTrue(sea[outer].all(), "open water out to the area's edge: sea") + self.assertFalse(sea[inner].any(), "behind the strip: not sea") + big = RF._sea(g, z, ocean_world, halo, min_km2=float(g.area_km2[inner].sum()) * 0.9) + self.assertTrue(big[inner].all(), "a separate basin as big as a world sea stays sea") + self.assertFalse(RF._sea(g, z, np.zeros(g.n, bool), halo).any(), "never where the world has no ocean") + + def test_world_lake_keeps_its_world_outflow(self): + """A world lake that drains to the world sea keeps that way out at its world level: the fine relief must not + dam it higher (r5 2026-10-04: lakes at 5 m stood at 29 m and drowned their shores). Only its world drainage + path is cut; a lake that already drains low enough is left alone.""" + (cells,) = RF.area_cells([square(10, 20, 3)], 4) + g, halo = RF.subgrid(cells, 12742.0) + parent = np.where(g.lon < 19, 0, np.where(g.lon < 21, 1, 2)) # world: lake → land → ocean (west to east) + parent = np.where((g.lat > 11.5) & (parent == 1), 3, parent) # and land off the drainage path + world = {"lake": np.array([1, 0, 0, 0], bool), "ocean": np.array([0, 0, 1, 0], bool), + "lake_id": np.array([7, -1, -1, -1]), "lake_level_m": np.array([5.0, np.nan, np.nan, np.nan]), + "recv": np.array([1, 2, 2, 3])} + z = np.select([parent == 0, parent == 2], [-20.0, -900.0], 30.0) # a 30 m sill between lake and a deep sea + sea = RF._sea(g, z, world["ocean"][parent], halo) + lake = (parent == 0) & ~halo + + def level(zz): + return float(RF.priority_flood(g, zz, sea)[lake].max()) + self.assertGreater(level(z), 25.0, "the fine sill dams the lake") + cut = RF.world_outflows(g, z, halo, sea, parent, world) + self.assertLessEqual(level(cut), 5.0, "the lake drains at its world level") + self.assertTrue((cut <= z).all(), "only lowers") + self.assertTrue((cut[parent == 3] == z[parent == 3]).all(), "off the world path: untouched") + changed = cut < z + self.assertTrue(0 < changed.sum() < (parent == 1).sum() / 3, "a channel, not the whole cell") + self.assertTrue((cut[changed] >= 0.0).all(), "never cut below sea level (the sea's surface, not its bed)") + low = np.where(parent == 1, 2.0, z) + self.assertTrue((RF.world_outflows(g, low, halo, sea, parent, world) == low).all(), "drains already: untouched") + + def test_subgrid_with_halo(self): + (cells,) = RF.area_cells([square(10, 20, 1)], 4) + g, halo = RF.subgrid(cells, 12742.0) + self.assertEqual(int((~halo).sum()), len(cells)) + self.assertTrue(halo.sum() > 0) + pairs = set(zip(g.src.tolist(), g.dst.tolist())) + self.assertTrue(all((j, i) in pairs for i, j in pairs), "symmetric neighbours") + inner = np.where(~halo)[0] + self.assertTrue(np.all(g.counts[inner] >= 5), "every area cell keeps all its neighbours (pentagons: 5)") + self.assertAlmostEqual(float(g.area_km2[inner].mean()), h3.average_hexagon_area(4, unit="km^2") * 4, delta=0.3 * h3.average_hexagon_area(4, unit="km^2") * 4) + + +from tests.test_serve import built_world +from tests.helpers import big_river +import serve +import tiles as T + + +class FieldsTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.root = built_world() + cls.world = RF.WorldCells(cls.root / "out" / "r2") + cls.src = T.TileSource(serve.World(cls.root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp())) + i = big_river(cls.world.a) + lat, lon = float(cls.world.a["g_lat"][i]), float(cls.world.a["g_lon"][i]) + (cls.cells,) = RF.area_cells([square(lat, lon, 4)], 4) + cls.g, cls.halo = RF.subgrid(cls.cells, 12742.0) + cls.f = RF.initial_fields(cls.g, cls.halo, cls.world, cls.src) + cls.cfg = C.load(cls.root)[0] + cls.r = RF.refine_area(cls.g, cls.halo, cls.world, cls.src, cls.cfg) + + def test_parents_are_the_world_cells(self): + p = self.f["parent"] + np.testing.assert_array_equal(self.world.a["g_ids"][p], [h3.cell_to_parent(int(c), 2) for c in self.g.ids]) + + def test_elevation_is_the_world_surface_plus_coarse_relief(self): + z = self.src.relief_at(self.g.lat, self.g.lon, RF.RELIEF_MIN_KM) + np.testing.assert_allclose(self.f["elevation_m"], z) + + def test_temperature_follows_the_lapse_rate(self): + p = self.f["parent"] + dz = (self.f["elevation_m"] - self.world.a["z_surface_m"][p]) / 1000.0 + np.testing.assert_allclose(self.f["T_mean"], self.world.a["T_mean"][p] - RF.LAPSE_C_PER_KM * dz, atol=1e-4) + np.testing.assert_array_equal(self.f["lithology"], self.world.a["lithology"][p]) + + def test_crossings_find_the_big_river(self): + inflow, exits, lev = RF.crossings(self.g, self.halo, self.world) + self.assertTrue(inflow.sum() > 0 or len(exits) > 0, "the widest world river crosses the square") + self.assertTrue(np.all(self.halo[exits])) + self.assertTrue(np.all(inflow[self.halo] == 0), "inflow enters inside the area") + + def test_crossings_match_scanning_every_world_river(self): + """The near-area prefilter drops only segments that cannot cross (oracle: the unfiltered scan).""" + from unittest import mock + got = RF.crossings(self.g, self.halo, self.world) + with mock.patch.object(RF, "world_near", lambda w, cells, rings=3: np.arange(len(w.ids))): + want = RF.crossings(self.g, self.halo, self.world) + for a, b in zip(got, want): + np.testing.assert_array_equal(a, b) + + def test_erosion_keeps_the_halo_and_the_sea_and_cuts_valleys(self): + f = self.f + sea = f["ocean_world"] & (f["elevation_m"] <= 0) + inflow, _, _ = RF.crossings(self.g, self.halo, self.world) + water = RF.runoff_km3(self.g, f) + inflow + z = RF.erode_area(self.g, f["elevation_m"], self.halo, sea, water, np.ones(self.g.n)) + np.testing.assert_array_equal(z[self.halo], f["elevation_m"][self.halo]) + np.testing.assert_array_equal(z[sea], f["elevation_m"][sea]) + self.assertTrue((f["elevation_m"] - z).max() > 5.0, "valleys cut") + self.assertTrue(np.all(z <= f["elevation_m"] + 1e-6), "erosion only lowers") + + def test_schema_and_halo(self): + for k in ("z_surface_m", "river", "recv", "discharge_km3_yr", "lake", "holdridge", "ground", "ice", "po2_bar"): + self.assertEqual(len(self.r[k]), self.g.n, k) + np.testing.assert_array_equal(self.r["halo"], self.halo) + self.assertFalse(self.r["river"][self.halo].any()) + + def test_water_balance(self): + s = RF.stats(self.r) + self.assertLess(s["water_balance_error"], 0.01) + + def test_exit_where_the_world_exits(self): + _, exits, _ = RF.crossings(self.g, self.halo, self.world) + into = self.r["recv"][~self.halo] + big = np.where(self.r["river"] & ~self.halo & self.halo[self.r["recv"]])[0] # rivers leaving the area + if len(exits): + leave = set(self.r["recv"][big].tolist()) + self.assertTrue(set(exits.tolist()) & leave, "the main river leaves through a world exit cell") + + def test_no_river_dead_ends(self): + r, recv = self.r, self.r["recv"] + riv = np.where(r["river"])[0] + ok = r["river"][recv[riv]] | r["lake"][recv[riv]] | r["ocean"][recv[riv]] | r["halo"][recv[riv]] | r["endorheic"][recv[riv]] + self.assertTrue(np.all(ok)) + + def test_sea_stays_sea(self): + sea = self.f["ocean_world"] & (self.f["elevation_m"] <= 0) + self.assertTrue(np.all(self.r["ocean"][sea])) + + def test_deterministic(self): + again = RF.refine_area(self.g, self.halo, self.world, self.src, self.cfg) + for k in ("z_surface_m", "discharge_km3_yr", "river"): + np.testing.assert_array_equal(again[k], self.r[k]) + + def test_streams_from_a_fifth_of_a_km3(self): + q = self.r["discharge_km3_yr"] + land = ~self.r["ocean"] & ~self.r["lake"] & ~self.halo + np.testing.assert_array_equal(self.r["river"][land], q[land] >= RF.RIVER_MIN_KM3_YR) + + def test_lakes_only_in_real_hollows(self): + deep = self.r["depression_depth_m"] > RF.LAKE_MIN_DEPTH_M + self.assertFalse((self.r["lake"] & ~deep & ~self.r["endorheic"] & ~self.halo).any(), "no lakes in shallow dips") + + def test_shoulder_heights_bridge_the_rivers(self): + env = RF.shoulders(self.r) + riv = self.r["river"] & ~self.halo + self.assertTrue(np.all(env[riv] >= self.r["z_surface_m"][riv] - 1e-6)) + self.assertTrue(np.all(env[~riv] == self.r["z_surface_m"][~riv])) + + def test_an_inflowing_river_carries_its_water_into_the_area(self): + inflow, _, _ = RF.crossings(self.g, self.halo, self.world) + if inflow.max() > 0: + q = self.r["discharge_km3_yr"][~self.halo] + self.assertGreaterEqual(q.max(), 0.9 * inflow.max(), "not bounced straight back out through the halo") + outlets = RF.outlets(self.g, self.halo, self.world, self.f["parent"]) + self.assertTrue(outlets.any() and np.all(self.halo[outlets])) + self.assertLess(outlets.sum(), self.halo.sum(), "only some halo cells are ways out") + + def test_no_block_steps_between_world_cells(self): + z, p, g = self.r["elevation_eroded_m"], self.f["parent"], self.g + inner = ~self.halo[g.src] & ~self.halo[g.dst] & ~self.r["ocean"][g.src] & ~self.r["ocean"][g.dst] + d = np.abs(z[g.src] - z[g.dst])[inner] + across = (p[g.src] != p[g.dst])[inner] + if across.sum() > 10 and (~across).sum() > 10: + self.assertLess(np.percentile(d[across], 95), 1.3 * np.percentile(d[~across], 95) + 5.0) + + def test_world_cell_means_of_bedrock_are_kept_smoothly(self): + z, p = self.r["elevation_eroded_m"], self.f["parent"] + for w in np.unique(p[~self.halo]): + m = (p == w) & ~self.halo + floor = (self.r["canyon_m"][m] > 0).any() or (self.r["fan_m"][m] > 0).any() # the sea-floor pass: sea only + if m.sum() == 49 and not self.world.a["ocean"][w] and not floor: + self.assertAlmostEqual(float(z[m].mean()), float(self.world.a["elevation_eroded_m"][w]), delta=25.0) + + def test_no_fake_lakes_against_walls(self): + g, r = self.g, self.r + wall = self.halo & ~r["outlet"] + touching = set(r["lake_id"][g.src[wall[g.dst] & r["lake"][g.src]]].tolist()) + for lid in touching: # a wall lower than the lake would have let it spill out + m = r["lake"] & (r["lake_id"] == lid) + walls = g.dst[m[g.src] & wall[g.dst]] + level = float(r["z_filled_m"][m].max()) + self.assertTrue(np.all(r["elevation_eroded_m"][walls] >= level - 1e-6), f"lake {lid} dammed by a wall") + + def test_temperatures_follow_the_final_heights(self): + p = self.f["parent"] + dz = (self.r["elevation_eroded_m"] - self.world.a["elevation_eroded_m"][p]) / 1000.0 + np.testing.assert_allclose(self.r["T_mean"], self.world.a["T_mean"][p] - RF.LAPSE_C_PER_KM * dz, atol=1e-3) + + def test_border_report(self): + s = RF.stats(self.r) + self.assertIn("exit_level_error_m", s) + if s["exit_level_error_m"] is not None: + self.assertLess(abs(s["exit_level_error_m"]), 50.0, "rivers leave at the world river's level") + + def test_zone_fields_come_from_the_world_cells(self): + from mapgen import fields as FL + p, a, r = self.f["parent"], self.world.a, self.r + for k in ("gravity_g", "o2_fraction", "fire_reactivity", "plant_height_x"): + np.testing.assert_array_equal(r[k], a[k][p], err_msg=k) + H = float(self.cfg["planet"]["scale_height_km"]) * 1000.0 + sea = a["pressure_bar"][p] / FL.pressure(1.0, a["z_surface_m"][p], H) + np.testing.assert_allclose(r["pressure_bar"], sea * FL.pressure(1.0, r["z_surface_m"], H), rtol=1e-5) + np.testing.assert_allclose(r["po2_bar"], r["o2_fraction"] * r["pressure_bar"], rtol=1e-5) + + +def clear_regions(root): # built_world() is shared across tests: leave no regions behind + import shutil + shutil.rmtree(root / "out" / "r2" / "regions", ignore_errors=True) + (root / "places" / "regions.json").unlink(missing_ok=True) + + +class BuildTest(unittest.TestCase): + def tearDown(self): + clear_regions(built_world()) + + def test_build_caches_per_world_and_skips_when_done(self): + root = built_world() + w = RF.WorldCells(root / "out" / "r2") + i = big_river(w.a) + reg = root / "places" / "regions.json" + reg.parent.mkdir(exist_ok=True) + reg.write_text(json.dumps({"regions": [square(float(w.a["g_lat"][i]), float(w.a["g_lon"][i]), 2.5)]})) # its own area (built_world is shared) + m1 = RF.build_areas(root, 2, reg, log=lambda s: None) + self.assertEqual(len(m1), 1) + self.assertTrue(m1[0]["built"]) + d = Path(m1[0]["dir"]) + self.assertTrue((d / "cells.npz").exists() and (d / "meta.json").exists()) + m2 = RF.build_areas(root, 2, reg, log=lambda s: None) + self.assertFalse(m2[0]["built"], "cached: not rebuilt") + stale = d.parent.parent / "0000000000-m1" + stale.mkdir() + RF.build_areas(root, 2, reg, log=lambda s: None) + self.assertFalse(stale.exists(), "results of other world builds are removed") + + def test_redrawn_and_deleted_regions_leave_no_old_area(self): + root = built_world() + w = RF.WorldCells(root / "out" / "r2") + i = big_river(w.a) + la, lo = float(w.a["g_lat"][i]), float(w.a["g_lon"][i]) + reg = root / "places" / "regions.json" + reg.parent.mkdir(exist_ok=True) + reg.write_text(json.dumps({"regions": [square(la, lo, 2)]})) + (m1,) = RF.build_areas(root, 2, reg, log=lambda s: None) + reg.write_text(json.dumps({"regions": [square(la, lo, 3)]})) # redrawn larger + (m2,) = RF.build_areas(root, 2, reg, log=lambda s: None) + self.assertFalse(Path(m1["dir"]).exists(), "the old area is gone") + (Path(m2["dir"]) / "meta.json").write_text("{broken") # a crash mid-save + (m3,) = RF.build_areas(root, 2, reg, log=lambda s: None) + self.assertTrue(m3["built"], "a broken result is rebuilt, not fatal") + reg.write_text(json.dumps({"regions": []})) + self.assertEqual(RF.build_areas(root, 2, reg, log=lambda s: None), []) + self.assertEqual(RF.built_areas(root / "out" / "r2" / "regions", Path(m2["dir"]).parent.name.rsplit("-m", 1)[0]), []) + + + def test_a_failed_build_keeps_the_previous_result(self): + from unittest import mock + root = built_world() + w = RF.WorldCells(root / "out" / "r2") + i = big_river(w.a) + la, lo = float(w.a["g_lat"][i]), float(w.a["g_lon"][i]) + tmp = Path(tempfile.mkdtemp()) + reg = tmp / "regions.json" + reg.write_text(json.dumps({"regions": [square(la, lo, 2)]})) + (m1,) = RF.build_areas(root, 2, reg, log=lambda s: None, regions_root=tmp / "res") + reg.write_text(json.dumps({"regions": [square(la, lo, 3)]})) # extended … + with mock.patch.object(RF, "refine_area", side_effect=RuntimeError("boom")), self.assertRaises(RuntimeError): + RF.build_areas(root, 2, reg, log=lambda s: None, regions_root=tmp / "res") # … and the build fails + self.assertTrue((Path(m1["dir"]) / "cells.npz").exists(), "the old area stays until a build succeeds") + + +class LandOutlineTest(unittest.TestCase): + def test_outline_of_a_landmass_covers_it_simply(self): + root = built_world() + a = RF.WorldCells(root / "out" / "r2").a + land = ~a["ocean"].astype(bool) + i = int(np.argmax(land * a["z_surface_m"])) # a point on the highest landmass + outline = RF.land_outline(a, float(a["g_lat"][i]), float(a["g_lon"][i]), step_deg=2.0) + self.assertGreaterEqual(len(outline), 4) + self.assertLessEqual(len(outline), 500) + import serve + self.assertFalse(serve.RegionStore._crosses([(lo, la) for la, lo in outline]), "a simple polygon") + cells = set(h3.polygon_to_cells(h3.LatLngPoly([tuple(p) for p in outline]), 2)) + comp = RF.landmass(a, float(a["g_lat"][i]), float(a["g_lon"][i])) + inside = np.isin(a["g_ids"][comp], np.array(sorted(cells), dtype=np.uint64)) + self.assertGreater(inside.mean(), 0.97, "the landmass lies inside its outline") + + +class RecordTest(unittest.TestCase): + def test_the_record_keeps_what_was_built_not_later_edits(self): + from unittest import mock + root = built_world() + w = RF.WorldCells(root / "out" / "r2") + i = big_river(w.a) + la, lo = float(w.a["g_lat"][i]), float(w.a["g_lon"][i]) + tmp = Path(tempfile.mkdtemp()) + reg = tmp / "regions.json" + a, b = square(la, lo, 2, "a"), square(-la, lo + 90, 2, "b") + reg.write_text(json.dumps({"regions": [a]})) + real = RF.refine_area + + def edit_meanwhile(*args, **kw): + reg.write_text(json.dumps({"regions": [a, b]})) # region b drawn during the build + return real(*args, **kw) + with mock.patch.object(RF, "refine_area", side_effect=edit_meanwhile): + RF.build_areas(root, 2, reg, log=lambda s: None, regions_root=tmp / "res") + key = next((tmp / "res").glob("*-m*")).name.rsplit("-m", 1)[0] + rec = RF.read_record(tmp / "res", key)["regions"] + self.assertEqual(set(rec), {"a"}, "b was not built: it must not look built") + + +def _subgrid_reference(cells, radius_km): # the first implementation (sets and lists): the oracle + inner = set(int(c) for c in cells) + ring = {n for c in inner for n in h3.grid_ring(c, 1)} - inner + ids = np.array(sorted(inner | ring), dtype=np.uint64) + rings = [[n for n in h3.grid_ring(int(c), 1)] for c in ids] + pos = {int(c): k for k, c in enumerate(ids)} + nb = [[pos[n] for n in r if n in pos] for r in rings] + ptr = np.concatenate([[0], np.cumsum([len(x) for x in nb])]).astype(np.int64) + idx = np.array([k for x in nb for k in x], dtype=np.int64) + return ids, ptr, idx, ~np.isin(ids, cells) + + +def _area_cells_reference(regions, fine_res): # the first implementation (flood fill over every cell) + cells = set() + for r in regions: + cells.update(h3.polygon_to_cells(h3.LatLngPoly([tuple(p) for p in r["outline"]]), fine_res)) + left, out = set(cells), [] + while left: + seed = left.pop() + comp, stack = [seed], [seed] + while stack: + for n in h3.grid_ring(stack.pop(), 1): + if n in left: + left.discard(n) + comp.append(n) + stack.append(n) + out.append(np.array(sorted(comp), dtype=np.uint64)) + return sorted(out, key=lambda a: int(a[0])) + + +class LeanBuildTest(unittest.TestCase): + def test_subgrid_is_the_same_grid(self): + (cells,) = RF.area_cells([square(10, 20, 4)], 4) + g, halo = RF.subgrid(cells, 12742.0) + ids, ptr, idx, h = _subgrid_reference(cells, 12742.0) + np.testing.assert_array_equal(g.ids, ids) + np.testing.assert_array_equal(g.nbr_ptr, ptr) + np.testing.assert_array_equal(g.nbr_idx, idx) + np.testing.assert_array_equal(halo, h) + np.testing.assert_allclose(g.lat, [h3.cell_to_latlng(int(c))[0] for c in ids]) + + def test_areas_are_the_same_areas(self): + cases = [[square(10, 20, 3, "a"), square(10, 25.9, 3, "b")], # touching: one area + [square(10, 20, 3, "a"), square(10, 40, 3, "b")], # apart: two + [square(10, 20, 6, "a"), square(10, 20, 1, "b")], # nested + [square(-30, 170, 3, "a"), square(-30, -176, 3, "b"), square(40, 0, 2, "c")]] # across ±180 + for regions in cases: + got, want = RF.area_cells(regions, 4), _area_cells_reference(regions, 4) + self.assertEqual(len(got), len(want)) + for a, b in zip(got, want): + np.testing.assert_array_equal(a, b) + + def test_a_build_does_not_load_the_refined_regions(self): + from unittest import mock + root = built_world() + tmp = Path(tempfile.mkdtemp()) + reg = tmp / "r.json" + reg.write_text(json.dumps({"regions": [square(10, 20, 3)]})) + RF.build_areas(root, 2, reg, log=lambda s: None, regions_root=tmp / "res") + reg.write_text(json.dumps({"regions": [square(10, 20, 3), square(-20, 60, 3, "b")]})) + real = RF.RegionSet + + def only_empty(root_, *a, **k): + rs = real(root_, *a, **k) + if not rs.empty: + raise AssertionError("a build loaded the refined regions (memory for nothing)") + return rs + with mock.patch.object(RF, "RegionSet", side_effect=only_empty): + RF.build_areas(root, 2, reg, log=lambda s: None, regions_root=tmp / "res") + + +class FingerprintTest(unittest.TestCase): + def test_region_fingerprint_changes_when_a_result_is_rebuilt(self): + d = Path(tempfile.mkdtemp()) + area = RF.results_dir(d, "abc") / "a1" + area.mkdir(parents=True) + np.savez(area / "cells.npz", g_ids=np.array([1], np.uint64)) + (area / "meta.json").write_text("{}") + f1 = RF.regions_fingerprint(d, "abc") + import os, time + st = (area / "cells.npz").stat() + os.utime(area / "cells.npz", ns=(st.st_atime_ns, st.st_mtime_ns + 10**9)) # rebuilt (new model, same cells) + self.assertNotEqual(RF.regions_fingerprint(d, "abc"), f1) + + +class HaloHeightsTest(unittest.TestCase): + def test_halo_keeps_real_heights_in_the_output(self): + root = built_world() + w = RF.WorldCells(root / "out" / "r2") + src = T.TileSource(serve.World(root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp())) + i = big_river(w.a) + (cells,) = RF.area_cells([square(float(w.a["g_lat"][i]), float(w.a["g_lon"][i]), 4)], 4) + g, halo = RF.subgrid(cells, 12742.0) + r = RF.refine_area(g, halo, w, src, C.load(root)[0]) + self.assertLess(float(np.max(r["z_surface_m"])), 20000.0, "no routing walls leak into heights") + + +class CoastTest(unittest.TestCase): + def test_coastal_area_keeps_land_and_rivers_above_the_sea(self): + root = built_world() + w = RF.WorldCells(root / "out" / "r2") + a = w.a + mouths = np.where(a["river"].astype(bool) & a["ocean"][a["recv"]].astype(bool))[0] + i = int(mouths[np.argmax(a["discharge_km3_yr"][mouths])]) + (cells,) = RF.area_cells([square(float(a["g_lat"][i]), float(a["g_lon"][i]), 4)], 4) + g, halo = RF.subgrid(cells, 12742.0) + src = T.TileSource(serve.World(root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp())) + r = RF.refine_area(g, halo, w, src, C.load(root)[0]) + inner = ~halo + self.assertTrue((r["ocean"] & inner).any(), "the test area reaches the sea") + self.assertFalse((r["river"] & inner & (r["elevation_eroded_m"] < 0)).any(), "no river below sea level") + lake_low = r["lake"] & inner & ~r["endorheic"] & (r["z_filled_m"] < 0) + self.assertFalse(lake_low.any(), "no open lake below sea level") + coast_land = inner & ~r["ocean"] & np.isin(np.arange(g.n), g.src[r["ocean"][g.dst]]) + self.assertTrue(np.all(r["elevation_eroded_m"][coast_land] >= 0), "land by the sea is above it") + + +ERA_CUT = """ +[[event]] +name = "cut" +kind = "disintegrate" +center = [{lat}, {lon}] +radius_km = 1000.0 +depth_m = 3000.0 +[[event]] +name = "haze" +kind = "zone" +shape = "circle" +center = [{lat}, {lon}] +radius_km = 1500.0 +edge_km = 5.0 +fields = {{ gravity_g = 0.5, pressure_bar = 2.0, fire_reactivity = 0.4 }} +[eras] +order = ["before", "after"] +default = "after" +[eras.before] +label = "Before" +events = [] +[eras.after] +label = "After" +events = ["cut", "haze"] +""" + +_ERA_WORLD = [] + + +def era_world(): + """(root, far point, cut point, km between them): a small world with an era 'after' that cuts into the land + farthest from the biggest river and holds a zone there (built once per test run).""" + if not _ERA_WORLD: + import atexit + import shutil + from mapgen import eras as ER, pipeline as P + from mapgen.sphere import gc_dist_km + from mapgen.testing import small_world + root = Path(tempfile.mkdtemp(prefix="worldmap-viewer-era-")) + atexit.register(shutil.rmtree, root, True) + small_world(root) + ctx = P.build(root, 2, log=lambda m: None) + g, a = ctx.grid, ctx.data + i = big_river(a) + d = gc_dist_km(g.xyz, g.xyz[i], g.radius_km) + j = int(np.argmax(np.where(~np.asarray(a["ocean"]), d, -1.0))) + (root / "config" / "eras.toml").write_text(ERA_CUT.format(lat=round(float(g.lat[j]), 3), + lon=round(float(g.lon[j]), 3))) + ER.build_all(root, 2, log=lambda m: None, base=ctx) + _ERA_WORLD.append((root, (float(g.lat[i]), float(g.lon[i])), (float(g.lat[j]), float(g.lon[j])), float(d[j]))) + return _ERA_WORLD[0] + + +def write_regions(path: Path, *squares) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"regions": list(squares)})) + return path + + +class EraAreasTest(unittest.TestCase): + def test_an_area_the_era_left_alone_is_built_once(self): + root, far, cut, dist = era_world() + self.assertGreater(dist, 5500.0, "the fixture needs the river far from the cut") + tmp = Path(tempfile.mkdtemp()) + reg = write_regions(tmp / "regions.json", square(*far, 2, "far"), square(*cut, 2, "cut")) + metas = RF.build_areas(root, 2, reg, log=lambda s: None, regions_root=tmp / "res") + self.assertEqual(sorted({m["era"] for m in metas}), ["after", "before"]) + by = {(m["era"], m["hash"]): m for m in metas} + hf = RF.area_hash(RF.area_cells([square(*far, 2)], 4)[0]) + hc = RF.area_hash(RF.area_cells([square(*cut, 2)], 4)[0]) + self.assertEqual(by[("before", hf)]["key"], by[("after", hf)]["key"]) + self.assertNotEqual(by[("before", hc)]["key"], by[("after", hc)]["key"]) + self.assertEqual(sorted(p.name for p in RF.areas_dir(tmp / "res").iterdir() if p.is_dir()), + sorted({m["key"] for m in metas})) + self.assertEqual(len({m["key"] for m in metas}), 3) + sets = [p for p in (tmp / "res").iterdir() if p.name != RF.AREAS] + self.assertEqual(len(sets), 2) + for s in sets: + links = RF.built_areas(tmp / "res", s.name.rsplit("-m", 1)[0]) + self.assertEqual(len(links), 2) + self.assertTrue(all(p.is_symlink() for p in links)) + self.assertIn(json.loads((s / "era.json").read_text())["era"], ("before", "after")) + again = RF.build_areas(root, 2, reg, log=lambda s: None, regions_root=tmp / "res") + self.assertFalse(any(m["built"] for m in again), "cached for both worlds") + + def test_key_follows_the_values_the_area_reads(self): + import copy + root, far, _, _ = era_world() + cfg, _ = C.load(root) + (cells,) = RF.area_cells([square(*far, 2)], 4) + w = RF.WorldCells(root / "out" / "r2") + k0 = RF.area_key(w, cells, cfg, []) + self.assertEqual(RF.area_key(RF.WorldCells(root / "out" / "r2"), cells, cfg, []), k0, "deterministic") + idx = RF.world_near(w, cells) + par = np.unique([h3.cell_to_parent(int(c), 2) for c in cells]) + self.assertTrue(np.isin(np.searchsorted(w.ids, par.astype(np.uint64)), idx).all(), "the cells under it") + near = RF.WorldCells(root / "out" / "r2") + near.a["T_mean"] = near.a["T_mean"].copy() + near.a["T_mean"][idx[0]] += 1.0 + self.assertNotEqual(RF.area_key(near, cells, cfg, []), k0) + away = RF.WorldCells(root / "out" / "r2") + away.a["T_mean"] = away.a["T_mean"].copy() + away.a["T_mean"][np.setdiff1d(np.arange(len(away.ids)), idx)[0]] += 1.0 + self.assertEqual(RF.area_key(away, cells, cfg, []), k0, "values far away don't matter") + cfg2 = copy.deepcopy(cfg) + cfg2["build"]["seed"] = int(cfg2["build"]["seed"]) + 1 + self.assertNotEqual(RF.area_key(w, cells, cfg2, []), k0) + + def test_an_unbuilt_era_waits_with_a_message(self): + root = built_world() + p = root / "config" / "eras.toml" + p.write_text(ERA_CUT.format(lat=0.0, lon=0.0)) + try: + tmp = Path(tempfile.mkdtemp()) + w = RF.WorldCells(root / "out" / "r2") + i = big_river(w.a) + reg = write_regions(tmp / "regions.json", square(float(w.a["g_lat"][i]), float(w.a["g_lon"][i]), 2)) + logs = [] + metas = RF.build_areas(root, 2, reg, log=logs.append, regions_root=tmp / "res") + self.assertEqual({m["era"] for m in metas}, {"before"}) + self.assertTrue(any("era after is not built" in s for s in logs), logs) + finally: + p.unlink() + + def test_refined_cells_keep_the_eras_air(self): + from mapgen import environment as EN, eras as ER, fields as FL + root, _, cut, _ = era_world() + tmp = Path(tempfile.mkdtemp()) + reg = write_regions(tmp / "regions.json", square(*cut, 2, "cut")) + metas = RF.build_areas(root, 2, reg, log=lambda s: None, regions_root=tmp / "res") + (m,) = [m for m in metas if m["era"] == "after"] + w = RF.WorldCells(ER.era_dir(root, 2, "after")) + with np.load(Path(m["dir"]) / "cells.npz") as z: + r = {k: z[k] for k in z.files} + p = r["parent"] + np.testing.assert_array_equal(r["gravity_g"], w.a["gravity_g"][p]) + self.assertTrue(np.any(np.isclose(r["fire_reactivity"], 0.4)), "the zone reaches the refined cells") + cfg = C.load(root)[0] + H = float(cfg["planet"]["scale_height_km"]) * 1000.0 + sea_p = r["pressure_bar"] / FL.pressure(1.0, r["z_surface_m"], H) + wet = np.sqrt(sea_p / float(cfg["planet"]["sea_level_pressure_bar"])) + zone, _ = EN.holdridge(r["biotemp"], r["P_ann"] * wet, r["T_min"]) + inner = ~r["halo"] + np.testing.assert_array_equal(r["holdridge"][inner], zone[inner]) + + +class LayoutTest(unittest.TestCase): + def setUp(self): + self.root = built_world() + w = RF.WorldCells(self.root / "out" / "r2") + i = big_river(w.a) + self.at = (float(w.a["g_lat"][i]), float(w.a["g_lon"][i])) + self.sq = square(*self.at, 2) + self.tmp = Path(tempfile.mkdtemp()) + self.reg = write_regions(self.tmp / "regions.json", self.sq) + self.res = self.tmp / "res" + + def fingerprint(self): + return T.TileSource(serve.World(self.root, 2), seed=int(C.load(self.root)[0]["build"]["seed"]), + regions_dir=self.res, with_regions=False).fingerprint + + def test_old_layout_is_replaced(self): + h = RF.area_hash(RF.area_cells([self.sq], 4)[0]) + old = RF.results_dir(self.res, self.fingerprint()) / h + old.mkdir(parents=True) + (old / "cells.npz").write_bytes(b"old") + (self.res / "0123456789-m5" / "x").mkdir(parents=True) + (m,) = RF.build_areas(self.root, 2, self.reg, log=lambda s: None, regions_root=self.res) + link = Path(m["dir"]) + self.assertTrue(link.is_symlink()) + with np.load(link / "cells.npz") as z: + self.assertIn("g_ids", z.files) + self.assertFalse((self.res / "0123456789-m5").exists()) + + def test_a_dangling_link_is_rebuilt(self): + import shutil + (m,) = RF.build_areas(self.root, 2, self.reg, log=lambda s: None, regions_root=self.res) + shutil.rmtree(Path(m["dir"]).resolve()) + (m2,) = RF.build_areas(self.root, 2, self.reg, log=lambda s: None, regions_root=self.res) + self.assertTrue(m2["built"]) + self.assertTrue((Path(m2["dir"]) / "cells.npz").exists()) + + def test_a_new_model_rebuilds_the_area(self): + from unittest import mock + (m,) = RF.build_areas(self.root, 2, self.reg, log=lambda s: None, regions_root=self.res) + with mock.patch.object(RF, "MODEL", RF.MODEL + 1): + (m2,) = RF.build_areas(self.root, 2, self.reg, log=lambda s: None, regions_root=self.res) + self.assertNotEqual(m2["key"], m["key"]) + self.assertTrue(m2["built"]) + self.assertEqual(m2["model"], RF.MODEL + 1) + + def test_unused_areas_and_links_go(self): + (m,) = RF.build_areas(self.root, 2, self.reg, log=lambda s: None, regions_root=self.res) + write_regions(self.reg, square(self.at[0] + 5.0, self.at[1] + 5.0, 2)) # redrawn elsewhere + (m2,) = RF.build_areas(self.root, 2, self.reg, log=lambda s: None, regions_root=self.res) + self.assertFalse(Path(m["dir"]).is_symlink() or Path(m["dir"]).exists()) + self.assertEqual([p.name for p in RF.areas_dir(self.res).iterdir() if p.is_dir()], [m2["key"]]) + + +class SeaFloorRefineTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + from unittest import mock + import seafloor as SF + cls.root = built_world() + w = RF.WorldCells(cls.root / "out" / "r2") + a = w.a + river, ocean = a["river"].astype(bool), a["ocean"].astype(bool) + mouth = river & ocean[a["recv"]] & ~ocean + i = int(np.argmax(np.where(mouth, a["discharge_km3_yr"], -1.0))) + cls.at = (float(a["g_lat"][i]), float(a["g_lon"][i])) + (cells,) = RF.area_cells([square(*cls.at, 3)], 4) + cls.g, cls.halo = RF.subgrid(cells, 12742.0) + src = T.TileSource(serve.World(cls.root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp())) + cfg = C.load(cls.root)[0] + cls.r = RF.refine_area(cls.g, cls.halo, w, src, cfg) + real = SF.run + + def untouched(g, z, *args): # the same pass, heights left as they were + _, f, v = real(g, z, *args) + return np.asarray(z, dtype=np.float64), f, v + with mock.patch.object(SF, "run", side_effect=untouched): + cls.r0 = RF.refine_area(cls.g, cls.halo, w, src, cfg) + + def test_the_pass_changes_only_the_sea(self): + r, r0 = self.r, self.r0 + np.testing.assert_array_equal(r["ocean"], r0["ocean"]) + land = ~r0["ocean"] + np.testing.assert_array_equal(r["elevation_eroded_m"][land], r0["elevation_eroded_m"][land]) + np.testing.assert_array_equal(r["river"], r0["river"]) + np.testing.assert_array_equal(r["elevation_eroded_m"][self.halo], r0["elevation_eroded_m"][self.halo]) + sea = r["ocean"] & ~self.halo + self.assertTrue(sea.any(), "the square reaches the sea") + self.assertTrue(np.all(r["elevation_eroded_m"][sea] <= np.maximum(r0["elevation_eroded_m"][sea], -5.0))) + + def test_vents_and_sea_fields_only_in_the_sea(self): + r = self.r + cells = np.searchsorted(self.g.ids, r["vents"]["cell"]) + self.assertTrue(np.all(r["ocean"][cells])) + land = ~r["ocean"] + self.assertTrue(np.all(r["seabed_type"][land] == 0) and np.all(r["vent"][land] == 0)) + for k in ("seabed_type", "seabed_mineral", "bottom_temp_c", "sediment_m", "vent", "canyon_m", "fan_m", + "plateau_id"): + self.assertEqual(len(r[k]), self.g.n, k) + + def test_build_saves_the_vents(self): + tmp = Path(tempfile.mkdtemp()) + reg = write_regions(tmp / "regions.json", square(*self.at, 3, "coast")) + (m,) = RF.build_areas(self.root, 2, reg, log=lambda s: None, regions_root=tmp / "res") + self.assertEqual(m["model"], RF.MODEL) + self.assertIn("vents", m) + self.assertIn("canyon_max_m", m) + with np.load(Path(m["dir"]) / "vents.npz") as z: + self.assertEqual(set(z.files), {"cell", "lat", "lon", "type", "temp_c", "flow", "mineral"}) + self.assertEqual(len(z["cell"]), m["vents"]) + + def test_preview_of_a_refined_region(self): + import seafloor as SF + tmp = Path(tempfile.mkdtemp()) + reg = write_regions(tmp / "regions.json", square(*self.at, 3, "coast")) + RF.build_areas(self.root, 2, reg, log=lambda s: None, regions_root=tmp / "res") + d = SF.preview_region(self.root, 2, "coast", regions_path=reg, regions_root=tmp / "res", out_dir=tmp / "prev") + self.assertGreater((d / "coast-base.jpg").stat().st_size, 1000) + with self.assertRaises(SystemExit): + SF.preview_region(self.root, 2, "nowhere", regions_path=reg, regions_root=tmp / "res", out_dir=tmp / "prev") + + +class PlateauOutlineTest(unittest.TestCase): + def test_outline_holds_the_plateau_and_its_margin(self): + from mapgen import plateaus as PL + from mapgen.sphere import gc_dist_km, latlon_to_xyz + from tests.helpers import small_grid + p = {"name": "east-flank", "center": [-39.9, -18.0], "area_km2": 3.0e6, "elongation": 1.9, + "azimuth_deg": 14.6, "top_m": [1500.0, 3000.0]} + o = RF.plateau_outline(p, 7, 12742.0) + self.assertEqual(len(o), 96) + self.assertFalse(serve.RegionStore._crosses([(lo, la) for la, lo in o]), "a simple polygon") + cells = np.array(sorted(h3.polygon_to_cells(h3.LatLngPoly([tuple(q) for q in o]), 3)), dtype=np.uint64) + g = small_grid(3) + pid = PL.cell_ids(g.xyz, [p], 7, 12742.0) + self.assertGreater(float(np.isin(g.ids[pid == 0], cells).mean()), 0.99) + q = np.array(o) + d = gc_dist_km(latlon_to_xyz(q[:, 0], q[:, 1]), latlon_to_xyz(*p["center"]), 12742.0) + self.assertTrue(np.all(d >= 300.0)) + + +class FastPathsTest(unittest.TestCase): + """Speed-ups with the same results as the per-cell h3 calls and loops they replaced.""" + + def test_cell_parents_match_h3(self): + rng = np.random.default_rng(1) + base = [c for r0 in h3.get_res0_cells() for c in h3.cell_to_children(r0, 1)] + for res_child, res_parent in ((7, 5), (4, 2), (5, 5), (3, 0)): + cells = np.array([h3.cell_to_center_child(int(c), res_child) for c in rng.choice(base, 50)] + + [h3.cell_to_children(int(c), res_child)[-1] for c in rng.choice(base, 50)], np.uint64) + want = [h3.cell_to_parent(int(c), res_parent) for c in cells] + np.testing.assert_array_equal(RF.cell_parents(cells, res_parent), np.array(want, np.uint64)) + with self.assertRaises(ValueError): + RF.cell_parents(np.array([h3.cell_to_center_child(int(base[0]), 3)], np.uint64), 4) + + def test_k_mult_matches_the_lookup(self): + age = np.array([0, 1, 2, 3, 4, 5, 6, 7, 8, -1, 3.9, 9], dtype=np.float64) + want = np.vectorize(lambda c: RF.K_MULT.get(int(c), 1.0))(age).astype(np.float64) + np.testing.assert_array_equal(RF.k_mult(age), want) + np.testing.assert_array_equal(RF.k_mult(age.astype(np.int8)[:9]), want[:9]) + + def test_subgrid_in_parallel_is_the_serial_grid(self): + from unittest import mock + import h3par + (cells,) = RF.area_cells([square(10, 20, 4)], 4) + g1, h1 = RF.subgrid(cells, 12742.0) + with mock.patch.object(h3par, "MIN_PARALLEL", 1), mock.patch.dict("os.environ", {"MAPVIEW_H3_WORKERS": "3"}): + g2, h2 = RF.subgrid(cells[::-1].copy(), 12742.0) # unsorted input too + for k in ("ids", "lat", "lon", "xyz", "area_km2", "nbr_ptr", "nbr_idx"): + np.testing.assert_array_equal(getattr(g1, k), getattr(g2, k)) + np.testing.assert_array_equal(h1, h2) diff --git a/tests/test_rivers.py b/tests/test_rivers.py new file mode 100644 index 0000000..2fb0d42 --- /dev/null +++ b/tests/test_rivers.py @@ -0,0 +1,266 @@ +import json +import unittest + +import numpy as np +from scipy.spatial import cKDTree + +import rivers as RV +from tests.test_serve import built_world + + +def net_and_arrays(): + root = built_world() + d = dict(np.load(root / "out" / "r2" / "cells.npz")) + meta = json.loads((root / "out" / "r2" / "cells_meta.json").read_text()) + return RV.RiverNet(d, meta["legends"], meta["radius_km"]), d + + +class RiverNetTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.net, cls.a = net_and_arrays() + cls.r = np.where(cls.a["river"].astype(bool))[0] + + def test_levels_never_above_ground_or_below_the_cap(self): + z, lev = self.a["z_surface_m"][self.r], self.net.level[self.r] + self.assertTrue(np.all(np.isfinite(lev))) + self.assertTrue(np.all(lev <= z + 1e-6)) + self.assertTrue(np.all(z - lev <= self.net.cap[self.r] + 1e-6)) + + def test_levels_never_rise_downstream(self): + recv, riv = self.a["recv"], self.a["river"].astype(bool) + down = recv[self.r] + on = riv[down] + self.assertTrue(np.all(self.net.level[down[on]] <= self.net.level[self.r[on]] + 1e-6)) + + def test_levels_meet_their_mouth(self): + recv, ocean, lake = self.a["recv"], self.a["ocean"].astype(bool), self.a["lake"].astype(bool) + sea = self.r[ocean[recv[self.r]]] + self.assertTrue(len(sea) > 0) + k = np.isin(self.net.seg_a, sea) + np.testing.assert_allclose(self.net.level_b[k], 0.0) + lk = self.r[lake[recv[self.r]]] + k = np.isin(self.net.seg_a, lk) + np.testing.assert_allclose(self.net.level_b[k], self.a["z_filled_m"][recv[self.net.seg_a[k]]]) + + def test_drops_below_lakes_are_cataracts_not_cliffs(self): + recv, riv, lake = self.a["recv"], self.a["river"].astype(bool), self.a["lake"].astype(bool) + src = np.where(lake & riv[recv] & (recv != np.arange(len(recv))))[0] + for i in src: + j = recv[i] + dist = np.linalg.norm(self.a["g_xyz"][i] - self.a["g_xyz"][j]) * self.net.R + drop = self.a["z_filled_m"][i] - self.net.level[j] + self.assertLessEqual(drop, RV.KNICK_M_KM * dist + 1e-6 + max(0.0, self.a["z_filled_m"][i] - self.a["z_surface_m"][j])) + + def test_geometry_per_segment(self): + n = len(self.net.seg_a) # river cells + lake outlets + self.assertGreaterEqual(n, len(self.r)) + for k in ("a_xyz", "b_xyz", "level_a", "level_b", "half_w", "floor", "wall", "meander", "reach", "seg_len"): + self.assertEqual(len(getattr(self.net, k)), n, k) + self.assertTrue(np.all(self.net.half_w >= 0.03) and np.all(self.net.wall > 0)) + + +class StyleTest(unittest.TestCase): + def test_hard_dry_rock_makes_steeper_narrower_valleys(self): + g = RV.valley_style("plateau", "—", "granite/gneiss", 400.0) + s = RV.valley_style("plateau", "—", "sandstone/shale", 1800.0) + self.assertGreater(g[1], 3 * s[1]) # walls + self.assertLess(g[0], s[0]) # floor + self.assertGreater(g[1], 2500) # the granite gorge: sheer + + def test_floodplains_are_wide_and_meander(self): + f = RV.valley_style("plain", "floodplain", "sandstone/shale", 900.0) + m = RV.valley_style("mountains", "—", "granite/gneiss", 900.0) + self.assertGreater(f[0], 10 * m[0]) + self.assertGreater(f[2], 0) + self.assertEqual(m[2], 0) + + def test_channel_width(self): + self.assertAlmostEqual(RV.half_width_km(2.0), 0.004 * np.sqrt(2.0 * 31.7), places=9) + + +class ValleyTest(unittest.TestCase): + def test_profile(self): + d = np.array([0.0, 0.05, 0.2, 1.0, 2.0]) + v = RV.valley_surface(d, 100.0, 0.1, 0.3, 2000.0) + self.assertEqual(v[0], 100.0) # channel: the water level + self.assertEqual(v[1], 100.0) + self.assertTrue(101.0 <= v[2] <= 103.0) # floor + self.assertAlmostEqual(v[4] - v[3], 2000.0, places=6) # wall: 2,000 m per km + + def test_points_are_a_global_lattice(self): + net, _ = net_and_arrays() + s = int(np.argmax(net.seg_len)) + c = net.a_xyz[s] + p1, _ = net.segment_points(s, c, 50.0, 5.0) + p2, _ = net.segment_points(s, c, 400.0, 5.0) + self.assertGreater(len(p2), len(p1)) + a = {tuple(np.round(v, 12)) for v in p1} + b = {tuple(np.round(v, 12)) for v in p2} + self.assertTrue(a <= b, "a wider query returns a superset of the same points") + + def test_antimeridian_segment(self): + net, _ = net_and_arrays() + a = np.array([np.cos(np.radians(10)) * np.cos(np.radians(179.9)), np.cos(np.radians(10)) * np.sin(np.radians(179.9)), np.sin(np.radians(10))]) + b = np.array([np.cos(np.radians(10)) * np.cos(np.radians(-179.9)), np.cos(np.radians(10)) * np.sin(np.radians(-179.9)), np.sin(np.radians(10))]) + pts = RV.densify(a, b, 0.0, 1.0, 1.0 / net.R) # 1 km steps (a chord in planet radii) + self.assertTrue(np.all(np.abs(np.linalg.norm(pts, axis=1) - 1) < 1e-12)) + self.assertLess(np.max(np.linalg.norm(np.diff(pts, axis=0), axis=1)) * net.R, 1.01) + + +class RoughWallTest(unittest.TestCase): + def test_walls_keep_the_terrain_roughness_but_the_channel_stays_flat(self): + d = np.array([0.0, 0.3, 1.0, 2.0]) + smooth = RV.valley_surface(d, 100.0, 0.1, 0.1, 500.0) + rough = RV.valley_surface(d, 100.0, 0.1, 0.1, 500.0, rough=np.array([80.0, 80.0, 80.0, -40.0])) + self.assertEqual(rough[0], smooth[0]) # the water level + self.assertAlmostEqual(rough[2] - smooth[2], 80.0) # a wall well away from the floor: full roughness + self.assertAlmostEqual(rough[3] - smooth[3], -40.0) + self.assertTrue(0 < rough[1] - smooth[1] < 80.0) # fading in above the floor + + +LEG = {"landform": ["ocean", "plain", "hills"], "ground": ["—", "floodplain"], "lithology": ["granite/gneiss", "sandstone/shale"], + "age_class": ["oceanic", "craton (3g era)"]} + + +def synthetic(cells): + """A tiny world. cells: (lat, lon, z, discharge, landform, ground, receiver index, kind: river|ocean|lake|land).""" + ll = np.radians([(c[0], c[1]) for c in cells]) + a = {"g_xyz": np.stack([np.cos(ll[:, 0]) * np.cos(ll[:, 1]), np.cos(ll[:, 0]) * np.sin(ll[:, 1]), np.sin(ll[:, 0])], 1), + "z_surface_m": np.array([c[2] for c in cells], float), "z_filled_m": np.array([c[2] for c in cells], float), + "discharge_km3_yr": np.array([c[3] for c in cells], float), + "landform": np.array([LEG["landform"].index(c[4]) for c in cells]), "ground": np.array([LEG["ground"].index(c[5]) for c in cells]), + "recv": np.array([c[6] for c in cells]), "river": np.array([c[7] == "river" for c in cells]), + "ocean": np.array([c[7] == "ocean" for c in cells]), "lake": np.array([c[7] == "lake" for c in cells]), + "lithology": np.ones(len(cells), int), "age_class": np.ones(len(cells), int), "P_ann": np.full(len(cells), 900.0)} + return RV.RiverNet(a, LEG, 12742.0) + + +def xyz(lat, lon): + la, lo = np.radians(lat), np.radians(lon) + return np.array([[np.cos(la) * np.cos(lo), np.cos(la) * np.sin(lo), np.sin(la)]]) + + +class PerValleyTest(unittest.TestCase): + # a big floodplain river along the equator (east to the sea) and a small hill stream 22 km north joining it + net = synthetic([(0, 0.0, 100, 5000, "plain", "floodplain", 1, "river"), (0, 0.3, 90, 5000, "plain", "floodplain", 2, "river"), + (0, 0.6, 80, 5000, "plain", "floodplain", 3, "river"), (0, 0.9, 70, 5000, "plain", "floodplain", 4, "river"), + (0, 1.2, -50, 0, "ocean", "—", 4, "ocean"), + (0.1, 0.3, 400, 2, "hills", "—", 6, "river"), (0.1, 0.6, 380, 2, "hills", "—", 2, "river")]) + + def test_a_big_valley_is_not_crowded_out_by_a_nearby_stream(self): + q = xyz(0.05, 0.45) # 11 km from both centrelines + V, ch, dr = self.net.valleys(q, 0.5) + big = float(np.interp(0.45, [0.3, 0.6], [self.net.level[1], self.net.level[2]])) + self.assertLessEqual(V[0], big + 3.0 + 1e-6, "on the big river's floodplain floor") + + def test_the_answer_does_not_depend_on_the_other_query_points(self): + q = xyz(0.05, 0.45) + many = np.vstack([q, xyz(3.0, 20.0), xyz(-0.2, 0.1), xyz(0.08, 0.5)]) + a, b = self.net.valleys(q, 0.5)[0], self.net.valleys(many, 0.5)[0] + self.assertEqual(a[0], b[0]) + + def test_floors_stay_inside_the_reach(self): + self.assertTrue(np.all(self.net.floor <= RV.FLOOR_MAX_KM + 1e-9)) + self.assertTrue(np.all(self.net.reach >= self.net.half_w + self.net.floor)) + + def test_rough_walls_never_dip_below_the_water(self): + v = RV.valley_surface(np.array([0.6, 1.0]), 100.0, 0.1, 0.3, 500.0, rough=np.array([-500.0, -500.0])) + self.assertTrue(np.all(v >= 101.0)) + + +class BankTest(unittest.TestCase): + def test_a_river_doubling_back_below_itself_keeps_its_banks_above_its_water(self): + # a canyon river runs east, then back west 5.5 km north and 400 m lower: its lower floor must not + # undercut the upper reach (dry pits below the water beside it) + net = synthetic([(0, 0.0, 900, 300, "hills", "—", 1, "river"), (0, 0.3, 700, 300, "hills", "—", 2, "river"), + (0.05, 0.0, 300, 300, "hills", "—", 3, "river"), (0.05, -0.3, -50, 0, "ocean", "—", 3, "ocean")]) + la, lo = np.meshgrid(np.linspace(-0.03, 0.08, 40), np.linspace(-0.05, 0.35, 80), indexing="ij") + q = np.vstack([xyz(a, b) for a, b in zip(la.ravel(), lo.ravel())]) + V, ch, _ = net.valleys(q, 0.1) + best, lev, bank = np.full(len(q), np.inf), np.zeros(len(q)), np.zeros(len(q), bool) + c = q.mean(0) / np.linalg.norm(q.mean(0)) + for s in range(len(net.seg_a)): # the nearest reach, its water, its floor zone + p, t = net.segment_points(s, c, 60.0, 0.05) + d, k = cKDTree(p).query(q) + d *= net.R + b = d < best + best[b], bank[b] = d[b], (d <= net.half_w[s] + net.floor[s])[b] + lev[b] = (net.level_a[s] + (net.level_b[s] - net.level_a[s]) * t[k])[b] + dry = ~ch & bank + self.assertGreater(dry.sum(), 100) + self.assertEqual(int((V[dry] < lev[dry]).sum()), 0, "floor below the water of the reach beside it") + + +class LakeOutletTest(unittest.TestCase): + def test_a_lake_outlet_joins_its_river_starting_at_the_lake_surface(self): + net = synthetic([(0, 0.0, 1100, 0, "hills", "—", 1, "lake"), (0, 0.3, 1100, 300, "plain", "—", 2, "river"), + (0, 0.6, 600, 300, "plain", "—", 3, "river"), (0, 0.9, -50, 0, "ocean", "—", 3, "ocean")]) + k = np.where(net.seg_a == 0)[0] + self.assertEqual(len(k), 1, "the lake outlet has a segment") + self.assertEqual(net.level_a[k[0]], 1100.0) + self.assertEqual(net.seg_b[k[0]], 1) + + +class GroundTest(unittest.TestCase): + def test_given_levels_reach_their_shoulders_and_sub_pixel_streams_are_not_drawn(self): + cells = [(0, 0.0, 800, 5000, "plateau", "—", 1, "river"), (0, 0.3, 700, 5000, "plateau", "—", 2, "river"), + (0, 0.6, -50, 0, "ocean", "—", 2, "ocean")] + levels = np.array([400.0, 300.0, 0.0]) + ground = np.array([800.0, 700.0, 0.0]) + LEG2 = dict(LEG, landform=LEG["landform"] + ["plateau"]) + ll = np.radians([(c[0], c[1]) for c in cells]) + arr = {"g_xyz": np.stack([np.cos(ll[:, 0]) * np.cos(ll[:, 1]), np.cos(ll[:, 0]) * np.sin(ll[:, 1]), np.sin(ll[:, 0])], 1), + "z_surface_m": ground.copy(), "z_filled_m": ground.copy(), "discharge_km3_yr": np.array([5000.0, 5000.0, 0.0]), + "landform": np.array([3, 3, 0]), "ground": np.zeros(3, int), "recv": np.array([1, 2, 2]), + "river": np.array([True, True, False]), "ocean": np.array([False, False, True]), "lake": np.zeros(3, bool), + "lithology": np.zeros(3, int), "age_class": np.ones(3, int), "P_ann": np.full(3, 300.0)} + net = RV.RiverNet(arr, LEG2, 12742.0, levels=levels, ground=ground) + k = 0 + self.assertGreaterEqual(net.reach[k], net.half_w[k] + net.floor[k] + (ground[0] - levels[0]) / net.wall[k] - 1e-9) + small = synthetic([(0, 0.0, 100, 0.2, "plain", "—", 1, "river"), (0, 0.3, 90, 0.2, "plain", "—", 2, "river"), + (0, 0.6, -50, 0, "ocean", "—", 2, "ocean")]) + q = xyz(0.0, 0.15) + self.assertFalse(small.channel(q, 5.0)[1][0], "a 0.2 km³/yr stream is not drawn at 5 km per pixel") + self.assertTrue(small.channel(q, 0.01)[1][0], "but is at street level") + + +class ChunkTest(unittest.TestCase): + def test_a_coarse_tile_splits_into_compact_chunks(self): + net = synthetic([(0, 0.0, 100, 5, "plain", "—", 1, "river"), (0, 0.3, 90, 5, "plain", "—", 2, "river"), + (0, 0.6, -50, 0, "ocean", "—", 2, "ocean")]) + la, lo = np.meshgrid(np.linspace(-52.0, -46.4, 256), np.linspace(-45.0, -33.75, 256), indexing="ij") # a z5 tile + a, o = np.radians(la.ravel()), np.radians(lo.ravel()) + chunks = net._chunks(np.stack([np.cos(a) * np.cos(o), np.cos(a) * np.sin(o), np.sin(a)], 1)) + self.assertLess(len(chunks), 300, "not thousands of one-row strips") + self.assertEqual(sorted(np.concatenate([c[0] for c in chunks]).tolist()), list(range(256 * 256))) + + +class RiverNetStateTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + import refine as RF + from tests.test_serve import built_world + w = RF.WorldCells(built_world() / "out" / "r2") + cls.a = {k: np.asarray(w.a[k]) for k in RF.NET_KEYS} + cls.legends, cls.R = w.meta["legends"], float(w.meta["radius_km"]) + + def test_state_round_trip_answers_the_same(self): + import rivers as RV + net = RV.RiverNet(self.a, self.legends, self.R) + arrays, meta = net.state() + self.assertEqual(set(arrays), set(RV.STATE)) + back = RV.RiverNet.from_state(arrays, meta) + for k in RV.STATE: + np.testing.assert_array_equal(getattr(back, k), getattr(net, k)) + self.assertEqual(back.max_extent, net.max_extent) + p = np.asarray(self.a["g_xyz"][:200], dtype=np.float64) + np.testing.assert_array_equal(back.valleys(p, 5.0)[1], net.valleys(p, 5.0)[1]) + np.testing.assert_array_equal(back.channel(p, 5.0)[1], net.channel(p, 5.0)[1]) + + def test_tree_is_built_on_first_use(self): + import rivers as RV + net = RV.RiverNet.from_state(*RV.RiverNet(self.a, self.legends, self.R).state()) + self.assertIsNone(net._tree) + self.assertIsNotNone(net.tree) + self.assertIs(net.tree, net.tree) diff --git a/tests/test_seafloor.py b/tests/test_seafloor.py new file mode 100644 index 0000000..fed467d --- /dev/null +++ b/tests/test_seafloor.py @@ -0,0 +1,195 @@ +import unittest + +import numpy as np + +import refine as RF +import seafloor as SF +from mapgen.graph import priority_flood +from mapgen.sphere import gc_dist_km, latlon_to_xyz +from tests.test_refine import square + +R = 12742.0 + + +def grid(lat=0.0, lon=0.0, d=4.0): + (cells,) = RF.area_cells([square(lat, lon, d)], 5) + return RF.subgrid(cells, R) + + +def shelf(): + """A res-5 square across a coast: land west of 3° W, a shelf, a slope with gullies, an abyssal plain with a deep + hollow (≈ 300 m, at 0°, 2.5° E) and a shallow one (≈ 30 m, at 2.5° S, 1.5° E).""" + g, halo = grid() + x = g.lon + z = np.where(x < -3.0, 200.0, + np.where(x < -1.5, -100.0 - 100.0 * (x + 3.0) / 1.5, + np.where(x < 0.0, -200.0 - 3800.0 * (x + 1.5) / 1.5, -4000.0 - 75.0 * x))) + z = z + np.where((x >= -1.5) & (x < 0.0), 40.0 * np.sin(np.radians(g.lat) * 400.0), 0.0) # gullies + for la, lo, r_km, depth in ((0.0, 2.5, 60.0, 300.0), (-2.5, 1.5, 40.0, 30.0)): + d = gc_dist_km(g.xyz, latlon_to_xyz(la, lo), R) + z = z - depth * np.clip(1.0 - (d / r_km) ** 2, 0.0, 1.0) + return g, halo, z, z <= 0 + + +class CanyonTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.g, cls.halo, cls.z, cls.sea = shelf() + cls.zc, cls.cut, cls.fan = SF.canyons(cls.g, cls.z, cls.sea, cls.halo) + + def test_canyons_cut_the_slope(self): + self.assertGreater(self.cut.max(), 50.0) + self.assertLessEqual(self.cut.max(), SF.CANYON["cap_m"]) + slope = (self.g.lon > -1.5) & (self.g.lon < 0.0) + self.assertGreater(self.cut[slope].max(), self.cut[~slope].max(initial=0.0)) + + def test_only_lower_in_canyons_and_raise_at_most_100_m_on_fans(self): + dz = self.zc - self.z + self.assertGreaterEqual(dz.min(), -SF.CANYON["cap_m"] - 1e-9) + self.assertLessEqual(dz.max(), SF.FAN["max_m"] + 1e-9) + np.testing.assert_array_equal(dz < 0, self.cut > 0) + self.assertTrue(np.all(self.fan[dz > 0] > 0)) + self.assertGreater(self.fan.max(), 0.0, "fans where the flows stop") + + def test_land_and_halo_stay(self): + keep = ~self.sea | self.halo + np.testing.assert_array_equal(self.zc[keep], self.z[keep]) + self.assertTrue(np.all(self.zc[self.sea] <= np.maximum(self.z[self.sea], SF.SEA_TOP_M))) + + def test_deterministic(self): + again = SF.canyons(self.g, self.z, self.sea, self.halo) + for a, b in zip(again, (self.zc, self.cut, self.fan)): + np.testing.assert_array_equal(a, b) + + +class PondTest(unittest.TestCase): + def setUp(self): + self.g, self.halo, self.z, self.sea = shelf() + self.sed = np.full(self.g.n, 500.0) + + def test_deep_hollows_fill_flat_below_their_spill(self): + zp, fill = SF.ponds(self.g, self.z, self.sea, self.halo, self.sed) + zf = priority_flood(self.g, np.where(self.sea, self.z, SF.WALL_M), self.sea & self.halo, eps=0.0) + self.assertTrue(np.all(zp <= np.maximum(zf, self.z) + 1e-9), "never above the spill level") + self.assertTrue(np.all(zp >= self.z)) + np.testing.assert_allclose(fill, zp - self.z) + deep = gc_dist_km(self.g.xyz, latlon_to_xyz(0.0, 2.5), R) < 30.0 + self.assertTrue(np.all(fill[deep] > 0)) + self.assertLess(float(np.ptp(zp[deep])), 1e-6, "a flat floor") + self.assertAlmostEqual(float(zp[deep][0]), float(self.z[deep].min()) + 0.3 * 500.0, delta=1.0) + shallow = gc_dist_km(self.g.xyz, latlon_to_xyz(-2.5, 1.5), R) < 20.0 + self.assertTrue(np.all(fill[shallow] == 0), "hollows under 50 m stay as they are") + + def test_no_sea_is_a_no_op(self): + z = np.full(self.g.n, 300.0) + sea = np.zeros(self.g.n, bool) + zc, cut, fan = SF.canyons(self.g, z, sea, self.halo) + zp, fill = SF.ponds(self.g, z, sea, self.halo, self.sed) + np.testing.assert_array_equal(zc, z) + np.testing.assert_array_equal(zp, z) + self.assertFalse(cut.any() or fan.any() or fill.any()) + + def test_sea_without_halo_sea_still_ponds(self): + d = gc_dist_km(self.g.xyz, latlon_to_xyz(0.0, 0.0), R) + z = np.where(d < 200.0, -1000.0 - 300.0 * np.clip(1.0 - (d / 100.0) ** 2, 0.0, 1.0), 500.0) + sea = z <= 0 + self.assertFalse((sea & self.halo).any()) + zp, fill = SF.ponds(self.g, z, sea, self.halo, self.sed) + zc, _, _ = SF.canyons(self.g, z, sea, self.halo) + self.assertTrue(np.all(np.isfinite(zp)) and np.all(np.isfinite(zc))) + + +from mapgen import plateaus as PL, seabed as SB + +PLAT = {"name": "p", "center": [0.0, 0.0], "area_km2": 2.0e5, "top_m": [1500.0, 2500.0]} + + +def world_values(n=1): + return {"vent_potential": np.full(n, 0.8), "seabed_type": np.full(n, SB.SB_CONTINENTAL, np.int8), + "seabed_mineral": np.zeros(n, np.int8), "bottom_temp_c": np.full(n, 3.0), + "sediment_m": np.full(n, 400.0), "d_div_km": np.full(n, 900.0), "T_mean": np.full(n, 10.0)} + + +class CellDrawTest(unittest.TestCase): + def test_cell_draws_are_uniform_and_stable(self): + ids = np.arange(10 ** 5, dtype=np.uint64) * np.uint64(7919) + np.uint64(612345678901) + u = SF._u(ids, 3) + self.assertTrue(0.0 <= u.min() and u.max() < 1.0) + self.assertAlmostEqual(float(u.mean()), 0.5, delta=0.01) + np.testing.assert_array_equal(SF._u(ids, 3), u) + self.assertFalse(np.array_equal(SF._u(ids, 4), u)) + + +class VolcanicTest(unittest.TestCase): + def test_fine_features_on_the_plateau_only(self): + g, halo = grid() + pid = PL.cell_ids(g.xyz, [PLAT], 7, R) + add, near = SF.volcanic(g, np.ones(g.n, bool), halo, pid, [PLAT], 7) + self.assertGreater(add.max(), 150.0) + self.assertTrue(np.all(add[pid < 0] == 0) and np.all(near[pid < 0] == 0)) + self.assertTrue(np.all(add[halo] == 0)) + + def test_features_and_vents_do_not_depend_on_the_area(self): + outs = [] + for g, h in (grid(0.0, 0.0, 4.0), grid(0.5, 0.5, 3.0)): + pid = PL.cell_ids(g.xyz, [PLAT], 7, R) + sea = np.ones(g.n, bool) + add, near = SF.volcanic(g, sea, h, pid, [PLAT], 7) + _, v = SF.vents(g, np.full(g.n, -2000.0) + add, sea, h, np.full(g.n, 0.8), near, np.full(g.n, 500.0), + np.full(g.n, 100.0), np.full(g.n, 2.0), 7) + outs.append((g, h, add, v)) + (ga, ha, aa, va), (gb, hb, ab, vb) = outs + ca, cb = np.flatnonzero(~ha), np.flatnonzero(~hb) + common, ia, ib = np.intersect1d(ga.ids[ca], gb.ids[cb], return_indices=True) + self.assertGreater(len(common), 100) + np.testing.assert_allclose(aa[ca[ia]], ab[cb[ib]], atol=1e-9) + np.testing.assert_array_equal(np.intersect1d(va["cell"], common), np.intersect1d(vb["cell"], common)) + + def test_hidden_plateaus_stay_500_m_down_and_no_sea_rises_to_land(self): + g, halo = grid() + z = np.full(g.n, -600.0) + sea = np.ones(g.n, bool) + zn, f, v = SF.run(g, z, sea, halo, np.zeros(g.n, np.int64), world_values(), [PLAT], 7) + on = (f["plateau_id"] >= 0) & ~halo + self.assertTrue(on.any()) + self.assertLessEqual(float(zn[on].max()), -500.0) + isl = {**PLAT, "islands": True} + zi, fi, _ = SF.run(g, z, sea, halo, np.zeros(g.n, np.int64), world_values(), [isl], 7) + self.assertLessEqual(float(zi.max()), SF.SEA_TOP_M) + np.testing.assert_array_equal(zn[halo], z[halo]) + + +class VentTest(unittest.TestCase): + def test_vents_only_in_the_sea_and_smokers_on_the_ridge(self): + g, halo = grid() + z = np.where(g.lon < -3.0, 100.0, -2500.0) + sea = z <= 0 + d_div = np.abs(g.lon - 2.0) * 111.0 # a ridge axis along 2° E + pot = np.where(sea, np.clip(1.0 - d_div / 300.0, 0.05, 1.0), 0.0) + st, v = SF.vents(g, z, sea, halo, pot, np.zeros(g.n), d_div, np.full(g.n, 1500.0), np.full(g.n, 2.0), 7) + cells = np.searchsorted(g.ids, v["cell"]) + self.assertGreater(len(cells), 0) + self.assertTrue(np.all(sea[cells] & ~halo[cells] & (z[cells] < -SF.VENT["min_depth_m"]))) + self.assertTrue(np.all(v["type"][d_div[cells] < SF.VENT["ridge_km"]] == 0), "black smokers on the ridge") + t, temp = v["type"], v["temp_c"] + self.assertTrue(np.all((temp[t == 0] >= 300) & (temp[t == 0] <= 400))) + self.assertTrue(np.all(st[~sea] == 0) and st.max() <= 1.0) + np.testing.assert_array_equal(v["mineral"], v["type"]) + + +class FieldsTest(unittest.TestCase): + def test_fields_zero_on_land_and_features_mark_the_floor(self): + pv = {"seabed_type": np.array([SB.SB_CLAY, SB.SB_NONE, SB.SB_CLAY, SB.SB_CLAY, SB.SB_NONE], np.int8), + "seabed_mineral": np.zeros(5, np.int8), "bottom_temp_c": np.array([2.0, 0.0, 2.0, 2.0, 0.0]), + "sediment_m": np.array([400.0, 0.0, 400.0, 400.0, 0.0]), "T_mean": np.array([10.0, 12.0, 10.0, 10.0, 12.0])} + sea = np.array([True, True, True, True, False]) + z0 = np.zeros(5) + f = SF.fields(pv, sea, np.zeros(5, bool), np.array([0.0, 0, 100, 0, 0]), np.array([0.0, 0, 0, 5, 0]), z0, + np.array([0.0, 0, 0, 0, 0]), np.array([0.9, 0, 0, 0, 0])) + np.testing.assert_array_equal(f["seabed_type"], [SB.SB_VENTS, SB.SB_TERRIGENOUS, SB.SB_TERRIGENOUS, + SB.SB_TERRIGENOUS, SB.SB_NONE]) + self.assertEqual(int(f["seabed_mineral"][0]), SB.MI_SULFIDES) + np.testing.assert_allclose(f["sediment_m"], [0.0, 300.0, 120.0, 405.0, 0.0]) + np.testing.assert_allclose(f["bottom_temp_c"], [20.0, 12.0, 2.0, 2.0, 0.0]) + np.testing.assert_allclose(f["vent"], [0.9, 0, 0, 0, 0], rtol=1e-6) + np.testing.assert_allclose(f["canyon_m"], [0, 0, 100, 0, 0]) diff --git a/tests/test_serve.py b/tests/test_serve.py new file mode 100644 index 0000000..372e41e --- /dev/null +++ b/tests/test_serve.py @@ -0,0 +1,1346 @@ +import contextlib +import io +import http.client +import json +import os +import socket +import shutil +import tempfile +import threading +import time +import unittest +import urllib.error +import urllib.request +from pathlib import Path +from unittest import mock + +import numpy as np + +from mapgen import pipeline as P +from mapgen.testing import built_world, small_world # noqa: F401 (other tests import built_world from here) +from tests.helpers import big_river + +import serve + + +class Running: + def __init__(self, server): + self.server = server + + def __enter__(self): + threading.Thread(target=self.server.serve_forever, daemon=True).start() + return f"http://127.0.0.1:{self.server.server_address[1]}" + + def __exit__(self, *a): + self.server.shutdown() + self.server.server_close() + + +def request(url, method="GET", body=None): + data = None if body is None else (body if isinstance(body, bytes) else json.dumps(body).encode()) + req = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req) as r: + return r.status, r.read() + except urllib.error.HTTPError as e: + with e: + return e.code, e.read() + + +class HubTest(unittest.TestCase): + def test_routes_per_world(self): + root = built_world() + with Running(serve.make_hub([("a", root, 2), ("b", root, 2)], 0)) as base: + status, body = request(base + "/api/worlds") + self.assertEqual(status, 200) + self.assertEqual([w["id"] for w in json.loads(body)["worlds"]], ["a", "b"]) + status, body = request(base + "/w/b/api/worlds") + self.assertEqual(json.loads(body)["current"], "b") + status, body = request(base + "/w/b/api/meta") + self.assertEqual(status, 200) + self.assertIn("/w/b/", body.decode()) + self.assertNotIn('"/tiles/', body.decode()) + self.assertEqual(request(base + "/api/meta")[0], 200) + self.assertEqual(request(base + "/w/nope/api/meta")[0], 404) + with urllib.request.urlopen(base + "/w/a") as r: + self.assertTrue(r.url.endswith("/w/a/")) + + +class CoreTest(unittest.TestCase): + def setUp(self): + self.root = built_world() + self.pins_dir = Path(tempfile.mkdtemp()) + + def tearDown(self): + shutil.rmtree(self.pins_dir) + + def server(self): + lore = self.pins_dir / "lore" + lore.mkdir(exist_ok=True) + (lore / "physical-world.md").write_text("---\nid: physical-world\n---\nNotes.\n") + return serve.make_server(self.root, 2, 0, pins_path=self.pins_dir / "pins.json", lore_dir=lore) + + def test_meta(self): + with Running(self.server()) as base: + status, body = request(base + "/api/meta") + meta = json.loads(body) + self.assertEqual(status, 200) + self.assertEqual(meta["res"], 2) + self.assertAlmostEqual(meta["radius_km"], 12742.0) + self.assertAlmostEqual(meta["units"]["league_km"], 15.4437) + self.assertEqual(meta["layers"][0]["id"], "relief") + self.assertIn("coast", [o["id"] for o in meta["overlays"]]) + self.assertIn("physical-world", meta["lore_ids"]) + + def test_static_and_traversal(self): + with Running(self.server()) as base: + s1, html = request(base + "/") + s2, _ = request(base + "/data/viewer/relief.jpg") + s3, _ = request(base + "/data/geo/coast.geojson") + s4, _ = request(base + "/data/viewer/../cells_meta.json") + s5, _ = request(base + "/../serve.py") + self.assertEqual((s1, s2, s3), (200, 200, 200)) + self.assertIn(b"World Maps", html) + self.assertEqual((s4, s5), (404, 404)) + + def test_pick_res_needs_viewer_outputs(self): + tmp = Path(tempfile.mkdtemp()) + try: + cfg = {"build": {"res_final": 5, "res_dev": 4}} + (tmp / "out" / "r5").mkdir(parents=True) + (tmp / "out" / "r5" / "cells.npz").write_bytes(b"") + self.assertEqual(serve.pick_res(tmp, cfg), 4) # final cells but no viewer textures yet + (tmp / "out" / "r5" / "viewer").mkdir() + (tmp / "out" / "r5" / "viewer" / "layers.json").write_text("[]") + self.assertEqual(serve.pick_res(tmp, cfg), 5) + finally: + shutil.rmtree(tmp) + + def test_main_stops_cleanly_on_sigterm(self): + import signal + + def term_then_wait(self_): + os.kill(os.getpid(), signal.SIGTERM) + time.sleep(5) # the handler interrupts this + raise AssertionError("SIGTERM did not stop the server") + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + before = signal.getsignal(signal.SIGTERM) + with mock.patch.object(serve.ThreadingHTTPServer, "serve_forever", term_then_wait), \ + contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(serve.main(["--res", "2", "--port", str(port), "--workers", "0"], root=self.root), 0) + self.assertEqual(signal.getsignal(signal.SIGTERM), before, "the handler is put back") + + def test_main_workers_option(self): + seen = [] + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + real = serve.make_server + + def spy(*a, **k): + seen.append(k.get("workers")) + return real(*a, **{**k, "workers": 0}) # no forking in the test + with mock.patch.object(serve, "make_server", spy), \ + mock.patch.object(serve.ThreadingHTTPServer, "serve_forever", side_effect=KeyboardInterrupt), \ + contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(serve.main(["--res", "2", "--port", str(port), "--workers", "3"], root=self.root), 0) + self.assertEqual(serve.main(["--res", "2", "--port", str(port)], root=self.root), 0) + self.assertEqual(seen, [3, serve.default_workers()]) + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + serve.main(["--res", "2", "--workers", "-1"], root=self.root) + + def test_main_prunes_stale_serve_caches_at_start(self): + import refine + import servecache as SC + import tiles as T + out = self.root / "out" / "r2" + stale_world = SC.cache_dir(out, "stale") + stale_world.mkdir(parents=True, exist_ok=True) + from mapgen import config as C + seed = int(C.load(self.root)[0]["build"]["seed"]) + fp = T.TileSource(serve.World(self.root, 2), seed=seed, cache_dir=Path(tempfile.mkdtemp())).fingerprint + stale_regions = SC.cache_dir(refine.results_dir(out / "regions", fp), "stale") + stale_regions.mkdir(parents=True, exist_ok=True) + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + with mock.patch.object(serve.ThreadingHTTPServer, "serve_forever", side_effect=KeyboardInterrupt), \ + contextlib.redirect_stdout(io.StringIO()): + code = serve.main(["--res", "2", "--port", str(port)], root=self.root) + self.assertEqual(code, 0) + self.assertFalse(stale_world.exists()) + self.assertFalse(stale_regions.exists()) + self.assertTrue(any(out.glob(f"{SC.PREFIX}*")), "the world's own cache stays") + + def test_main_carries_saved_tiles_over_before_rendering_ahead(self): + import tiles as T + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + calls = [] + with mock.patch.object(serve.ThreadingHTTPServer, "serve_forever", side_effect=KeyboardInterrupt), \ + mock.patch.object(T.TileSource, "carry_stale", lambda src: calls.append("carry")), \ + mock.patch.object(serve.RefineJobs, "prerender", lambda jobs: calls.append("prerender")), \ + contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(serve.main(["--res", "2", "--port", str(port)], root=self.root), 0) + self.assertEqual(calls[-1], "prerender") + self.assertIn("carry", calls) + + def test_port_in_use_is_a_clear_error(self): + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + s.listen() + port = s.getsockname()[1] + err = io.StringIO() + with contextlib.redirect_stderr(err): + code = serve.main(["--res", "2", "--port", str(port)], root=self.root) + self.assertEqual(code, 2) + self.assertIn(f"port {port} is in use", err.getvalue()) + + def test_failed_setup_frees_the_port(self): + made = [] + + class Kept(serve.ThreadingHTTPServer): # keeps the instance alive: garbage collection can't hide a leak + def __init__(self, *a, **k): + super().__init__(*a, **k) + made.append(self) + with mock.patch("serve.ThreadingHTTPServer", Kept), \ + mock.patch("tiles.TileSource", side_effect=RuntimeError("no rasters")), self.assertRaises(RuntimeError): + serve.make_server(self.root, 2, 0, pins_path=Path(tempfile.mkdtemp()) / "p.json") + self.assertEqual(made[0].socket.fileno(), -1, "socket closed") + + def test_port_in_use_reported_before_loading_the_world(self): + import time + real = serve.World + def slow(*a, **k): + time.sleep(3) + return real(*a, **k) + with socket.socket() as s, mock.patch("serve.World", side_effect=slow): + s.bind(("127.0.0.1", 0)) + s.listen() + t = time.time() + with contextlib.redirect_stderr(io.StringIO()): + code = serve.main(["--res", "2", "--port", str(s.getsockname()[1])], root=self.root) + self.assertEqual(code, 2) + self.assertLess(time.time() - t, 1.5) + + def test_missing_cell_is_404_not_a_neighbour(self): + server = self.server() + i = server.world.index_of(10.0, 20.0) + server.world.ids = np.delete(server.world.ids, i) + with self.assertRaises(LookupError): + server.world.index_of(10.0, 20.0) + with Running(server) as base: + self.assertEqual(request(f"{base}/api/cell?lat=10&lon=20")[0], 404) + + def test_missing_outputs_error(self): + empty = Path(tempfile.mkdtemp()) + try: + with self.assertRaisesRegex(serve.MissingBuild, "mapgen.py build"): + serve.make_server(empty, 2, 0) + finally: + shutil.rmtree(empty) + + +class CellTest(unittest.TestCase): + def setUp(self): + self.root = built_world() + self.cells = np.load(self.root / "out" / "r2" / "cells.npz") + + def test_cell_centre_and_inner_radius(self): + import h3.api.basic_int as h3 + with Running(serve.make_server(self.root, 2, 0, pins_path=Path(tempfile.mkdtemp()) / "p.json")) as base: + c = json.loads(request(f"{base}/api/cell?lat=12.3&lon=45.6")[1]) + lat, lon = h3.cell_to_latlng(h3.str_to_int(c["cell"])) + self.assertAlmostEqual(c["center"][0], lat, places=6) + self.assertAlmostEqual(c["center"][1], lon, places=6) + self.assertGreater(c["inner_km"], 0) + + def test_reuse_circle_stays_inside_its_own_cell(self): # the hover shortcut must never show a neighbour's data + import h3.api.basic_int as h3 + server = serve.make_server(self.root, 2, 0, pins_path=Path(tempfile.mkdtemp()) / "p.json") + server.server_close() + R, bad = 12742.0, [] + for i in range(0, len(self.cells["g_ids"]), 7): + cell = int(self.cells["g_ids"][i]) + clat, clon = h3.cell_to_latlng(cell) + d = server.world.describe(clat, clon) + la1, lo1, ang = np.radians(clat), np.radians(clon), d["inner_km"] / R + for az in np.radians(np.arange(0, 360, 15)): # 24 points on the circle + la2 = np.arcsin(np.sin(la1) * np.cos(ang) + np.cos(la1) * np.sin(ang) * np.cos(az)) + lo2 = lo1 + np.arctan2(np.sin(az) * np.sin(ang) * np.cos(la1), np.cos(ang) - np.sin(la1) * np.sin(la2)) + if h3.latlng_to_cell(np.degrees(la2), np.degrees(lo2), 2) != cell: + bad.append(h3.int_to_str(cell)) + break + self.assertEqual(bad, []) + + def test_known_cell(self): + server = serve.make_server(self.root, 2, 0, pins_path=Path(tempfile.mkdtemp()) / "pins.json") + i = 1234 + lat, lon = float(self.cells["g_lat"][i]), float(self.cells["g_lon"][i]) + with Running(server) as base: + status, body = request(f"{base}/api/cell?lat={lat}&lon={lon}") + bad, _ = request(f"{base}/api/cell?lat=95&lon=0") + badlon = json.loads(request(f"{base}/api/cell?lat=10&lon=inf")[1])["error"] + junk, _ = request(f"{base}/api/cell?lat=x") + self.assertEqual((status, bad, junk), (200, 400, 400)) + self.assertIn("lon", badlon) + cell = json.loads(body) + import h3.api.basic_int as h3 + self.assertEqual(cell["cell"], h3.int_to_str(int(self.cells["g_ids"][i]))) + f = {x["key"]: x for x in cell["fields"]} + self.assertAlmostEqual(f["z_surface_m"]["value"], float(self.cells["z_surface_m"][i]), delta=0.5) + self.assertEqual(f["z_surface_m"]["alt"], f"{float(self.cells['z_surface_m'][i]) / 0.331013:,.0f} spans") + meta = json.loads((self.root / "out" / "r2" / "cells_meta.json").read_text()) + self.assertEqual(f["holdridge"]["text"], meta["legends"]["holdridge"][int(self.cells["holdridge"][i])]) + self.assertEqual(f["plate"]["text"], meta["legends"]["plates"][int(self.cells["plate"][i])]) + self.assertIn(f["lake"]["text"], ("yes", "no")) + self.assertIn("T_summer", f) + g = float(self.cells["gravity_g"][i]) # 1.05 g = 180 (500₆) spans/moment²; low-g zones differ + self.assertEqual(f["gravity_g"]["alt"], f"{g * 9.80665 / 0.331013 * 2.403473 ** 2:,.0f} spans/moment²") + + +class PinsTest(unittest.TestCase): + def setUp(self): + self.root = built_world() + self.dir = Path(tempfile.mkdtemp()) + self.path = self.dir / "places" / "pins.json" + + def tearDown(self): + shutil.rmtree(self.dir) + + def server(self): + return serve.make_server(self.root, 2, 0, pins_path=self.path) + + def test_crud(self): + with Running(self.server()) as base: + s, body = request(base + "/api/pins", "POST", {"name": "Cell Am Zee", "lat": 12.5, "lon": 190.0, + "lore": "cell-am-zee", "note": "the University"}) + pin = json.loads(body) + self.assertEqual(s, 201) + self.assertRegex(pin["id"], r"^pin-[0-9a-f]{8}$") + self.assertAlmostEqual(pin["lon"], -170.0) + s, body = request(f"{base}/api/pins/{pin['id']}", "PUT", {"name": "The University"}) + self.assertEqual((s, json.loads(body)["name"], json.loads(body)["lore"]), (200, "The University", "cell-am-zee")) + s, body = request(base + "/api/pins") + self.assertEqual([p["name"] for p in json.loads(body)["pins"]], ["The University"]) + self.assertEqual(request(f"{base}/api/pins/{pin['id']}", "DELETE")[0], 204) + self.assertEqual(request(f"{base}/api/pins/{pin['id']}", "DELETE")[0], 404) + self.assertEqual(json.loads(self.path.read_text()), {"pins": []}) + + def test_validation(self): + with Running(self.server()) as base: + for bad in ({"name": "", "lat": 0, "lon": 0}, {"name": "x", "lat": 100, "lon": 0}, + {"name": "x", "lat": 0, "lon": 0, "lore": "Bad Id"}, + {"name": "x" * 121, "lat": 0, "lon": 0}, {"name": "x", "lat": 0, "lon": 0, "note": "n" * 2001}): + self.assertEqual(request(base + "/api/pins", "POST", bad)[0], 400, bad) + + def test_pin_file_format_and_cell_res(self): + import h3.api.basic_int as h3 + with Running(self.server()) as base: + pin = json.loads(request(base + "/api/pins", "POST", {"name": "x", "lat": 1, "lon": 2})[1]) + self.assertEqual(h3.get_resolution(h3.str_to_int(pin["cell"])), 5) # res_final, not the dev server's res 2 + self.assertTrue(self.path.read_text().endswith("\n")) + self.assertEqual(self.path.stat().st_mode & 0o777, 0o644) + + def test_epoch_and_move(self): + with Running(self.server()) as base: + pin = json.loads(request(base + "/api/pins", "POST", {"name": "x", "lat": 1, "lon": 2, "epoch": "epoch-high-fantasy"})[1]) + bad = request(base + "/api/pins", "POST", {"name": "x", "lat": 1, "lon": 2, "epoch": "High Fantasy"})[0] + moved = json.loads(request(f"{base}/api/pins/{pin['id']}", "PUT", {"lat": -15.5, "lon": -34.2})[1]) + cleared = json.loads(request(f"{base}/api/pins/{pin['id']}", "PUT", {"epoch": ""})[1]) + self.assertEqual(pin["epoch"], "epoch-high-fantasy") + self.assertEqual(bad, 400) + self.assertEqual((moved["lat"], moved["lon"], moved["epoch"]), (-15.5, -34.2, "epoch-high-fantasy")) + self.assertNotEqual(moved["cell"], pin["cell"]) + self.assertIsNone(cleared["epoch"]) + + def test_validation_types(self): + with Running(self.server()) as base: + for bad in ({"name": None, "lat": 0, "lon": 0}, {"name": "x", "lat": 0, "lon": 0, "lore": 12}, + {"name": "x", "lat": 0, "lon": 0, "lore": "abc\n"}, {"name": "x", "lat": 0, "lon": 0, "note": 5}): + self.assertEqual(request(base + "/api/pins", "POST", bad)[0], 400, bad) + pid = json.loads(request(base + "/api/pins", "POST", {"name": "ok", "lat": 0, "lon": 0})[1])["id"] + for body in ([], b"", b"null"): + self.assertEqual(request(f"{base}/api/pins/{pid}", "PUT", body)[0], 400, body) + + def test_cross_site_and_rebinding_blocked(self): + with Running(self.server()) as base: + port = int(base.rsplit(":", 1)[1]) + c = http.client.HTTPConnection("127.0.0.1", port) # "simple" CORS request: no preflight + c.request("POST", "/api/pins", body=json.dumps({"name": "evil", "lat": 1, "lon": 1}), + headers={"Content-Type": "text/plain", "Origin": "http://evil.example"}) + self.assertEqual(c.getresponse().status, 415) + for path in ("/api/pins", "/"): # DNS rebinding: foreign Host header + c = http.client.HTTPConnection("127.0.0.1", port) + c.request("GET", path, headers={"Host": "evil.example"}) + self.assertEqual(c.getresponse().status, 403, path) + c = http.client.HTTPConnection("127.0.0.1", port) + c.request("GET", "/api/pins", headers={"Host": f"localhost:{port}"}) + self.assertEqual(c.getresponse().status, 200) + self.assertFalse(self.path.exists()) + + def test_cross_site_reads_refused(self): + with Running(self.server()) as base: + port = int(base.rsplit(":", 1)[1]) + def get(path, site): + c = http.client.HTTPConnection("127.0.0.1", port) + c.request("GET", path, headers={"Sec-Fetch-Site": site}) + return c.getresponse().status + self.assertEqual(get("/api/meta", "cross-site"), 403) + self.assertEqual(get("/tiles/v1-0/relief/5/1/1.jpg", "cross-site"), 403) + self.assertEqual(get("/api/meta", "same-origin"), 200) + self.assertEqual(get("/", "cross-site"), 200) # following a shared link still opens the map + + def test_negative_length_and_nul_path(self): + with Running(self.server()) as base: + port = int(base.rsplit(":", 1)[1]) + with socket.create_connection(("127.0.0.1", port), timeout=5) as sk: + sk.sendall(f"POST /api/pins HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nContent-Type: application/json\r\n" + "Content-Length: -1\r\n\r\n".encode()) + self.assertIn(b" 400 ", sk.recv(200)) + self.assertEqual(request(base + "/js/%00x.js")[0], 404) + self.assertEqual(request(base + "/api/meta")[0], 200) + + def test_bad_json_and_oversize(self): + with Running(self.server()) as base: + self.assertEqual(request(base + "/api/pins", "POST", b"{not json")[0], 400) + self.assertEqual(request(base + "/api/pins", "POST", b"x" * (serve.MAX_BODY + 1))[0], 400) + self.assertEqual(request(base + "/api/meta")[0], 200) # still alive + + def test_crash_mid_write_keeps_file(self): + with Running(self.server()) as base: + request(base + "/api/pins", "POST", {"name": "keep", "lat": 1, "lon": 2}) + before = self.path.read_text() + with mock.patch("serve.os.replace", side_effect=OSError("disk full")): + s, _ = request(base + "/api/pins", "POST", {"name": "lost", "lat": 3, "lon": 4}) + self.assertEqual(s, 500) + self.assertEqual(self.path.read_text(), before) + self.assertEqual([p.name for p in self.path.parent.iterdir()], ["pins.json"]) # no temp files left + + def test_concurrent_creates(self): + with Running(self.server()) as base: + threads = [threading.Thread(target=request, args=(base + "/api/pins", "POST", + {"name": f"p{k}", "lat": k, "lon": k})) + for k in range(20)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(len(json.loads(self.path.read_text())["pins"]), 20) + +class ProfileTest(unittest.TestCase): + def setUp(self): + self.root = built_world() + self.cells = np.load(self.root / "out" / "r2" / "cells.npz") + + def server(self): + return serve.make_server(self.root, 2, 0, pins_path=Path(tempfile.mkdtemp()) / "pins.json") + + def test_profile_samples(self): + with Running(self.server()) as base: + s, body = request(f"{base}/api/profile?path=0,0;0,10;5,10&n=64") + p = json.loads(body) + mid = p["samples"][20] + _, cbody = request(f"{base}/api/cell?lat={mid['lat']}&lon={mid['lon']}") + self.assertEqual(s, 200) + S = p["samples"] + self.assertEqual(len(S), 64) + d = [x["d_km"] for x in S] + self.assertTrue(all(b >= a for a, b in zip(d, d[1:]))) + length = 12742.0 * (np.radians(10) + np.radians(5)) + self.assertAlmostEqual(p["stats"]["length_km"], length, delta=0.05) + self.assertAlmostEqual(d[-1], length, delta=0.05) + self.assertAlmostEqual(S[0]["lat"], 0.0, places=4) + self.assertAlmostEqual(S[-1]["lat"], 5.0, places=4) + self.assertAlmostEqual(S[-1]["lon"], 10.0, places=4) + cell = {f["key"]: f for f in json.loads(cbody)["fields"]} + self.assertAlmostEqual(mid["z_cell_m"], cell["z_surface_m"]["value"], delta=0.5) + self.assertIsInstance(mid["water"], bool) # procedural coast: tests/test_tiles.py ProfileSurfaceTest + z = [x["z_m"] for x in S] + st = p["stats"] + self.assertAlmostEqual(st["ascent_m"] - st["descent_m"], z[-1] - z[0], delta=1.0) + self.assertAlmostEqual(st["min_m"], min(z), delta=0.1) + self.assertAlmostEqual(st["max_m"], max(z), delta=0.1) + self.assertTrue(0.0 <= st["water_km"] <= st["length_km"] + 1e-6) + + def test_profile_limits(self): + with Running(self.server()) as base: + bad = ["path=0,0&n=64", "path=" + ";".join(f"{k * 0.1},0" for k in range(101)), "path=0,0;1,1&n=8", + "path=0,0;1,1&n=2000", "path=95,0;1,1", "path=a,b;c", "path=0,0;0,180", "n=64", "path=nan,0;1,1", + "path=0,inf;1,1"] + codes = {q: request(f"{base}/api/profile?{q}") for q in bad} + same = request(f"{base}/api/profile?path=10,10;10,10&n=16") + for q, (s, body) in codes.items(): + self.assertEqual(s, 400, q) + self.assertTrue(json.loads(body)["error"], q) + self.assertIn("antipodal", json.loads(codes["path=0,0;0,180"][1])["error"]) + self.assertIn("lon", json.loads(codes["path=0,inf;1,1"][1])["error"]) + self.assertIn("lat", json.loads(codes["path=95,0;1,1"][1])["error"]) + self.assertEqual(same[0], 200) + self.assertEqual(json.loads(same[1])["stats"]["length_km"], 0.0) + + +class LocateTest(unittest.TestCase): + def test_locate(self): + import h3.api.basic_int as h3 + cell = h3.int_to_str(h3.latlng_to_cell(12.5, 40.0, 5)) + with Running(serve.make_server(built_world(), 2, 0, pins_path=Path(tempfile.mkdtemp()) / "pins.json")) as base: + s, body = request(f"{base}/api/locate?cell={cell.upper()}") + bad = [request(f"{base}/api/locate?cell={q}")[0] for q in ("zz", "", "0528308ffffffff")] + r = json.loads(body) + self.assertEqual(s, 200) + self.assertEqual((r["cell"], r["res"]), (cell, 5)) + self.assertLess(abs(r["lat"] - 12.5) + abs(r["lon"] - 40.0), 0.5) + self.assertEqual(bad, [400, 400, 400]) + + +class RegionsApiTest(unittest.TestCase): + def setUp(self): + self.tmp = Path(tempfile.mkdtemp()) + self.server = serve.make_server(built_world(), 2, 0, pins_path=self.tmp / "p.json", + regions_path=self.tmp / "regions.json", regions_dir=self.tmp / "results") + + def test_read_only_regions_refuse_edits_and_start_no_build(self): + server = serve.make_server(built_world(), 2, 0, pins_path=self.tmp / "p2.json", + regions_path=self.tmp / "ro.json", regions_dir=self.tmp / "ro", regions_read_only=True) + ok = {"name": "Test area", "outline": [[10, 20], [10, 24], [14, 24], [14, 20]]} + with Running(server) as base: + s, _ = request(base + "/api/regions", "POST", ok) + d = request(base + "/api/regions/region-00000000", "DELETE")[0] + listed = json.loads(request(base + "/api/regions")[1]) + self.assertEqual((s, d), (403, 403)) + self.assertEqual(listed["regions"], []) + self.assertFalse((self.tmp / "ro.json").exists()) + + def test_create_list_delete(self): + ok = {"name": "Test area", "outline": [[10, 20], [10, 24], [14, 24], [14, 20]]} + with Running(self.server) as base: + s, body = request(base + "/api/regions", "POST", ok) + self.assertEqual(s, 201) + r = json.loads(body) + self.assertRegex(r["id"], r"^region-[0-9a-f]{8}$") + self.assertGreater(r["area_km2"], 100000) + listed = json.loads(request(base + "/api/regions")[1]) + self.assertEqual([x["id"] for x in listed["regions"]], [r["id"]]) + self.assertIn("url", listed["tiles"]) + self.assertEqual(request(base + f"/api/regions/{r['id']}", "DELETE")[0], 204) + self.assertEqual(request(base + f"/api/regions/{r['id']}", "DELETE")[0], 404) + self.assertEqual(json.loads((self.tmp / "regions.json").read_text())["regions"], []) + + def test_rejects_bad_outlines(self): + bad = [{"name": "", "outline": [[0, 0], [0, 1], [1, 1]]}, {"name": "x", "outline": [[0, 0], [0, 1]]}, + {"name": "x", "outline": [[0, 0], [2, 2], [0, 2], [2, 0]]}, # a bow tie crosses itself + {"name": "x", "outline": [[0, 0], [0, "a"], [1, 1]]}, {"name": "x", "outline": [[95, 0], [0, 1], [1, 1]]}, + "not an object"] + with Running(self.server) as base: + codes = [request(base + "/api/regions", "POST", b)[0] for b in bad] + self.assertEqual(codes, [400] * len(bad)) + + def test_writes_need_json(self): + with Running(self.server) as base: + req = urllib.request.Request(base + "/api/regions", data=b"name=x", method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}) + try: + s = urllib.request.urlopen(req).status + except urllib.error.HTTPError as e: + s = e.code + e.close() + self.assertEqual(s, 415) + + def test_regions_live_where_the_server_was_told(self): + self.assertEqual(self.server.tiles.regions_dir, self.tmp / "results") + + +def wait_status(base, want, timeout=120): + t, st = time.time(), None + while time.time() - t < timeout: + st = json.loads(request(base + "/api/regions")[1]) + if st["status"]["state"] in want: + return st + time.sleep(0.2) + raise AssertionError(f"status never reached {want}: {st and st['status']}") + + +class RegionBuildTest(unittest.TestCase): + def _server(self, tmp): + return serve.make_server(built_world(), 2, 0, pins_path=tmp / "p.json", regions_path=tmp / "regions.json", + regions_dir=tmp / "results") + + def _outline(self): + a = np.load(built_world() / "out" / "r2" / "cells.npz") + i = big_river(a) + la, lo = float(a["g_lat"][i]), float(a["g_lon"][i]) + return [[la - 3, lo - 3], [la - 3, lo + 3], [la + 3, lo + 3], [la + 3, lo - 3]] + + def test_save_builds_and_tiles_switch(self): + tmp = Path(tempfile.mkdtemp()) + server = self._server(tmp) + with Running(server) as base: + url0 = json.loads(request(base + "/api/regions")[1])["tiles"]["url"] + request(base + "/api/regions", "POST", {"name": "A", "outline": self._outline()}) + st = wait_status(base, {"ready", "failed"}) + self.assertEqual(st["status"]["state"], "ready", st["status"]) + self.assertNotEqual(st["tiles"]["url"], url0) + self.assertFalse(server.tiles.regions.empty) + lat, lon = self._outline()[0][0] + 3, self._outline()[0][1] + 3 + self.assertTrue(json.loads(request(base + f"/api/cell?lat={lat}&lon={lon}")[1])["refined"]) + self.assertEqual(request(base + url0.format(layer="relief", z=5, x=0, y=0))[0], 404, "old tile URLs stop") + + def test_delete_rebuilds_and_drops_the_area(self): + tmp = Path(tempfile.mkdtemp()) + server = self._server(tmp) + with Running(server) as base: + r = json.loads(request(base + "/api/regions", "POST", {"name": "A", "outline": self._outline()})[1]) + wait_status(base, {"ready"}) + request(base + f"/api/regions/{r['id']}", "DELETE") + wait_status(base, {"ready"}) + self.assertTrue(server.tiles.regions.empty) + + def test_saves_coalesce_and_failures_report(self): + tmp = Path(tempfile.mkdtemp()) + server = self._server(tmp) + with mock.patch("refine.refine_area", side_effect=RuntimeError("boom")), Running(server) as base: + for k in range(3): + request(base + "/api/regions", "POST", {"name": f"R{k}", "outline": self._outline()}) + st = wait_status(base, {"failed"}) + self.assertIn("boom", st["status"]["error"]) + self.assertTrue(server.tiles.regions.empty, "the map stays as it was") + + def test_progress_is_reported(self): + seen = [] + tmp = Path(tempfile.mkdtemp()) + (tmp / "regions.json").write_text(json.dumps({"regions": [{"id": "region-00000002", "name": "A", + "outline": self._outline(), "created": "2026-09-25"}]})) + import refine as RF + RF.build_areas(built_world(), 2, tmp / "regions.json", log=lambda s: None, regions_root=tmp / "results", + progress=lambda stage, f: seen.append((stage, f))) + self.assertTrue({"fields", "erosion", "rivers", "ground", "saving"} <= {s for s, _ in seen}) + fr = [f for _, f in seen] + self.assertEqual(fr, sorted(fr), "fractions only grow") + self.assertTrue(0.0 <= fr[0] and fr[-1] <= 1.0) + + def test_after_a_build_the_region_tiles_are_prerendered(self): + tmp = Path(tempfile.mkdtemp()) + server = self._server(tmp) + server.prerender_z = 5 + with Running(server) as base: + request(base + "/api/regions", "POST", {"name": "A", "outline": self._outline()}) + wait_status(base, {"ready"}) + t, st = time.time(), {} + while time.time() - t < 120: + st = json.loads(request(base + "/api/regions")[1])["status"] + if st.get("tiles") and st["tiles"]["done"] == st["tiles"]["total"] > 0: + break + time.sleep(0.3) + self.assertEqual(st["tiles"]["done"], st["tiles"]["total"], st) + self.assertTrue(any(server.tiles.cache_dir.rglob("*.bin"))) + + def test_rejects_regions_far_over_the_size_limit(self): + tmp = Path(tempfile.mkdtemp()) + with Running(self._server(tmp)) as base: + s, body = request(base + "/api/regions", "POST", {"name": "Huge", "outline": [[-30, 0], [-30, 50], [30, 50], [30, 0]]}) + self.assertEqual(s, 400) + self.assertIn("too large", json.loads(body)["error"]) + + def test_ready_is_never_reported_while_a_save_waits(self): + import refine as RF + tmp = Path(tempfile.mkdtemp()) + server = self._server(tmp) + seen, real = [], RF.build_areas + + def build(*a, **k): + if not seen: + server.jobs.request() # a save lands during the build + seen.append("build") + return real(*a, **k) + server.jobs.prerender = lambda: seen.append(server.jobs.status()["state"]) + with mock.patch("refine.build_areas", side_effect=build), Running(server) as base: + request(base + "/api/regions", "POST", {"name": "A", "outline": self._outline()}) + t = time.time() + while seen.count("build") < 2 or seen[-1] == "build": + self.assertLess(time.time() - t, 60, seen) + time.sleep(0.1) + self.assertEqual(seen, ["build", "build", "ready"], "ready (and pre-rendering) only once nothing waits") + + def test_a_crashing_build_never_wedges_the_queue(self): + tmp = Path(tempfile.mkdtemp()) + server = self._server(tmp) + with mock.patch("refine.build_areas", side_effect=[serve.MissingBuild("world rebuilding"), []]), \ + Running(server) as base: + server.jobs.request() + self.assertEqual(wait_status(base, {"failed"}, 20)["status"]["state"], "failed") + server.jobs.request() + self.assertEqual(wait_status(base, {"ready"}, 20)["status"]["state"], "ready") + + def test_a_malformed_regions_file_is_reported_not_fatal(self): + tmp = Path(tempfile.mkdtemp()) + (tmp / "regions.json").write_text('{"regions": [ {"name": broken') + server = self._server(tmp) # starts anyway + with Running(server) as base: + st = json.loads(request(base + "/api/regions")[1]) + self.assertEqual(st["regions"], []) + self.assertEqual(st["status"]["state"], "failed") + self.assertIn("regions.json", st["status"]["error"]) + s, _ = request(base + "/api/regions", "POST", {"name": "A", "outline": self._outline()}) + self.assertGreaterEqual(s, 400) + self.assertEqual((tmp / "regions.json").read_text(), '{"regions": [ {"name": broken', "never overwritten") + + def test_each_region_says_whether_it_is_refined(self): + tmp = Path(tempfile.mkdtemp()) + server = self._server(tmp) + o = self._outline() + far = [[la + 40 if la < 40 else la - 40, lo + 60] for la, lo in o] + with Running(server) as base: + request(base + "/api/regions", "POST", {"name": "A", "outline": o}) + wait_status(base, {"ready"}) + with mock.patch("refine.refine_area", side_effect=RuntimeError("boom")): + request(base + "/api/regions", "POST", {"name": "B", "outline": far}) + st = wait_status(base, {"failed"}) + self.assertEqual({r["name"]: r["refined"] for r in st["regions"]}, {"A": True, "B": False}) + + def test_saves_during_a_build_coalesce_into_one_more_build(self): + tmp = Path(tempfile.mkdtemp()) + server = self._server(tmp) + builds = [] + + def slow(*a, **k): + builds.append(1) + time.sleep(1.0) + return [] + with mock.patch("refine.build_areas", side_effect=slow), Running(server) as base: + for k in range(4): + request(base + "/api/regions", "POST", {"name": f"R{k}", "outline": self._outline()}) + wait_status(base, {"ready"}) + time.sleep(0.3) + self.assertEqual(len(builds), 2, "the first save builds; the three during it share one more build") + self.assertEqual(len(json.loads(request(base + "/api/regions")[1])["regions"]), 4, "no save lost") + + def test_deleting_a_region_while_it_builds_leaves_nothing_behind(self): + import refine as RF + tmp = Path(tempfile.mkdtemp()) + server = self._server(tmp) + with Running(server) as base: + r = json.loads(request(base + "/api/regions", "POST", {"name": "A", "outline": self._outline()})[1]) + wait_status(base, {"building"}, 30) + self.assertEqual(request(base + f"/api/regions/{r['id']}", "DELETE")[0], 204) + t = time.time() + while True: + st = json.loads(request(base + "/api/regions")[1])["status"] + if st["state"] == "ready" and not server.jobs.pending: + break + self.assertLess(time.time() - t, 120, st) + time.sleep(0.2) + self.assertTrue(server.tiles.regions.empty, "the map shows no refined area") + self.assertEqual(RF.built_areas(tmp / "results", server.tiles.fingerprint), [], "no result left on disk") + + def test_a_restart_with_everything_built_does_no_heavy_area_work(self): + tmp = Path(tempfile.mkdtemp()) + server = self._server(tmp) + with Running(server) as base: + request(base + "/api/regions", "POST", {"name": "A", "outline": self._outline()}) + wait_status(base, {"ready"}) + with mock.patch("refine.area_cells", side_effect=AssertionError("area_cells at start")): + again = self._server(tmp) # a restart: nothing to build … + with Running(again) as base: + st = json.loads(request(base + "/api/regions")[1]) + self.assertEqual(st["status"]["state"], "idle") + self.assertEqual([r["refined"] for r in st["regions"]], [True], "… and answered from the build's record") + + def test_restart_queues_missing_areas(self): + tmp = Path(tempfile.mkdtemp()) + (tmp / "regions.json").write_text(json.dumps({"regions": [{"id": "region-00000001", "name": "A", + "outline": self._outline(), "created": "2026-09-25"}]})) + server = self._server(tmp) + with Running(server) as base: + wait_status(base, {"ready"}) + self.assertFalse(server.tiles.regions.empty) + + +class ExportsApiTest(unittest.TestCase): + def _server(self, tmp, workers=0): + return serve.make_server(built_world(), 2, 0, pins_path=tmp / "p.json", regions_path=tmp / "regions.json", + regions_dir=tmp / "results", exports_dir=tmp / "exports", workers=workers) + + def _wait(self, base, want, timeout=120): + t, st = time.time(), None + while time.time() - t < timeout: + st = json.loads(request(base + "/api/exports")[1]) + if st["status"]["state"] in want: + return st + time.sleep(0.2) + raise AssertionError(f"export status never reached {want}: {st}") + + def test_export_from_the_viewer(self): + tmp = Path(tempfile.mkdtemp()) + with Running(self._server(tmp)) as base: + s, body = request(base + "/api/exports", "POST", {"lat": 10.0, "lon": 20.0, "size_km": 100, "res_m": 2000, + "name": "test-area"}) + self.assertEqual(s, 202, body) + st = self._wait(base, {"done", "failed"}) + self.assertEqual(st["status"]["state"], "done", st["status"]) + (e,) = st["exports"] + self.assertEqual((e["name"], e["samples"], e["res_m"]), ("test-area", 65, 2000)) + self.assertTrue((tmp / "exports" / "test-area" / "height.f32").exists()) + self.assertEqual(request(base + "/api/exports", "POST", {"lat": 10, "lon": 20, "size_km": 100, "res_m": 2000, + "name": "test-area"})[0], 409, "never overwritten") + + def test_rejects_bad_exports(self): + tmp = Path(tempfile.mkdtemp()) + bad = [{"lat": 95, "lon": 0, "size_km": 10, "res_m": 10}, {"lat": 0, "lon": 0, "size_km": 0, "res_m": 10}, + {"lat": 0, "lon": 0, "size_km": 10, "res_m": 0.1}, {"lat": 0, "lon": 0, "size_km": 400, "res_m": 10}, + {"lat": 0, "lon": 0, "size_km": 10, "res_m": 10, "name": "../escape"}, + {"lat": 0, "lon": 0, "size_km": 100, "res_m": 1}, "nope", + {"lat": 84.9, "lon": 0, "size_km": 40, "res_m": 100}, # reaches within 5° of the pole + {"lat": 75, "lon": 0, "size_km": 80, "res_m": 10}] # ≈ 5,000 tiles: far too much memory + with Running(self._server(tmp)) as base: + codes = [request(base + "/api/exports", "POST", b)[0] for b in bad] + self.assertEqual(codes, [400] * len(bad)) + self.assertFalse((tmp / "exports").exists() and any((tmp / "exports").iterdir())) + + def test_half_written_exports_are_neither_listed_nor_kept(self): + tmp = Path(tempfile.mkdtemp()) + junk = tmp / "exports" / ".island-x1y2z3" # an export interrupted by Ctrl-C + junk.mkdir(parents=True) + (junk / "meta.json").write_text('{"name": "island"}') + server = self._server(tmp) # a restart clears it + with Running(server) as base: + self.assertEqual(json.loads(request(base + "/api/exports")[1])["exports"], []) + self.assertFalse(junk.exists()) + + def test_one_export_at_a_time(self): + import export as EX + tmp = Path(tempfile.mkdtemp()) + real = EX.export + + def slow(*a, **k): + time.sleep(1.0) + return real(*a, **k) + with mock.patch("export.export", side_effect=slow), Running(self._server(tmp)) as base: + ok = {"lat": 10, "lon": 20, "size_km": 50, "res_m": 2000} + self.assertEqual(request(base + "/api/exports", "POST", {**ok, "name": "a"})[0], 202) + self.assertEqual(request(base + "/api/exports", "POST", {**ok, "name": "b"})[0], 409, "busy") + self._wait(base, {"done"}) + + def test_render_workers_make_the_same_export(self): + tmp = Path(tempfile.mkdtemp()) + with Running(self._server(tmp, workers=2)) as base: + request(base + "/api/exports", "POST", {"lat": 10.0, "lon": 20.0, "size_km": 100, "res_m": 2000, "name": "w"}) + self._wait(base, {"done"}) + tmp2 = Path(tempfile.mkdtemp()) + with Running(self._server(tmp2)) as base: + request(base + "/api/exports", "POST", {"lat": 10.0, "lon": 20.0, "size_km": 100, "res_m": 2000, "name": "w"}) + self._wait(base, {"done"}) + for f in ("height.f32", "water.png", "preview.png"): + self.assertEqual((tmp / "exports" / "w" / f).read_bytes(), (tmp2 / "exports" / "w" / f).read_bytes(), f) + + +class ServeCacheWorldTest(unittest.TestCase): + def test_world_arrays_come_from_the_cache_and_equal_the_npz(self): + root = built_world() + w1 = serve.World(root, 2) + w2 = serve.World(root, 2) + self.assertTrue((w1.cache / "done").exists()) + self.assertEqual(w1.cache, w2.cache) + with np.load(root / "out" / "r2" / "cells.npz") as z: + for k in serve.LOAD_KEYS: + if k not in z.files: + continue + want = z[k].astype(np.float32) if z[k].dtype == np.float64 else z[k] + self.assertIsInstance(w2.arrays[k], np.memmap, k) + self.assertEqual(w2.arrays[k].dtype, want.dtype, k) + np.testing.assert_array_equal(w2.arrays[k], want, err_msg=k) + + def test_world_without_cache_dir_write(self): + root = built_world() + with mock.patch.object(serve.servecache, "write", side_effect=OSError("read-only file system")): + w = serve.World(root, 2, use_cache=True) + plain = serve.World(root, 2, use_cache=False) + for k in plain.arrays: + np.testing.assert_array_equal(w.arrays[k], plain.arrays[k], err_msg=k) + + def test_low_memory_world_cache_equals_plain_and_cleans_up(self): + src = built_world() + plain = serve.World(src, 2, use_cache=False) # oracle: arrays read straight into RAM + with tempfile.TemporaryDirectory() as d, mock.patch.dict(os.environ, {"WORLDGEN_LOW_MEMORY": "1"}): + root = Path(d) / "w" + shutil.copytree(src, root, symlinks=True) + for p in (root / "out" / "r2").glob("serve-*"): + shutil.rmtree(p) + with mock.patch.object(serve.servecache, "spiller", wraps=serve.servecache.spiller) as sp: + w = serve.World(root, 2) + self.assertTrue(any(c.args[0] is not None for c in sp.call_args_list), "fields spilled to disk") + for k in plain.arrays: + np.testing.assert_array_equal(w.arrays[k], plain.arrays[k], err_msg=k) + self.assertEqual(list((root / "out" / "r2").glob(".spill-*")), []) + + def test_cache_key_follows_the_build(self): + root = built_world() + out = root / "out" / "r2" + k1 = serve.world_cache_key(out) + st = (out / "cells.npz").stat() + try: + os.utime(out / "cells.npz") + self.assertNotEqual(serve.world_cache_key(out), k1) + finally: # the shared test world: later tests keep their fingerprints + os.utime(out / "cells.npz", ns=(st.st_atime_ns, st.st_mtime_ns)) + + def test_cached_arrays_are_read_only_and_everything_still_works(self): + root = built_world() + w = serve.World(root, 2) + self.assertFalse(w.arrays["z_surface_m"].flags.writeable) + i = len(w.ids) // 3 + lat, lon = float(np.degrees(np.arcsin(w.arrays["g_xyz"][i][2]))), float(np.degrees(np.arctan2(w.arrays["g_xyz"][i][1], w.arrays["g_xyz"][i][0]))) + d = w.describe(lat, lon) # the inspector path on memory maps + self.assertIn("fields", d) + + +class InspectorFieldsTest(unittest.TestCase): + def test_sea_floor_and_zone_fields(self): + import h3.api.basic_int as h3 + w = serve.World(built_world(), 2, era={"name": "x", "label": "Era X"}) + i = int(np.flatnonzero(w.arrays["ocean"].astype(bool))[0]) + la, lo = h3.cell_to_latlng(int(w.ids[i])) + d = w.describe(la, lo) + keys = {f["key"]: f for f in d["fields"]} + for k in ("seabed_type", "seabed_mineral", "bottom_temp_c", "sediment_m", "vent_potential", + "o2_fraction", "pressure_bar", "fire_reactivity", "po2_bar", "gravity_g"): + self.assertIn(k, keys, k) + self.assertEqual(keys["seabed_type"]["group"], "Sea floor") + self.assertEqual(d["era"], {"name": "x", "label": "Era X"}) + j = int(np.flatnonzero(~w.arrays["ocean"].astype(bool))[0]) + la, lo = h3.cell_to_latlng(int(w.ids[j])) + self.assertNotIn("seabed_type", {f["key"] for f in w.describe(la, lo)["fields"]}, "land: no sea floor") + + def test_sea_rows(self): + # a sea cell lists SST, current speed and productivity in group "Sea"; a land cell lists none of them + import h3.api.basic_int as h3 + w = serve.World(built_world(), 2) + i = int(np.flatnonzero(w.arrays["ocean"].astype(bool))[0]) + keys = {f["key"]: f for f in w.describe(*h3.cell_to_latlng(int(w.ids[i])))["fields"]} + for k in ("sst", "current_speed", "productivity"): + self.assertIn(k, keys, k) + self.assertEqual(keys[k]["group"], "Sea") + j = int(np.flatnonzero(~w.arrays["ocean"].astype(bool))[0]) + land = {f["key"] for f in w.describe(*h3.cell_to_latlng(int(w.ids[j])))["fields"]} + for k in ("sst", "current_speed", "productivity"): + self.assertNotIn(k, land, k) + + +class EraServerTest(unittest.TestCase): + def test_every_era_is_served(self): + from tests.test_refine import era_world + root, far, cut, _ = era_world() + with Running(serve.make_server(root, 2, 0, pins_path=Path(tempfile.mkdtemp()) / "p.json", + regions_dir=Path(tempfile.mkdtemp()))) as base: + m = json.loads(request(base + "/api/meta")[1]) + self.assertEqual(m["default_era"], "after") + self.assertEqual(m["era"], "after") + self.assertEqual([e["name"] for e in m["eras"]], ["before", "after"]) + b = json.loads(request(base + "/api/meta?era=before")[1]) + self.assertNotEqual(b["tiles"]["url"], m["tiles"]["url"]) + self.assertTrue(b["layers"][0]["file"].endswith("?era=before")) + for meta in (m, b): + tag = meta["tiles"]["url"].split("/")[2] + self.assertEqual(request(base + f"/tiles/{tag}/relief/5/10/10.jpg")[0], 200) + q = f"lat={cut[0]}&lon={cut[1]}" + ca = json.loads(request(base + f"/api/cell?{q}&era=after")[1]) + cb = json.loads(request(base + f"/api/cell?{q}&era=before")[1]) + self.assertEqual((ca["era"]["name"], cb["era"]["name"]), ("after", "before")) + za = {f["key"]: f["value"] for f in ca["fields"]}["z_surface_m"] + zb = {f["key"]: f["value"] for f in cb["fields"]}["z_surface_m"] + self.assertLess(za, zb, "the cut lowered the ground") + self.assertEqual(request(base + f"/api/cell?{q}&era=nope")[0], 400) + self.assertEqual(request(base + "/data/viewer/layers.json?era=before")[0], 200) + + def test_single_world_server_has_one_era(self): + with Running(serve.make_server(built_world(), 2, 0, pins_path=Path(tempfile.mkdtemp()) / "p.json")) as base: + m = json.loads(request(base + "/api/meta")[1]) + self.assertEqual(len(m["eras"]), 1) + self.assertEqual(m["era"], m["default_era"]) + self.assertEqual(request(base + "/api/cell?lat=10&lon=20&era=nope")[0], 400) + self.assertEqual(request(base + "/api/cell?lat=10&lon=20")[0], 200) + + +class VentsApiTest(unittest.TestCase): + def test_vents_without_refined_areas(self): + with Running(serve.make_server(built_world(), 2, 0, pins_path=Path(tempfile.mkdtemp()) / "p.json", + regions_dir=Path(tempfile.mkdtemp()))) as base: + s, b = request(base + "/api/vents") + self.assertEqual(s, 200) + v = json.loads(b) + self.assertEqual(v["vents"], []) + self.assertEqual(v["types"][0], "black smoker") + + def test_vents_of_refined_areas(self): + from unittest import mock + import refine as RF + import seafloor as SF + from tests.test_refine import square, write_regions + root = built_world() + w = RF.WorldCells(root / "out" / "r2") + i = int(np.argmax(np.where(w.a["ocean"].astype(bool), w.a["vent_potential"], -1.0))) # the likeliest vents + tmp = Path(tempfile.mkdtemp()) + reg = write_regions(tmp / "regions.json", square(float(w.a["g_lat"][i]), float(w.a["g_lon"][i]), 3, "deep")) + with mock.patch.dict(SF.VENT, rate=1.0): # the small world's odds give ≈ 1 vent: make it many + RF.build_areas(root, 2, reg, log=lambda s: None, regions_root=tmp / "res") + with Running(serve.make_server(root, 2, 0, pins_path=tmp / "p.json", regions_path=reg, + regions_dir=tmp / "res")) as base: + v = json.loads(request(base + "/api/vents")[1]) + links = [p for p in (tmp / "res").glob(f"*-m{RF.MODEL}/*") if p.is_symlink()] + n = len({int(c) for p in links for c in np.load(p / "vents.npz")["cell"]}) + self.assertGreater(n, 0, "a deep-sea square holds vents") + self.assertEqual(len(v["vents"]), n) + for x in v["vents"]: + self.assertEqual(set(x), {"lat", "lon", "type", "temp_c", "flow", "mineral"}) + + def test_plateau_outlines(self): + root = built_world() + from mapgen import config as C + with Running(serve.make_server(root, 2, 0, pins_path=Path(tempfile.mkdtemp()) / "p.json")) as base: + meta = json.loads(request(base + "/api/meta")[1]) + s, b = request(base + "/api/plateaus.geojson") + fc = json.loads(b) + self.assertEqual(s, 200) + self.assertEqual(len(fc["features"]), len(C.load(root)[1].get("plateau", []))) + if fc["features"]: + self.assertIn("plateaus", [o["id"] for o in meta["overlays"]]) + + +class EraRegionsMissingTest(unittest.TestCase): + def test_a_non_default_era_without_its_areas_is_missing(self): + import shutil + import refine as RF + from tests.test_refine import era_world, square, write_regions + root, far, cut, _ = era_world() + tmp = Path(tempfile.mkdtemp()) + reg = write_regions(tmp / "regions.json", square(*far, 2, "far")) + RF.build_areas(root, 2, reg, log=lambda s: None, regions_root=tmp / "res") + server = serve.make_server(root, 2, 0, pins_path=tmp / "p.json", regions_path=reg, regions_dir=tmp / "res") + try: + self.assertFalse(serve.regions_missing(server), "every era built") + base = server.sources["before"] + shutil.rmtree(RF.results_dir(tmp / "res", base.fingerprint)) + self.assertTrue(serve.regions_missing(server), "the base era (not the default) lost its areas") + finally: + server.server_close() + + +class ClientGoneTest(unittest.TestCase): + def test_a_reply_to_a_client_that_hung_up_is_dropped_quietly(self): + class Gone(io.RawIOBase): + def write(self, b): + raise BrokenPipeError(32, "Broken pipe") + h = serve.Handler.__new__(serve.Handler) + h.wfile, h.request_version, h.requestline, h.command = Gone(), "HTTP/1.1", "GET / HTTP/1.1", "GET" + h.client_address, h.close_connection = ("127.0.0.1", 0), False + h.log_request = lambda *a, **k: None + h._json(200, {"x": 1}) # no exception: the client is gone, nothing to answer + self.assertTrue(h.close_connection) + + +class GzipTest(unittest.TestCase): + def _reply(self, accept, body, ctype): + out = io.BytesIO() + h = serve.Handler.__new__(serve.Handler) + h.wfile, h.request_version, h.requestline, h.command = out, "HTTP/1.1", "GET / HTTP/1.1", "GET" + h.client_address, h.close_connection = ("127.0.0.1", 0), False + h.log_request = lambda *a, **k: None + h.headers = {} if accept is None else {"Accept-Encoding": accept} + h._send(200, body, ctype) + head, _, got = out.getvalue().partition(b"\r\n\r\n") + return head.decode().lower(), got + + def test_mesh_and_json_gzipped_when_accepted(self): + import gzip + body = np.arange(20000, dtype=np.float32).tobytes() + for ctype in ("application/octet-stream", "application/json"): + head, got = self._reply("gzip, deflate, br", body, ctype) + self.assertIn("content-encoding: gzip", head) + self.assertIn("vary: accept-encoding", head) + self.assertIn(f"content-length: {len(got)}", head) + self.assertLess(len(got), len(body)) + self.assertEqual(gzip.decompress(got), body) + + def test_plain_when_not_accepted_small_or_already_compressed(self): + body = bytes(5000) + for accept, b, ctype in ((None, body, "application/json"), ("identity", body, "application/json"), + ("gzip;q=0", body, "application/json"), ("gzip", body, "image/jpeg"), + ("gzip", b"{}", "application/json")): + head, got = self._reply(accept, b, ctype) + self.assertNotIn("content-encoding", head, (accept, ctype)) + self.assertEqual(got, b) + + +class PublicModeTest(unittest.TestCase): + def setUp(self): + self.root = built_world() + self.dir = Path(tempfile.mkdtemp()) + lore = self.dir / "lore" + lore.mkdir() + (lore / "true-gods.md").write_text("---\nid: true-gods\n---\nTier-1.\n") + self.lore = lore + + def tearDown(self): + shutil.rmtree(self.dir) + + def server(self, public): + s = serve.make_server(self.root, 2, 0, pins_path=self.dir / "pins.json", lore_dir=self.lore, + regions_path=self.dir / "regions.json", exports_dir=self.dir / "exports") + s.public = public + return s + + def test_private_server_serves_edit_apis(self): + with Running(self.server(False)) as base: + meta = json.loads(request(base + "/api/meta")[1]) + self.assertFalse(meta["read_only"]) + self.assertEqual(meta["lore_ids"], ["true-gods"]) + self.assertEqual(request(base + "/api/pins")[0], 200) + self.assertEqual(request(base + "/api/pins", "POST", {"name": "A", "lat": 1, "lon": 2})[0], 201) + + def test_public_server_is_read_only_and_keeps_notes_private(self): + (self.dir / "pins.json").write_text(json.dumps({"pins": [{"id": "x", "name": "Secret", "lat": 1, "lon": 2}]})) + with Running(self.server(True)) as base: + meta = json.loads(request(base + "/api/meta")[1]) + self.assertTrue(meta["read_only"]) + self.assertEqual(meta["lore_ids"], [], "lore entry names stay private") + for path in ("/api/pins", "/api/regions", "/api/exports"): + self.assertEqual(request(base + path)[0], 404, path) + self.assertEqual(request(base + "/api/pins", "POST", {"name": "A", "lat": 1, "lon": 2})[0], 403) + self.assertEqual(request(base + "/api/pins/x", "PUT", {"name": "B", "lat": 1, "lon": 2})[0], 403) + self.assertEqual(request(base + "/api/pins/x", "DELETE")[0], 403) + self.assertEqual(request(base + "/api/regions", "POST", {"name": "R", "outline": [[0, 0], [0, 1], [1, 1]]})[0], 403) + self.assertEqual(request(base + "/api/exports", "POST", {})[0], 403) + self.assertEqual(request(base + "/api/cell?lat=10&lon=20")[0], 200, "the map itself still answers") + tile = meta["tiles"]["url"].replace("{layer}", "relief").replace("{z}", "5").replace("{x}", "0").replace("{y}", "0") + self.assertEqual(request(base + tile)[0], 200) + self.assertIn("Secret", (self.dir / "pins.json").read_text(), "nothing written") + + +class AllowHostTest(unittest.TestCase): + def setUp(self): + self.root = built_world() + self.dir = Path(tempfile.mkdtemp()) + + def tearDown(self): + shutil.rmtree(self.dir) + + def get(self, base, path, host): + port = int(base.rsplit(":", 1)[1]) + c = http.client.HTTPConnection("127.0.0.1", port) + try: + c.request("GET", path, headers={"Host": host}) + r = c.getresponse() + return r.status, r.getheader("Cache-Control"), r.read() + finally: + c.close() + + def test_public_name_accepted_only_when_allowed_and_tiles_cached_for_good(self): + s = serve.make_server(self.root, 2, 0, pins_path=self.dir / "pins.json") + s.public, s.allowed_hosts = True, {"maps.example.org"} + with Running(s) as base: + self.assertEqual(self.get(base, "/api/meta", "maps.example.org")[0], 200) + self.assertEqual(self.get(base, "/api/meta", "MAPS.example.org")[0], 200) + self.assertEqual(self.get(base, "/api/meta", "evil.example.com")[0], 403) + self.assertEqual(self.get(base, "/api/meta", f"localhost:{s.server_address[1]}")[0], 200) + meta = json.loads(request(base + "/api/meta")[1]) + tile = meta["tiles"]["url"].replace("{layer}", "relief").replace("{z}", "5").replace("{x}", "0").replace("{y}", "0") + st, cache, _ = self.get(base, tile, "maps.example.org") + self.assertEqual(st, 200) + self.assertIn("immutable", cache) + + def test_private_server_refuses_other_hosts_and_caches_tiles_a_day(self): + s = serve.make_server(self.root, 2, 0, pins_path=self.dir / "pins.json") + with Running(s) as base: + self.assertEqual(self.get(base, "/api/meta", "maps.example.org")[0], 403) + meta = json.loads(request(base + "/api/meta")[1]) + tile = meta["tiles"]["url"].replace("{layer}", "relief").replace("{z}", "5").replace("{x}", "0").replace("{y}", "0") + self.assertEqual(self.get(base, tile, f"127.0.0.1:{s.server_address[1]}")[1], "max-age=86400") + + def test_allow_host_requires_public(self): + with mock.patch.object(serve.ThreadingHTTPServer, "serve_forever", side_effect=KeyboardInterrupt), \ + contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()), \ + self.assertRaises(SystemExit): # not serving: refused before any server starts + serve.main(["--res", "2", "--port", "0", "--allow-host", "maps.example.org"], root=self.root) + + +class TileCapServeTest(unittest.TestCase): + def setUp(self): + self.root = built_world() + self.dir = Path(tempfile.mkdtemp()) + + def tearDown(self): + shutil.rmtree(self.dir) + + def test_saved_tiles_count_and_disk_reads_touch(self): + from tilecap import TileCap + s = serve.make_server(self.root, 2, 0, pins_path=self.dir / "pins.json") + src = s.tiles + src.cache_root = self.dir / "tiles" # a fresh cache: this tile renders + cap = TileCap(10**9) + cap.add_root(src.cache_root) + cap.sweep() + src.cap = cap + z, x, y = 7, 100, 40 + body = src.tile("relief", z, x, y, remember=False) + self.assertEqual(cap.total, len(body)) + f = next(src.cache_root.rglob(f"relief/{z}/{x}/{y}.jpg")) + os.utime(f, (1, 1)) + self.assertEqual(src.tile("relief", z, x, y, remember=False), body) + self.assertGreater(f.stat().st_mtime, 1e9, "a disk read marks the tile used") + s.server_close() + + def test_main_tile_cache_options(self): + made = [] + real = serve.make_server + + def spy(*a, **k): + made.append(real(*a, **{**k, "workers": 0})) + return made[-1] + with socket.socket() as so: + so.bind(("127.0.0.1", 0)) + port = so.getsockname()[1] + with mock.patch.object(serve, "make_server", spy), \ + mock.patch.object(serve.ThreadingHTTPServer, "serve_forever", side_effect=KeyboardInterrupt), \ + contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(serve.main(["--res", "2", "--port", str(port), "--tile-cache-gb", "1.5", + "--tile-keep-z", "9"], root=self.root), 0) + self.assertEqual(serve.main(["--res", "2", "--port", str(port)], root=self.root), 0) + cap = made[0].tiles.cap + self.assertEqual((cap.cap, cap.keep_z), (1_500_000_000, 9)) + self.assertIn(made[0].tiles.cache_root, cap.roots) + self.assertIsNone(made[1].tiles.cap, "no cap unless asked") + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + serve.main(["--res", "2", "--tile-cache-gb", "0"], root=self.root) + + +class RenderGateTest(unittest.TestCase): + def setUp(self): + self.root = built_world() + self.dir = Path(tempfile.mkdtemp()) + + def tearDown(self): + shutil.rmtree(self.dir) + + def test_full_render_slots_answer_503_but_saved_tiles_still_serve(self): + s = serve.make_server(self.root, 2, 0, pins_path=self.dir / "pins.json") + s.tiles.cache_root = self.dir / "tiles" + s.render_gate = threading.BoundedSemaphore(1) + with Running(s) as base: + meta = json.loads(request(base + "/api/meta")[1]) + url = lambda x: base + meta["tiles"]["url"].replace("{layer}", "relief").replace("{z}", "7") \ + .replace("{x}", str(x)).replace("{y}", "40") + self.assertEqual(request(url(100))[0], 200) # a free slot: renders and saves + self.assertTrue(s.render_gate.acquire(blocking=False)) # every slot taken + try: + r = urllib.request.urlopen(url(101)) + self.fail("rendered without a slot") + except urllib.error.HTTPError as e: + self.assertEqual(e.code, 503) + self.assertEqual(e.headers["Retry-After"], str(serve.BUSY_RETRY_S)) + s.tiles.mem.clear() + self.assertEqual(request(url(100))[0], 200, "a saved tile needs no slot") + s.render_gate.release() + self.assertEqual(request(url(101))[0], 200, "a slot again: renders") + + def test_main_render_limit_defaults(self): + made = [] + real = serve.make_server + + def spy(*a, **k): + made.append(real(*a, **{**k, "workers": 0})) + return made[-1] + with socket.socket() as so: + so.bind(("127.0.0.1", 0)) + port = so.getsockname()[1] + argv = ["--res", "2", "--port", str(port), "--workers", "3"] + with mock.patch.object(serve, "make_server", spy), \ + mock.patch.object(serve.ThreadingHTTPServer, "serve_forever", side_effect=KeyboardInterrupt), \ + contextlib.redirect_stdout(io.StringIO()): + for extra in ([], ["--public"], ["--public", "--max-renders", "5"]): + self.assertEqual(serve.main(argv + extra, root=self.root), 0) + self.assertIsNone(made[0].render_gate, "private: no limit") + self.assertEqual(made[1].render_gate._initial_value, 12) # 4 × 3 workers + self.assertEqual(made[2].render_gate._initial_value, 5) + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + serve.main(["--res", "2", "--max-renders", "0"], root=self.root) + + +class EnvOptionsTest(unittest.TestCase): + def parser(self): + import argparse + ap = argparse.ArgumentParser() + ap.add_argument("--port", type=int, default=8765) + ap.add_argument("--public", action="store_true") + ap.add_argument("--tile-cache-gb", type=float) + ap.add_argument("--world", action="append") + return ap + + def test_env_values_are_defaults_and_flags_win(self): + ap = self.parser() + serve.env_defaults(ap, {"WORLDMAP_PORT": "9000", "WORLDMAP_PUBLIC": "yes", "WORLDMAP_TILE_CACHE_GB": "140", + "WORLDMAP_WORLD": "/ignored/here", "OTHER": "1"}) + a = ap.parse_args([]) + self.assertEqual((a.port, a.public, a.tile_cache_gb, a.world), (9000, True, 140.0, None)) + self.assertEqual(ap.parse_args(["--port", "8000"]).port, 8000) + ap = self.parser() + serve.env_defaults(ap, {"WORLDMAP_PUBLIC": "off"}) + self.assertFalse(ap.parse_args([]).public) + + def test_bad_env_values_name_the_variable(self): + for env in ({"WORLDMAP_PORT": "x"}, {"WORLDMAP_PUBLIC": "maybe"}): + with self.assertRaisesRegex(ValueError, list(env)[0]): + serve.env_defaults(self.parser(), env) + + def test_env_list(self): + self.assertEqual(serve.env_list("a.org, b.org c.org"), ["a.org", "b.org", "c.org"]) + + def test_main_reads_the_environment(self): + root = built_world() + made = [] + real = serve.make_server + + def spy(*a, **k): + made.append((k["bind"], real(*a, **{**k, "workers": 0, "bind": "127.0.0.1"}))) + return made[-1][1] + with socket.socket() as so: + so.bind(("127.0.0.1", 0)) + port = so.getsockname()[1] + env = {"WORLDMAP_PUBLIC": "1", "WORLDMAP_BIND": "0.0.0.0", "WORLDMAP_ALLOW_HOST": "maps.example.org,m.example.org", + "WORLDMAP_PORT": str(port), "WORLDMAP_TILE_CACHE_GB": "2", "WORLDMAP_RES": "2", "WORLDMAP_WORKERS": "3", + "WORLDMAP_WORLD": str(root)} + with mock.patch.dict(os.environ, env), mock.patch.object(serve, "make_server", spy), \ + mock.patch.object(serve.ThreadingHTTPServer, "serve_forever", side_effect=KeyboardInterrupt), \ + contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(serve.main([], root=Path("/nonexistent")), 0) # the world comes from the environment + bind, s = made[0] + self.assertEqual(bind, "0.0.0.0") + self.assertTrue(s.public) + self.assertEqual(s.allowed_hosts, {"maps.example.org", "m.example.org"}) + self.assertEqual(s.tiles.cap.cap, 2_000_000_000) + self.assertEqual(s.render_gate._initial_value, 12) + + def test_outside_bind_needs_public(self): + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + serve.main(["--res", "2", "--bind", "0.0.0.0"], root=built_world()) + + +class StaleStyleTest(unittest.TestCase): + def test_a_build_with_a_renamed_style_says_rebuild(self): + tmp = Path(tempfile.mkdtemp()) + try: + root = tmp / "w" + shutil.copytree(built_world(), root) + meta = root / "out" / "r2" / "cells_meta.json" + m = json.loads(meta.read_text()) + m["style"] = "dark-moon" # the moons' palette before its rename + meta.write_text(json.dumps(m)) + with self.assertRaisesRegex(serve.MissingBuild, "rebuild"): + serve.World(root, 2, use_cache=False) + m["style"] = "tidal-lock" + meta.write_text(json.dumps(m)) + self.assertEqual(serve.World(root, 2, use_cache=False).cells_meta["style"], "tidal-lock") + finally: + shutil.rmtree(tmp) diff --git a/tests/test_servecache.py b/tests/test_servecache.py new file mode 100644 index 0000000..6f45825 --- /dev/null +++ b/tests/test_servecache.py @@ -0,0 +1,182 @@ +# map/tests/test_servecache.py +import os +import tempfile +import time +import unittest +from pathlib import Path +from unittest import mock + +import numpy as np + +import servecache as SC + + +class ServeCacheTest(unittest.TestCase): + def setUp(self): + self.tmp = Path(tempfile.mkdtemp()) + self.arrays = {"a": np.arange(10, dtype=np.float32), "ids": np.array([3, 1, 2], np.uint64), + "m": np.array([True, False]), "xyz": np.random.default_rng(1).normal(size=(5, 3))} + + def test_round_trip_keeps_values_and_dtypes(self): + d = SC.cache_dir(self.tmp, "abc") + self.assertEqual(d.name, f"serve-abc-v{SC.VERSION}") + SC.write(d, self.arrays, {"note": "x"}) + arrays, meta = SC.read(d) + self.assertEqual(meta["note"], "x") + for k, v in self.arrays.items(): + self.assertIsInstance(arrays[k], np.memmap) + self.assertEqual(arrays[k].dtype, v.dtype) + np.testing.assert_array_equal(arrays[k], v) + + def test_read_loads_more_arrays_than_the_soft_file_limit(self): + import resource + soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) + if hard != resource.RLIM_INFINITY and hard < 400: + self.skipTest("hard open-file limit too low") + d = SC.cache_dir(self.tmp, "many") + SC.write(d, {f"f{i}": np.arange(3) + i for i in range(300)}, {}) + got = None + try: + resource.setrlimit(resource.RLIMIT_NOFILE, (128, hard)) # each memory map keeps a file open + got = SC.read(d) + self.assertIsNotNone(got, "a complete cache must not read as damaged") + self.assertEqual(int(got[0]["f299"][2]), 301) + finally: + del got + resource.setrlimit(resource.RLIMIT_NOFILE, (soft, hard)) + + def test_read_ignores_a_folder_without_done(self): + d = SC.cache_dir(self.tmp, "abc") + SC.write(d, self.arrays, {}) + (d / "done").unlink() + self.assertIsNone(SC.read(d)) + self.assertIsNone(SC.read(self.tmp / "missing")) + + def test_failed_write_leaves_no_final_folder(self): + d = SC.cache_dir(self.tmp, "abc") + with mock.patch.object(SC, "save_npy", side_effect=OSError("disk full")): + with self.assertRaises(OSError): + SC.write(d, self.arrays, {}) + self.assertFalse(d.exists()) + self.assertEqual(list(self.tmp.iterdir()), [], "the temporary folder is removed too") + + def test_load_or_build_builds_once(self): + d = SC.cache_dir(self.tmp, "abc") + calls = [] + def build(): + calls.append(1) + return self.arrays, {"n": 1} + a1, _ = SC.load_or_build(d, build) + a2, meta = SC.load_or_build(d, build) + self.assertEqual(len(calls), 1) + self.assertEqual(meta["n"], 1) + self.assertIsInstance(a1["a"], np.memmap, "even the first load hands back the written maps") + np.testing.assert_array_equal(a2["xyz"], self.arrays["xyz"]) + + def test_unwritable_falls_back_to_ram(self): + d = SC.cache_dir(self.tmp, "abc") + logs = [] + with mock.patch.object(SC, "write", side_effect=OSError("read-only file system")): + arrays, _ = SC.load_or_build(d, lambda: (self.arrays, {}), log=logs.append) + self.assertIs(arrays["a"], self.arrays["a"]) + self.assertEqual(len(logs), 1) + self.assertIn("read-only file system", logs[0]) + + def test_prune_removes_only_unkept_serve_folders(self): + keep = SC.cache_dir(self.tmp, "new") + old = SC.cache_dir(self.tmp, "old") + for d in (keep, old): + SC.write(d, self.arrays, {}) + area = self.tmp / "0123abcd" + area.mkdir() + (area / "cells.npz").write_bytes(b"x") + tiles = self.tmp / "v4-abc" + tiles.mkdir() + gone = SC.prune(self.tmp, [keep]) + self.assertEqual(gone, [old]) + self.assertTrue(keep.exists() and area.exists() and tiles.exists()) + + def test_prune_keeps_young_temporary_folders_and_removes_old_ones(self): + young = self.tmp / f"serve-x-v{SC.VERSION}.tmp-1" + stale = self.tmp / f"serve-y-v{SC.VERSION}.tmp-2" + young.mkdir() + stale.mkdir() + t = time.time() - 7200 + os.utime(stale, (t, t)) + SC.prune(self.tmp, []) + self.assertTrue(young.exists(), "may belong to a server still writing it") + self.assertFalse(stale.exists()) + + def test_prune_removes_a_young_temporary_folder_of_a_process_that_is_gone(self): + import subprocess + pr = subprocess.Popen(["true"]) + pr.wait() + orphan = self.tmp / f"serve-z-v{SC.VERSION}.tmp-{pr.pid}" + orphan.mkdir() + SC.prune(self.tmp, []) + self.assertFalse(orphan.exists(), "its writer died: nobody finishes it") + + def test_prune_skips_a_folder_another_process_removed_meanwhile(self): + gone = self.tmp / f"serve-q-v{SC.VERSION}.tmp-1" + with mock.patch.object(Path, "iterdir", lambda _: iter([gone])), \ + mock.patch.object(Path, "is_dir", lambda _: True): + self.assertEqual(SC.prune(self.tmp, []), []) + + def test_another_server_finishing_first_is_no_failure(self): + import errno + d = SC.cache_dir(self.tmp, "race") + real = os.rename + + def other_first(src, dst): + with mock.patch.object(SC.os, "rename", real): + SC.write(Path(dst), self.arrays, {}) + raise OSError(errno.ENOTEMPTY, "Directory not empty") + logs = [] + with mock.patch.object(SC.os, "rename", other_first): + arrays, _ = SC.load_or_build(d, lambda: (self.arrays, {}), log=logs.append) + self.assertEqual(logs, [], "no RAM fallback") + self.assertIsInstance(arrays["a"], np.memmap) + self.assertFalse([p for p in self.tmp.iterdir() if ".tmp-" in p.name]) + + def test_a_damaged_cache_is_rebuilt(self): + for damage in ("empty", "truncated"): + d = SC.cache_dir(self.tmp, damage) + SC.write(d, self.arrays, {}) + f = d / "xyz.npy" + f.write_bytes(b"" if damage == "empty" else f.read_bytes()[:200]) # e.g. power lost after the rename + logs = [] + arrays, _ = SC.load_or_build(d, lambda: (self.arrays, {}), log=logs.append) + self.assertIsInstance(arrays["xyz"], np.memmap, damage) + np.testing.assert_array_equal(arrays["xyz"], self.arrays["xyz"]) + self.assertIsNotNone(SC.read(d), f"{damage}: healed for the next start") + + +if __name__ == "__main__": + unittest.main() + + +class SaveNpyTest(unittest.TestCase): + def test_same_bytes_as_np_save(self): + rng = np.random.default_rng(0) + arrays = [rng.normal(size=3_000_000), rng.random((1000, 3)), np.zeros(0), np.array(3.5), + rng.integers(0, 9, 50_000).astype(np.int8), rng.random(10_000) < 0.5, + np.arange(12, dtype=np.uint64).reshape(3, 4)[:, ::2], np.asfortranarray(rng.random((300, 7)))] + with tempfile.TemporaryDirectory() as t: + for i, a in enumerate(arrays): + with self.subTest(i=i, dtype=str(a.dtype), shape=a.shape): + np.save(Path(t) / "want.npy", np.ascontiguousarray(a), allow_pickle=False) + SC.save_npy(Path(t) / "got.npy", a) + self.assertEqual((Path(t) / "got.npy").read_bytes(), (Path(t) / "want.npy").read_bytes()) + + def test_written_in_plain_writes_not_tofile(self): + """numpy's tofile() path preallocates the file; btrfs then stores it uncompressed.""" + from numpy.lib import format as F + seen = [] + real = F.write_array + def spy(fp, *a, **k): + seen.append(F.isfileobj(fp)) + return real(fp, *a, **k) + with tempfile.TemporaryDirectory() as t, mock.patch.object(F, "write_array", spy): + SC.save_npy(Path(t) / "a.npy", np.ones(1000)) + self.assertTrue(np.array_equal(np.load(Path(t) / "a.npy"), np.ones(1000))) + self.assertEqual(seen, [False]) diff --git a/tests/test_tilecap.py b/tests/test_tilecap.py new file mode 100644 index 0000000..fbcb5e7 --- /dev/null +++ b/tests/test_tilecap.py @@ -0,0 +1,81 @@ +import os +import shutil +import tempfile +import time +import unittest +from pathlib import Path + +from tilecap import TileCap, tile_key + + +class TileCapTest(unittest.TestCase): + def setUp(self): + self.root = Path(tempfile.mkdtemp()) + + def tearDown(self): + shutil.rmtree(self.root) + + def put(self, rel, size, age): + f = self.root / "v5.4-abc" / rel + f.parent.mkdir(parents=True, exist_ok=True) + f.write_bytes(b"x" * size) + t = time.time() - age + os.utime(f, (t, t)) + return f + + def test_tile_key(self): + self.assertEqual(tile_key("t/v5/relief/9/12/34.jpg"), ("relief", 9)) + self.assertEqual(tile_key("t/v5/mesh/12/1/2.bin"), ("mesh", 12)) + self.assertIsNone(tile_key("t/v5/manifest.npz")) + self.assertIsNone(tile_key("t/v5/relief/9/12/tmpab.part")) + self.assertIsNone(tile_key("t/v5/relief/9/x/3.jpg")) + + def test_sweep_deletes_least_recently_used_down_to_slack(self): + # 10 deep tiles of 100 B, the oldest first; cap 700 → goal 630: the 4 oldest go (1000 → 600) + fs = [self.put(f"relief/12/{i}/0.jpg", 100, age=100 - i) for i in range(10)] + keep = self.put("relief/5/0/0.jpg", 5000, age=1000) # prerendered: kept and not counted + other = self.put("biomes/5/0/0.jpg", 100, age=1) # another layer at a kept zoom: counted + man = self.put("manifest.npz", 100, age=1000) # not a tile + cap = TileCap(700, keep_z=9) + cap.add_root(self.root) + freed = cap.sweep() + self.assertEqual(freed, 500) # 1100 → 600: the 5 oldest of 11 counted + self.assertEqual([f.exists() for f in fs], [False] * 5 + [True] * 5) + self.assertTrue(keep.exists() and other.exists() and man.exists()) + self.assertEqual(cap.total, 600) + self.assertFalse((self.root / "v5.4-abc" / "relief" / "12" / "0").exists(), "empty folders go") + + def test_used_tile_survives(self): + fs = [self.put(f"relief/12/{i}/0.jpg", 100, age=100 - i) for i in range(4)] + cap = TileCap(300, keep_z=-1) + cap.add_root(self.root) + cap.used(fs[0]) # the oldest is read again + cap.sweep() # 400 → goal 270: two go + self.assertEqual([f.exists() for f in fs], [True, False, False, True]) + + def test_under_cap_deletes_nothing_and_saves_count(self): + fs = [self.put(f"relief/12/{i}/0.jpg", 100, age=10) for i in range(3)] + cap = TileCap(1000) + cap.add_root(self.root) + self.assertEqual(cap.sweep(), 0) + self.assertEqual(cap.total, 300) + cap.saved(self.root / "v5.4-abc" / "relief" / "5" / "0" / "0.jpg", 50) + self.assertEqual(cap.total, 350) + self.assertTrue(all(f.exists() for f in fs)) + + def test_save_past_cap_sweeps(self): + cap = TileCap(250) + cap.add_root(self.root) + cap.sweep() + fs = [self.put(f"mesh/13/{i}/0.bin", 100, age=10 - i) for i in range(3)] + started = [] + cap.start = lambda: started.append(1) + cap.saved(fs[0], 100) + cap.saved(fs[1], 100) + self.assertEqual(started, []) + cap.saved(fs[2], 100) + self.assertEqual(started, [1]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tiles.py b/tests/test_tiles.py new file mode 100644 index 0000000..bf0e618 --- /dev/null +++ b/tests/test_tiles.py @@ -0,0 +1,1383 @@ +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) diff --git a/tests/test_viewer_fixtures.py b/tests/test_viewer_fixtures.py new file mode 100644 index 0000000..5c0b88b --- /dev/null +++ b/tests/test_viewer_fixtures.py @@ -0,0 +1,23 @@ +import json +import unittest +from pathlib import Path + +import numpy as np + +from mapgen import projections as PJ + +FIX = Path(__file__).resolve().parents[1] / "viewer" / "tests" / "fixtures" / "projections.json" +POINTS = [(0.0, 0.0), (30.0, 120.0), (-60.0, -170.0), (89.0, 45.0), (10.0, -179.0), (-45.5, 33.3)] + + +class ViewerFixtureTest(unittest.TestCase): + def test_write_projection_fixture(self): + lat = np.array([p[0] for p in POINTS]) + lon = np.array([p[1] for p in POINTS]) + x, y = PJ.equal_earth_forward(lat, lon) + data = {"ee_xmax": float(PJ.EE_XMAX), "ee_ymax": float(PJ.EE_YMAX), + "equal_earth": [{"lat": float(a), "lon": float(b), "x": float(c), "y": float(d)} + for a, b, c, d in zip(lat, lon, x, y)]} + FIX.parent.mkdir(parents=True, exist_ok=True) + FIX.write_text(json.dumps(data, indent=1)) + self.assertEqual(len(json.loads(FIX.read_text())["equal_earth"]), len(POINTS)) |