aboutsummaryrefslogtreecommitdiffziptar.gz
path: root/tests/test_serve.py
diff options
context:
space:
mode:
authorgodosa <godosa@godosa.eu>2026-10-07 00:14:38 +0200
committergodosa <godosa@godosa.eu>2026-10-07 00:14:38 +0200
commit3443c1c65e9f1753e1e656b35d08416c1fa298f2 (patch)
tree4e43236f460145a4d75d1b4616dcb7aa6ef08f51 /tests/test_serve.py
downloadworldmap-viewer-3443c1c65e9f1753e1e656b35d08416c1fa298f2.tar.gz
worldmap-viewer-3443c1c65e9f1753e1e656b35d08416c1fa298f2.zip
worldmap-viewer: initial public history
Diffstat (limited to 'tests/test_serve.py')
-rw-r--r--tests/test_serve.py1346
1 files changed, 1346 insertions, 0 deletions
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)