TEST: Refactor keep_file_open tests to be more comprehensive

Author: pauldmccarthy

nibabel/tests/test_arrayproxy.py

@@ -349,78 +349,128 @@ def __init__(self, *args, **kwargs):
 
 def _count_ImageOpeners(proxy, data, voxels):
     CountingImageOpener.num_openers = 0
+    # expected data is defined in the test_keep_file_open_* tests
     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
 
 
+@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_true_false_invalid():
     # Test the behaviour of the keep_file_open __init__ flag, when it is set to
     # True or False. Expected behaviour is as follows:
-    # igzip present | keep_file_open | persist ImageOpener | igzip.drop_handles
-    # False         | False          | False               | n/a
-    # False         | True           | True                | n/a
-    # True          | False          | True                | True
-    # True          | True           | True                | False
-    CountingImageOpener.num_openers = 0
-    fname = 'testdata'
+    # keep_open | igzip present    | persist ImageOpener | igzip.drop_handles
+    #           | and is gzip file |                     |
+    # ----------|------------------|---------------------|-------------------
+    # False     | False            | False               | n/a
+    # False     | True             | True                | True
+    # True      | False            | True                | n/a
+    # True      | True             | True                | False
+    # 'auto'    | False            | False               | n/a
+    # 'auto'    | True             | True                | False
+    #
+    # Each test tuple contains:
+    #  - file type - gzipped ('gz') or not ('bin'), or an open file handle
+    #    ('open')
+    #  - keep_file_open value passed to ArrayProxy
+    #  - whether or not indexed_gzip is present
+    #  - expected value for internal ArrayProxy._persist_opener flag
+    #  - expected value for internal ArrayProxy._keep_file_open flag
+    tests = [
+        # open file handle - kfo and have_igzip are both irrelevant
+        ('open', False,  False, False, False),
+        ('open', False,  True,  False, False),
+        ('open', True,   False, False, False),
+        ('open', True,   True,  False, False),
+        ('open', 'auto', False, False, False),
+        ('open', 'auto', True,  False, False),
+        # non-gzip file - have_igzip is irrelevant, decision should be made
+        # solely from kfo flag
+        ('bin', False,  False, False, False),
+        ('bin', False,  True,  False, False),
+        ('bin', True,   False, True,  True),
+        ('bin', True,   True,  True,  True),
+        ('bin', 'auto', False, False, False),
+        ('bin', 'auto', True,  False, False),
+        # gzip file. If igzip is present, we persist the ImageOpener. If kfo
+        # is 'auto':
+        #  - if igzip is present, kfo -> True
+        #  - otherwise, kfo -> False
+        ('gz', False,  False, False, False),
+        ('gz', False,  True,  True,  False),
+        ('gz', True,   False, True,  True),
+        ('gz', True,   True,  True,  True),
+        ('gz', 'auto', False, False, False),
+        ('gz', 'auto', True,  True,  True)]
+
     dtype = np.float32
     data  = np.arange(1000, dtype=dtype).reshape((10, 10, 10))
     voxels = np.random.randint(0, 10, (10, 3))
+
+    for test in tests:
+        filetype, kfo, have_igzip, exp_persist, exp_kfo = test
+        with InTemporaryDirectory(), \
+             mock.patch('nibabel.openers.ImageOpener', CountingImageOpener), \
+             patch_indexed_gzip(have_igzip):
+            fname = 'testdata.{}'.format(filetype)
+            # create the test data file
+            if filetype == 'gz':
+                with gzip.open(fname, 'wb') as fobj:
+                    fobj.write(data.tostring(order='F'))
+            else:
+                with open(fname, 'wb') as fobj:
+                    fobj.write(data.tostring(order='F'))
+            if filetype == 'open':
+                fname = open(fname, 'rb')
+            try:
+                proxy = ArrayProxy(fname, ((10, 10, 10), dtype),
+                                   keep_file_open=kfo)
+                # We also test that we get the same behaviour when the
+                # KEEP_FILE_OPEN_DEFAULT flag is changed
+                with patch_keep_file_open_default(kfo):
+                    proxy_def = ArrayProxy(fname, ((10, 10, 10), dtype))
+                # check internal flags
+                assert proxy._persist_opener == exp_persist
+                assert proxy._keep_file_open == exp_kfo
+                assert proxy_def._persist_opener == exp_persist
+                assert proxy_def._keep_file_open == exp_kfo
+                # check persist_opener behaviour - whether one imageopener is
+                # created for the lifetime of the ArrayProxy, or one is
+                # created on each access
+                if exp_persist:
+                    assert _count_ImageOpeners(proxy, data, voxels) == 1
+                    assert _count_ImageOpeners(proxy_def, data, voxels) == 1
+                else:
+                    assert _count_ImageOpeners(proxy, data, voxels) == 10
+                    assert _count_ImageOpeners(proxy_def, data, voxels) == 10
+                # if indexed_gzip is active, check that the file object was
+                # created correctly - the _opener.fobj will be a
+                # MockIndexedGzipFile, defined in test_openers.py
+                if filetype == 'gz' and have_igzip:
+                    assert proxy._opener.fobj._drop_handles == (not exp_kfo)
+                # if we were using an open file handle, check that the proxy
+                # didn't close it
+                if filetype == 'open':
+                    assert not fname.closed
+            except Exception:
+                print('Failed test', test)
+                raise
+            finally:
+                if filetype == 'open':
+                    fname.close()
+    # Test invalid values of keep_file_open
+    print('testinv')
     with InTemporaryDirectory():
