import contextlib
import io
import shutil
import tempfile
import unittest
from pathlib import Path
from mapgen import config as C
from mapgen.testing import built_world
import deploy
import refine
import servecache
import tiles
class BundleTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.tmp = Path(tempfile.mkdtemp())
cls.root = cls.tmp / "world"
shutil.copytree(built_world(), cls.root) # copy2: keeps mtimes (ns)
out = cls.root / "out" / "r2"
seed = int(C.load(cls.root)[0]["build"]["seed"])
cls.fp = tiles.fingerprint(out, seed)
tag = f"v{tiles.VERSION}.{tiles.DRAW}-{cls.fp}"
for rel, text in {
f"out/r2/regions/{cls.fp}-m{refine.MODEL}/a1/cells.npz": "current area",
"out/r2/regions/0000000000-m7/a1/cells.npz": "an earlier build's area",
"out/r2/regions/areas/k1/cells.npz": "shared area results",
f"out/r2/tiles/{tag}/relief/5/0/0.jpg": "current tile",
f"out/r2/tiles/{tag}+abc/relief/5/0/0.jpg": "current tile, refined set",
"out/r2/tiles/v1.1-0000000000/relief/5/0/0.jpg": "old tile",
f"out/r2/tiles/{tag}/relief/5/0/1.jpg.part": "half written",
"places/pins.json": "{\"pins\": []}",
"places/regions.json": "{\"regions\": []}",
"places/views.md": "notes",
"exports/x.zip": "export",
"lore/true-gods.md": "secret",
"sketch/a.png": "input",
}.items():
f = cls.root / rel
f.parent.mkdir(parents=True, exist_ok=True)
f.write_text(text)
cls.dest = cls.tmp / "dest"
with contextlib.redirect_stdout(io.StringIO()):
cls.rc = deploy.bundle(cls.root, str(cls.dest), 2)
@classmethod
def tearDownClass(cls):
shutil.rmtree(cls.tmp)
def test_copies_what_the_public_server_reads(self):
self.assertEqual(self.rc, 0)
tag = f"v{tiles.VERSION}.{tiles.DRAW}-{self.fp}"
for rel in ("config/world.toml", "places/regions.json", "out/r2/cells.npz", "out/r2/fields.json",
f"out/r2/regions/{self.fp}-m{refine.MODEL}/a1/cells.npz", "out/r2/regions/areas/k1/cells.npz",
f"out/r2/tiles/{tag}/relief/5/0/0.jpg", f"out/r2/tiles/{tag}+abc/relief/5/0/0.jpg"):
self.assertTrue((self.dest / rel).is_file(), rel)
def test_leaves_private_stale_and_input_files_behind(self):
for rel in ("places/pins.json", "places/views.md", "exports", "lore", "sketch",
"out/r2/regions/0000000000-m7", "out/r2/tiles/v1.1-0000000000"):
self.assertFalse((self.dest / rel).exists(), rel)
self.assertFalse(list(self.dest.rglob("*.part")))
def test_the_copy_keeps_the_build_fingerprint(self):
seed = int(C.load(self.dest)[0]["build"]["seed"])
self.assertEqual(tiles.fingerprint(self.dest / "out" / "r2", seed), self.fp)
def test_a_second_bundle_keeps_the_servers_own_tiles(self):
mine = self.dest / "out/r2/tiles/v9.9-ffffffffff/relief/9/1/1.jpg"
mine.parent.mkdir(parents=True, exist_ok=True)
mine.write_text("rendered on the server")
with contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(deploy.bundle(self.root, str(self.dest), 2), 0)
self.assertTrue(mine.exists())
class DedupeWorldTest(unittest.TestCase):
def test_pairs_each_era_cache_with_earlier_peers(self):
tmp = Path(tempfile.mkdtemp())
try:
out = tmp / "out" / "r2"
dirs = [out / "regions" / k / f"{servecache.PREFIX}abc-v1" for k in ("aaa-m7", "bbb-m7", "ccc-m7")]
for d in dirs:
d.mkdir(parents=True)
(d / servecache.DONE).write_text("ok")
calls = []
orig = deploy.dedupe.dedupe_dirs
deploy.dedupe.dedupe_dirs = lambda new, peers: calls.append((new, sorted(peers))) or 7
try:
self.assertEqual(deploy.dedupe_world(tmp, 2), 21)
finally:
deploy.dedupe.dedupe_dirs = orig
self.assertEqual(calls, [(dirs[0], []), (dirs[1], [dirs[0]]), (dirs[2], [dirs[0], dirs[1]])])
finally:
shutil.rmtree(tmp)
if __name__ == "__main__":
unittest.main()