codekingpro/portable-devtools

1import mmap
2import os
3import sys
4import warnings
5from pathlib import Path
6from tempfile import NamedTemporaryFile, TemporaryFile
7
8import pytest
9
10from numpy import (
11    add,
12    allclose,
13    arange,
14    asarray,
15    average,
16    isscalar,
17    memmap,
18    multiply,
19    ndarray,
20    prod,
21    subtract,
22    sum,
23)
24from numpy.testing import (
25    IS_PYPY,
26    assert_,
27    assert_array_equal,
28    assert_equal,
29    break_cycles,
30)
31
32
33@pytest.mark.thread_unsafe(reason="setup & memmap is thread-unsafe (gh-29126)")
34class TestMemmap:
35    def setup_method(self):
36        self.tmpfp = NamedTemporaryFile(prefix='mmap')
37        self.shape = (3, 4)
38        self.dtype = 'float32'
39        self.data = arange(12, dtype=self.dtype)
40        self.data.resize(self.shape)
41
42    def teardown_method(self):
43        self.tmpfp.close()
44        self.data = None
45        if IS_PYPY:
46            break_cycles()
47            break_cycles()
48
49    def test_roundtrip(self):
50        # Write data to file
51        fp = memmap(self.tmpfp, dtype=self.dtype, mode='w+',
52                    shape=self.shape)
53        fp[:] = self.data[:]
54        del fp  # Test __del__ machinery, which handles cleanup
55
56        # Read data back from file
57        newfp = memmap(self.tmpfp, dtype=self.dtype, mode='r',
58                       shape=self.shape)
59        assert_(allclose(self.data, newfp))
60        assert_array_equal(self.data, newfp)
61        assert_equal(newfp.flags.writeable, False)
62
63    def test_open_with_filename(self, tmp_path):
64        tmpname = tmp_path / 'mmap'
65        fp = memmap(tmpname, dtype=self.dtype, mode='w+',
66                       shape=self.shape)
67        fp[:] = self.data[:]
68        del fp
69
70    def test_unnamed_file(self):
71        with TemporaryFile() as f:
72            fp = memmap(f, dtype=self.dtype, shape=self.shape)
73            del fp
74
75    def test_attributes(self):
76        offset = 1
77        mode = "w+"
78        fp = memmap(self.tmpfp, dtype=self.dtype, mode=mode,
79                    shape=self.shape, offset=offset)
80        assert_equal(offset, fp.offset)
81        assert_equal(mode, fp.mode)
82        del fp
83
84    def test_filename(self, tmp_path):
85        tmpname = tmp_path / "mmap"
86        fp = memmap(tmpname, dtype=self.dtype, mode='w+',
87                       shape=self.shape)
88        abspath = Path(os.path.abspath(tmpname))
89        fp[:] = self.data[:]
90        assert_equal(abspath, fp.filename)
91        b = fp[:1]
92        assert_equal(abspath, b.filename)
93        del b
94        del fp
95
96    def test_path(self, tmp_path):
97        tmpname = tmp_path / "mmap"
98        fp = memmap(Path(tmpname), dtype=self.dtype, mode='w+',
99                       shape=self.shape)
100        # os.path.realpath does not resolve symlinks on Windows
101        # see: https://bugs.python.org/issue9949
102        # use Path.resolve, just as memmap class does internally
103        abspath = str(Path(tmpname).resolve())
104        fp[:] = self.data[:]
105        assert_equal(abspath, str(fp.filename.resolve()))
106        b = fp[:1]
107        assert_equal(abspath, str(b.filename.resolve()))
108        del b
109        del fp
110
111    def test_filename_fileobj(self):
112        fp = memmap(self.tmpfp, dtype=self.dtype, mode="w+",
113                    shape=self.shape)
114        assert_equal(fp.filename, self.tmpfp.name)
115
116    @pytest.mark.skipif(sys.platform == 'gnu0',
117                        reason="Known to fail on hurd")
118    def test_flush(self):
119        fp = memmap(self.tmpfp, dtype=self.dtype, mode='w+',
120                    shape=self.shape)
121        fp[:] = self.data[:]
122        assert_equal(fp[0], self.data[0])
123        fp.flush()
124
125    def test_del(self):
126        # Make sure a view does not delete the underlying mmap
127        fp_base = memmap(self.tmpfp, dtype=self.dtype, mode='w+',
128                    shape=self.shape)
129        fp_base[0] = 5
130        fp_view = fp_base[0:1]
131        assert_equal(fp_view[0], 5)
132        del fp_view
133        # Should still be able to access and assign values after
134        # deleting the view
135        assert_equal(fp_base[0], 5)
136        fp_base[0] = 6
137        assert_equal(fp_base[0], 6)
138
139    def test_arithmetic_drops_references(self):
140        fp = memmap(self.tmpfp, dtype=self.dtype, mode='w+',
141                    shape=self.shape)
142        tmp = (fp + 10)
143        if isinstance(tmp, memmap):
144            assert_(tmp._mmap is not fp._mmap)
145
146    def test_indexing_drops_references(self):
147        fp = memmap(self.tmpfp, dtype=self.dtype, mode='w+',
148                    shape=self.shape)
149        tmp = fp[(1, 2), (2, 3)]
150        if isinstance(tmp, memmap):
151            assert_(tmp._mmap is not fp._mmap)
152
153    def test_slicing_keeps_references(self):
154        fp = memmap(self.tmpfp, dtype=self.dtype, mode='w+',
155                    shape=self.shape)
156        assert_(fp[:2, :2]._mmap is fp._mmap)
157
158    def test_view(self):
159        fp = memmap(self.tmpfp, dtype=self.dtype, shape=self.shape)
160        new1 = fp.view()
161        new2 = new1.view()
162        assert_(new1.base is fp)
163        assert_(new2.base is fp)
164        new_array = asarray(fp)
165        assert_(new_array.base is fp)
166
167    def test_ufunc_return_ndarray(self):
168        fp = memmap(self.tmpfp, dtype=self.dtype, shape=self.shape)
169        fp[:] = self.data
170
171        with warnings.catch_warnings():
172            warnings.filterwarnings(
173                'ignore', "np.average currently does not preserve", FutureWarning)
174            for unary_op in [sum, average, prod]:
175                result = unary_op(fp)
176                assert_(isscalar(result))
177                assert_(result.__class__ is self.data[0, 0].__class__)
178
179                assert_(unary_op(fp, axis=0).__class__ is ndarray)
180                assert_(unary_op(fp, axis=1).__class__ is ndarray)
181
182        for binary_op in [add, subtract, multiply]:
183            assert_(binary_op(fp, self.data).__class__ is ndarray)
184            assert_(binary_op(self.data, fp).__class__ is ndarray)
185            assert_(binary_op(fp, fp).__class__ is ndarray)
186
187        fp += 1
188        assert fp.__class__ is memmap
189        add(fp, 1, out=fp)
190        assert fp.__class__ is memmap
191
192    def test_getitem(self):
193        fp = memmap(self.tmpfp, dtype=self.dtype, shape=self.shape)
194        fp[:] = self.data
195
196        assert_(fp[1:, :-1].__class__ is memmap)
197        # Fancy indexing returns a copy that is not memmapped
198        assert_(fp[[0, 1]].__class__ is ndarray)
199
200    def test_memmap_subclass(self):
201        class MemmapSubClass(memmap):
202            pass
203
204        fp = MemmapSubClass(self.tmpfp, dtype=self.dtype, shape=self.shape)
205        fp[:] = self.data
206
207        # We keep previous behavior for subclasses of memmap, i.e. the
208        # ufunc and __getitem__ output is never turned into a ndarray
209        assert_(sum(fp, axis=0).__class__ is MemmapSubClass)
210        assert_(sum(fp).__class__ is MemmapSubClass)
211        assert_(fp[1:, :-1].__class__ is MemmapSubClass)
212        assert fp[[0, 1]].__class__ is MemmapSubClass
213
214    def test_mmap_offset_greater_than_allocation_granularity(self):
215        size = 5 * mmap.ALLOCATIONGRANULARITY
216        offset = mmap.ALLOCATIONGRANULARITY + 1
217        fp = memmap(self.tmpfp, shape=size, mode='w+', offset=offset)
218        assert_(fp.offset == offset)
219
220    def test_empty_array_with_offset_multiple_of_allocation_granularity(self):
221        self.tmpfp.write(b'a' * mmap.ALLOCATIONGRANULARITY)
222        size = 0
223        offset = mmap.ALLOCATIONGRANULARITY
224        fp = memmap(self.tmpfp, shape=size, mode='w+', offset=offset)
225        assert_equal(fp.offset, offset)
226
227    def test_no_shape(self):
228        self.tmpfp.write(b'a' * 16)
229        mm = memmap(self.tmpfp, dtype='float64')
230        assert_equal(mm.shape, (2,))
231
232    def test_empty_array(self):
233        # gh-12653
234        with pytest.raises(ValueError, match='empty file'):
235            memmap(self.tmpfp, shape=(0, 4), mode='r')
236
237        # gh-27723
238        # empty memmap works with mode in ('w+','r+')
239        memmap(self.tmpfp, shape=(0, 4), mode='w+')
240
241        # ok now the file is not empty
242        memmap(self.tmpfp, shape=(0, 4), mode='w+')
243
244    def test_shape_type(self):
245        memmap(self.tmpfp, shape=3, mode='w+')
246        memmap(self.tmpfp, shape=self.shape, mode='w+')
247        memmap(self.tmpfp, shape=list(self.shape), mode='w+')
248        memmap(self.tmpfp, shape=asarray(self.shape), mode='w+')
249