test_memmap.py 7.4 KB

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