From 3443c1c65e9f1753e1e656b35d08416c1fa298f2 Mon Sep 17 00:00:00 2001 From: godosa Date: Wed, 7 Oct 2026 00:14:38 +0200 Subject: worldmap-viewer: initial public history --- tests/test_serve.py | 1346 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1346 insertions(+) create mode 100644 tests/test_serve.py (limited to 'tests/test_serve.py') 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) -- cgit