worldmap-viewer

git clone https://git.godosa.eu/worldmap-viewer

master

history zip tar.gz

raw ยท 4163 bytes

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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()