TEST: Update ImageOpener/ArrayProxy unit tests

Author: pauldmccarthy

nibabel/tests/test_arrayproxy.py

@@ -20,8 +20,7 @@
 
 import numpy as np
 
-from ..arrayproxy import (ArrayProxy, KEEP_FILE_OPEN_DEFAULT, is_proxy,
-                          reshape_dataobj)
+from ..arrayproxy import (ArrayProxy, is_proxy, reshape_dataobj)
 from ..openers import ImageOpener
 from ..nifti1 import Nifti1Header
 
@@ -342,15 +341,20 @@ def check_mmap(hdr, offset, proxy_class,
 # An image opener class which counts how many instances of itself have been
 # created
 class CountingImageOpener(ImageOpener):
-
     num_openers = 0
-
     def __init__(self, *args, **kwargs):
-
         super(CountingImageOpener, self).__init__(*args, **kwargs)
         CountingImageOpener.num_openers += 1
 
 
+def _count_ImageOpeners(proxy, data, voxels):
+    CountingImageOpener.num_openers = 0
+    for i in range(voxels.shape[0]):
+        x, y, z = [int(c) for c in voxels[i, :]]
+        assert proxy[x, y, z] == x * 100 + y * 10 + z
+    return CountingImageOpener.num_openers
+
+
 def test_keep_file_open_true_false_invalid():
     # Test the behaviour of the keep_file_open __init__ flag, when it is set to
     # True or False.
@@ -369,18 +373,11 @@ def test_keep_file_open_true_false_invalid():
             proxy_no_kfp = ArrayProxy(fname, ((10, 10, 10), dtype),
                                       keep_file_open=False)
             assert not proxy_no_kfp._keep_file_open
-            for i in range(voxels.shape[0]):
-                x , y, z = [int(c) for c in voxels[i, :]]
-                assert proxy_no_kfp[x, y, z] == x * 100 + y * 10 + z
-                assert CountingImageOpener.num_openers == i + 1
-            CountingImageOpener.num_openers = 0
+            assert _count_ImageOpeners(proxy_no_kfp, data, voxels) == 10
             proxy_kfp = ArrayProxy(fname, ((10, 10, 10), dtype),
                                    keep_file_open=True)
             assert proxy_kfp._keep_file_open
-            for i in range(voxels.shape[0]):
-                x , y, z = [int(c) for c in voxels[i, :]]
-                assert proxy_kfp[x, y, z] == x * 100 + y * 10 + z
-                assert CountingImageOpener.num_openers == 1
+            assert _count_ImageOpeners(proxy_kfp, data, voxels) == 1
             del proxy_kfp
             del proxy_no_kfp
         # Test that the keep_file_open flag has no effect if an open file
@@ -389,20 +386,15 @@ def test_keep_file_open_true_false_invalid():
             for kfo in (True, False, 'auto'):
                 proxy = ArrayProxy(fobj, ((10, 10, 10), dtype),
                                    keep_file_open=kfo)
-                if kfo == 'auto':
-                    kfo = False
-                assert proxy._keep_file_open is kfo
+                assert proxy._keep_file_open is False
                 for i in range(voxels.shape[0]):
+                    x, y, z = [int(c) for c in voxels[i, :]]
                     assert proxy[x, y, z] == x * 100 + y * 10 + z
                     assert not fobj.closed
                 del proxy
                 assert not fobj.closed
         assert fobj.closed
         # Test invalid values of keep_file_open
-        with assert_raises(ValueError):
-            ArrayProxy(fname, ((10, 10, 10), dtype), keep_file_open=0)
-        with assert_raises(ValueError):
-            ArrayProxy(fname, ((10, 10, 10), dtype), keep_file_open=1)
         with assert_raises(ValueError):
             ArrayProxy(fname, ((10, 10, 10), dtype), keep_file_open=55)
         with assert_raises(ValueError):
@@ -411,69 +403,80 @@ def test_keep_file_open_true_false_invalid():
             ArrayProxy(fname, ((10, 10, 10), dtype), keep_file_open='cauto')
 
 
-@contextlib.contextmanager
-def patch_keep_file_open_default(value):
-    # Patch arrayproxy.KEEP_FILE_OPEN_DEFAULT with the given value
-    with mock.patch('nibabel.arrayproxy.KEEP_FILE_OPEN_DEFAULT', value):
-        yield
-
-
 def test_keep_file_open_auto():
     # Test the behaviour of the keep_file_open __init__ flag, when it is set to
-    # 'auto'
+    # 'auto'.
+    # if indexed_gzip is present, the ArrayProxy should persist its ImageOpener.
+    # Otherwise the ArrayProxy should drop openers.
     dtype = np.float32
-    data  = np.arange(1000, dtype=dtype).reshape((10, 10, 10))
+    data = np.arange(1000, dtype=dtype).reshape((10, 10, 10))
+    voxels = np.random.randint(0, 10, (10, 3))
     with InTemporaryDirectory():
         fname  = 'testdata.gz'
         with gzip.open(fname, 'wb') as fobj:
             fobj.write(data.tostring(order='F'))
-        # If have_indexed_gzip, then keep_file_open should be True
-        with patch_indexed_gzip(True):
+        # If have_indexed_gzip, then the arrayproxy should create one
+        # ImageOpener
+        with patch_indexed_gzip(True), \
+             mock.patch('nibabel.arrayproxy.ImageOpener', CountingImageOpener):
+            CountingImageOpener.num_openers = 0
             proxy = ArrayProxy(fname, ((10, 10, 10), dtype),
                                keep_file_open='auto')
-            assert proxy._keep_file_open
+            assert proxy._keep_file_open == 'auto'
+            assert _count_ImageOpeners(proxy, data, voxels) == 1
         # If no have_indexed_gzip, then keep_file_open should be False
-        with patch_indexed_gzip(False):
+        with patch_indexed_gzip(False), \
+             mock.patch('nibabel.arrayproxy.ImageOpener', CountingImageOpener):
+            CountingImageOpener.num_openers = 0
             proxy = ArrayProxy(fname, ((10, 10, 10), dtype),
                                keep_file_open='auto')
-            assert not proxy._keep_file_open
+            assert proxy._keep_file_open is False
+            assert _count_ImageOpeners(proxy, data, voxels) == 10
+
+
+@contextlib.contextmanager
+def patch_keep_file_open_default(value):
+    # Patch arrayproxy.KEEP_FILE_OPEN_DEFAULT with the given value
+    with mock.patch('nibabel.arrayproxy.KEEP_FILE_OPEN_DEFAULT', value):
+        yield
 
 
 def test_keep_file_open_default():
     # Test the behaviour of the keep_file_open __init__ flag, when the
     # arrayproxy.KEEP_FILE_OPEN_DEFAULT value is changed
     dtype = np.float32
-    data  = np.arange(1000, dtype=dtype).reshape((10, 10, 10))
+    data = np.arange(1000, dtype=dtype).reshape((10, 10, 10))
     with InTemporaryDirectory():
         fname  = 'testdata.gz'
         with gzip.open(fname, 'wb') as fobj:
             fobj.write(data.tostring(order='F'))
-        # The default value of KEEP_FILE_OPEN_DEFAULT should cause
-        # keep_file_open to be False, regardless of whether or not indexed_gzip
-        # is present
-        assert KEEP_FILE_OPEN_DEFAULT is False
-        with patch_indexed_gzip(False):
-            proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
-            assert not proxy._keep_file_open
-        with patch_indexed_gzip(True):
-            proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
-            assert not proxy._keep_file_open
-        # KEEP_FILE_OPEN_DEFAULT=True should cause keep_file_open to be True,
-        # regardless of whether or not indexed_gzip is present
-        with patch_keep_file_open_default(True), patch_indexed_gzip(True):
-            proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
-            assert proxy._keep_file_open
-        with patch_keep_file_open_default(True), patch_indexed_gzip(False):
-            proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
-            assert proxy._keep_file_open
-        # KEEP_FILE_OPEN_DEFAULT=auto should cause keep_file_open to be True
-        # or False, depending on whether indeed_gzip is present,
-        with patch_keep_file_open_default('auto'), patch_indexed_gzip(True):
-            proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
-            assert proxy._keep_file_open
-        with patch_keep_file_open_default('auto'), patch_indexed_gzip(False):
-            proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
-            assert not proxy._keep_file_open
+        # If KEEP_FILE_OPEN_DEFAULT is False, ArrayProxy instances should
+        # interpret keep_file_open as False
+        with patch_keep_file_open_default(False):
+            with patch_indexed_gzip(False):
+                proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
+                assert proxy._keep_file_open is False
+            with patch_indexed_gzip(True):
+                proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
+                assert proxy._keep_file_open is False
+        # If KEEP_FILE_OPEN_DEFAULT is True, ArrayProxy instances should
+        # interpret keep_file_open as True
+        with patch_keep_file_open_default(True):
+            with patch_indexed_gzip(False):
+                proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
+                assert proxy._keep_file_open is True
+            with patch_indexed_gzip(True):
+                proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
+                assert proxy._keep_file_open is True
+        # If KEEP_FILE_OPEN_DEFAULT is auto, ArrayProxy instances should
+        # interpret it as auto if indexed_gzip is present, False otherwise.
+        with patch_keep_file_open_default('auto'):
+            with patch_indexed_gzip(False):
+                proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
+                assert proxy._keep_file_open is False
+            with patch_indexed_gzip(True):
+                proxy = ArrayProxy(fname, ((10, 10, 10), dtype))
+                assert proxy._keep_file_open == 'auto'
         # KEEP_FILE_OPEN_DEFAULT=any other value should cuse an error to be
         # raised
         with patch_keep_file_open_default('badvalue'):

nibabel/tests/test_openers.py

@@ -97,9 +97,10 @@ def test_BinOpener():
                       BinOpener, 'test.txt', 'r')
 
 
-class MockIndexedGzipFile(object):
+class MockIndexedGzipFile(GzipFile):
     def __init__(self, *args, **kwargs):
-        pass
+        kwargs.pop('drop_handles', False)
+        super(MockIndexedGzipFile, self).__init__(*args, **kwargs)
 
 
 @contextlib.contextmanager
@@ -112,7 +113,7 @@ def patch_indexed_gzip(state):
         values = (False, False, GzipFile)
     with mock.patch('nibabel.openers.HAVE_INDEXED_GZIP', values[0]), \
          mock.patch('nibabel.arrayproxy.HAVE_INDEXED_GZIP', values[1]), \
-         mock.patch('nibabel.openers.SafeIndexedGzipFile', values[2],
+         mock.patch('nibabel.openers.IndexedGzipFile', values[2],
                     create=True):
         yield