worldmap-viewer: initial public history

1 files changed, 95 insertions, 0 deletions

diff --git a/tests/test_dedupe.py b/tests/test_dedupe.py
new file mode 100644
index 0000000..2521f95
--- /dev/null
+++ b/tests/test_dedupe.py
@@ -0,0 +1,95 @@
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+import numpy as np
+
+import dedupe
+import servecache as SC
+
+B = dedupe.BLOCK
+
+
+class SameRunsTest(unittest.TestCase):
+    def test_runs_of_identical_whole_blocks(self):
+        rng = np.random.default_rng(0)
+        x = rng.integers(0, 256, 10 * B + 100, dtype=np.uint8)
+        y = x.copy()
+        y[2 * B + 5] ^= 1                      # block 2 differs
+        y[3 * B] ^= 1                          # block 3 differs
+        y[7 * B + B - 1] ^= 1                  # block 7 differs (its last byte)
+        with tempfile.TemporaryDirectory() as t:
+            a, b = Path(t) / "a", Path(t) / "b"
+            a.write_bytes(x.tobytes())
+            b.write_bytes(y.tobytes())
+            fa, fb = os.open(a, os.O_RDONLY), os.open(b, os.O_RDONLY)
+            try:
+                runs = dedupe.same_runs(fa, fb, len(x))
+            finally:
+                os.close(fa)
+                os.close(fb)
+        self.assertEqual(runs, [(0, 2 * B), (4 * B, 3 * B), (8 * B, 2 * B)])   # the 100-byte tail is never shared
+
+    def test_dedupe_asks_for_each_identical_run_in_bounded_calls(self):
+        x = np.zeros(40 * B, np.uint8)
+        y = x.copy()
+        y[5 * B] = 1
+        calls = []
+        with tempfile.TemporaryDirectory() as t, \
+                mock.patch.object(dedupe, "_ioctl", lambda a, b, o, n: calls.append((o, n)) or n), \
+                mock.patch.object(dedupe, "MAX_CALL", 8 * B):
+            (Path(t) / "a").write_bytes(x.tobytes())
+            (Path(t) / "b").write_bytes(y.tobytes())
+            shared = dedupe.dedupe_file(Path(t) / "a", Path(t) / "b")
+        self.assertEqual(calls, [(0, 5 * B), (6 * B, 8 * B), (14 * B, 8 * B), (22 * B, 8 * B), (30 * B, 8 * B),
+                                 (38 * B, 2 * B)])
+        self.assertEqual(shared, 39 * B)
+
+    def test_unsupported_file_system_changes_nothing(self):
+        with tempfile.TemporaryDirectory() as t:
+            data = os.urandom(3 * B)
+            for n in "ab":
+                (Path(t) / n).write_bytes(data)
+            with mock.patch.object(dedupe, "_ioctl", side_effect=OSError(95, "Operation not supported")):
+                self.assertEqual(dedupe.dedupe_file(Path(t) / "a", Path(t) / "b"), 0)
+            self.assertEqual((Path(t) / "b").read_bytes(), data)
+
+    def test_different_sizes_are_skipped(self):
+        with tempfile.TemporaryDirectory() as t:
+            (Path(t) / "a").write_bytes(bytes(2 * B))
+            (Path(t) / "b").write_bytes(bytes(3 * B))
+            with mock.patch.object(dedupe, "_ioctl", side_effect=AssertionError):
+                self.assertEqual(dedupe.dedupe_file(Path(t) / "a", Path(t) / "b"), 0)
+
+
+class PeersTest(unittest.TestCase):
+    def test_a_peer_vanishing_midway_is_skipped(self):
+        with tempfile.TemporaryDirectory() as t:
+            new = Path(t) / "new"
+            new.mkdir()
+            (new / "x.npy").write_bytes(bytes(B))
+            with mock.patch.object(dedupe, "dedupe_file", side_effect=FileNotFoundError):
+                self.assertEqual(dedupe.dedupe_dirs(new, [Path(t) / "gone"]), 0)
+            with mock.patch.object(Path, "is_file", return_value=True), \
+                    mock.patch.object(dedupe, "dedupe_file", side_effect=FileNotFoundError):
+                self.assertEqual(dedupe.dedupe_dirs(new, [Path(t) / "gone"]), 0)
+
+
+    def test_a_new_era_cache_is_shared_with_the_other_eras_complete_caches(self):
+        with tempfile.TemporaryDirectory() as t:
+            regions = Path(t) / "regions"
+            arrays = {"x": np.arange(10.0)}
+            old = SC.cache_dir(regions / "era-a", "k1")
+            SC.write(old, arrays, {})
+            (regions / "era-c" / f"{SC.PREFIX}half").mkdir(parents=True)       # incomplete: not a peer
+            new = SC.cache_dir(regions / "era-b", "k2")
+            with mock.patch.object(SC.dedupe, "dedupe_dirs") as dd:
+                SC.write(new, arrays, {})
+            dd.assert_called_once_with(new, [old])
+            self.assertTrue(np.array_equal(SC.read(new)[0]["x"], np.arange(10.0)))
+
+
+if __name__ == "__main__":
+    unittest.main()