+        fname = 'testdata'
         with open(fname, 'wb') as fobj:
             fobj.write(data.tostring(order='F'))
-        # Without indexed_gzip, test that ArrayProxy(keep_file_open=True) only
-        # creates one ImageOpener, and that ArrayProxy(keep_file_open=False)
-        # creates an ImageOpener on every data access.
-        with mock.patch('nibabel.openers.ImageOpener', CountingImageOpener), \
-             patch_indexed_gzip(False):
-            proxy_no_kfp = ArrayProxy(fname, ((10, 10, 10), dtype),
-                                      keep_file_open=False)
-            assert not proxy_no_kfp._keep_file_open
-            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
-            assert _count_ImageOpeners(proxy_kfp, data, voxels) == 1
-            del proxy_kfp
-            del proxy_no_kfp
-        # With indexed_gzip, test that both ArrayProxy(keep_file_open=True)
-        # and ArrayProxy(keep_file_open=False) only create one ImageOpener,
-        # but that the drop_handles parameter passed to the IndexedGzipFile
-        # is set appropriately
-        with mock.patch('nibabel.openers.ImageOpener', CountingImageOpener), \
-             patch_indexed_gzip(True):
-            proxy_no_kfp = ArrayProxy(fname, ((10, 10, 10), dtype),
-                                      keep_file_open=False)
-            assert proxy_no_kfp._keep_file_open
-            assert _count_ImageOpeners(proxy_no_kfp, data, voxels) == 1
-            # check that the drop_handles flag is set - the fobj attribute
-            # should be a MockIndexedGzipFile, defined in test_openers.
-            assert proxy_no_kfp._opener.fobj._drop_handles
-            proxy_kfp = ArrayProxy(fname, ((10, 10, 10), dtype),
-                                   keep_file_open=True)
-            assert proxy_kfp._keep_file_open
-            assert _count_ImageOpeners(proxy_kfp, data, voxels) == 1
-            assert not proxy_no_kfp._opener.fobj._drop_handles
-            del proxy_kfp
-            del proxy_no_kfp
-        # Test that the keep_file_open flag has no effect if an open file
-        # handle is passed in
-        with open(fname, 'rb') as fobj:
-            for kfo in (True, False, 'auto'):
-                proxy = ArrayProxy(fobj, ((10, 10, 10), dtype),
-                                   keep_file_open=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=55)
         with assert_raises(ValueError):
@@ -429,113 +479,6 @@ def test_keep_file_open_true_false_invalid():
             ArrayProxy(fname, ((10, 10, 10), dtype), keep_file_open='cauto')
 
 
-def test_keep_file_open_auto():
-    # Test the behaviour of the keep_file_open __init__ flag, when it is set to
-    # 'auto'. Expected behaviour is as follows:
-    # igzip present | keep_file_open | persist ImageOpener | igzip.drop_handles
-    # False         | 'auto'         | False               | n/a
-    # True          | 'auto'         | False               | False
-    dtype = np.float32
-    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 no have_indexed_gzip, then a separate ImageOpener should be
-        # created on every access.
-        with patch_indexed_gzip(False), \
-             mock.patch('nibabel.openers.ImageOpener', CountingImageOpener):
-            CountingImageOpener.num_openers = 0
-            proxy = ArrayProxy(fname, ((10, 10, 10), dtype),
-                               keep_file_open='auto')
-            assert not proxy._keep_file_open
-            assert _count_ImageOpeners(proxy, data, voxels) == 10
-        # If have_indexed_gzip, then the arrayproxy should create one
-        # ImageOpener, and the IndexedGzipFile drop_handles parameter should
-        # be set to False, so the file handle stays open.
-        with patch_indexed_gzip(True), \
-             mock.patch('nibabel.openers.ImageOpener', CountingImageOpener):
-            CountingImageOpener.num_openers = 0
-            proxy = ArrayProxy(fname, ((10, 10, 10), dtype),
-                               keep_file_open='auto')
-            assert proxy._keep_file_open
-            assert _count_ImageOpeners(proxy, data, voxels) == 1
-            assert not proxy._opener.fobj._drop_handles
-        # If not a gzip file, keep_file_open should be False
-        fname  = 'testdata'
-        with open(fname, 'wb') as fobj:
-            fobj.write(data.tostring(order='F'))
-        # regardless of whether indexed_gzip is present or not
-        with patch_indexed_gzip(True), \
-             mock.patch('nibabel.openers.ImageOpener', CountingImageOpener):
-            CountingImageOpener.num_openers = 0
-            proxy = ArrayProxy(fname, ((10, 10, 10), dtype),
-                               keep_file_open='auto')
-            assert proxy._keep_file_open is False
-            assert _count_ImageOpeners(proxy, data, voxels) == 10
-        with patch_indexed_gzip(False), \
-             mock.patch('nibabel.openers.ImageOpener', CountingImageOpener):
-            CountingImageOpener.num_openers = 0
-            proxy = ArrayProxy(fname, ((10, 10, 10), dtype),
-                               keep_file_open='auto')
-            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))
-    with InTemporaryDirectory():
-        fname  = 'testdata.gz'
-        with gzip.open(fname, 'wb') as fobj:
-            fobj.write(data.tostring(order='F'))
-        # 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'):
-            assert_raises(ValueError,  ArrayProxy, fname, ((10, 10, 10),
-                                                           dtype))
-        with patch_keep_file_open_default(None):
-            assert_raises(ValueError,  ArrayProxy, fname, ((10, 10, 10),
-                                                           dtype))
-
-
 def test_pickle_lock():
     # Test that ArrayProxy can be pickled, and that thread lock is created