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)