use unittest.skipIf

Author: alimanfoo

zarr/tests/test_convenience.py

@@ -579,38 +579,42 @@ def test_logging(self):
     h5py = None
 
 
-if h5py is not None:
+def temp_h5f():
+    fn = tempfile.mktemp()
+    atexit.register(os.remove, fn)
+    h5f = h5py.File(fn, mode='w')
+    atexit.register(lambda v: v.close(), h5f)
+    return h5f
 
-    def temp_h5f():
-        fn = tempfile.mktemp()
-        atexit.register(os.remove, fn)
-        h5f = h5py.File(fn, mode='w')
-        atexit.register(lambda v: v.close(), h5f)
-        return h5f
-
-    class TestCopyHDF5ToZarr(TestCopy):
-
-        def __init__(self, *args, **kwargs):
-            super(TestCopyHDF5ToZarr, self).__init__(*args, **kwargs)
-            self.source_h5py = True
-            self.dest_h5py = False
-            self.new_source = temp_h5f
-            self.new_dest = group
-
-    class TestCopyZarrToHDF5(TestCopy):
-
-        def __init__(self, *args, **kwargs):
-            super(TestCopyZarrToHDF5, self).__init__(*args, **kwargs)
-            self.source_h5py = False
-            self.dest_h5py = True
-            self.new_source = group
-            self.new_dest = temp_h5f
-
-    class TestCopyHDF5ToHDF5(TestCopy):
-
-        def __init__(self, *args, **kwargs):
-            super(TestCopyHDF5ToHDF5, self).__init__(*args, **kwargs)
-            self.source_h5py = True
-            self.dest_h5py = True
-            self.new_source = temp_h5f
-            self.new_dest = temp_h5f
+
+@unittest.skipIf(h5py is None, 'h5py is not installed')
+class TestCopyHDF5ToZarr(TestCopy):
+
+    def __init__(self, *args, **kwargs):
+        super(TestCopyHDF5ToZarr, self).__init__(*args, **kwargs)
+        self.source_h5py = True
+        self.dest_h5py = False
+        self.new_source = temp_h5f
+        self.new_dest = group
+
+
+@unittest.skipIf(h5py is None, 'h5py is not installed')
+class TestCopyZarrToHDF5(TestCopy):
+
+    def __init__(self, *args, **kwargs):
+        super(TestCopyZarrToHDF5, self).__init__(*args, **kwargs)
+        self.source_h5py = False
+        self.dest_h5py = True
+        self.new_source = group
+        self.new_dest = temp_h5f
+
+
+@unittest.skipIf(h5py is None, 'h5py is not installed')
+class TestCopyHDF5ToHDF5(TestCopy):
+
+    def __init__(self, *args, **kwargs):
+        super(TestCopyHDF5ToHDF5, self).__init__(*args, **kwargs)
+        self.source_h5py = True
+        self.dest_h5py = True
+        self.new_source = temp_h5f
+        self.new_dest = temp_h5f

zarr/tests/test_core.py

@@ -1251,24 +1251,23 @@ def test_nbytes_stored(self):
     bsddb3 = None
 
 
-if bsddb3 is not None:
+@unittest.skipIf(bsddb3 is None, 'bsddb3 is not installed')
+class TestArrayWithDBMStoreBerkeleyDB(TestArray):
 
-    class TestArrayWithDBMStoreBerkeleyDB(TestArray):
-
-        @staticmethod
-        def create_array(read_only=False, **kwargs):
-            path = mktemp(suffix='.dbm')
-            atexit.register(os.remove, path)
-            store = DBMStore(path, flag='n', open=bsddb3.btopen)
-            cache_metadata = kwargs.pop('cache_metadata', True)
-            cache_attrs = kwargs.pop('cache_attrs', True)
-            kwargs.setdefault('compressor', Zlib(1))
-            init_array(store, **kwargs)
-            return Array(store, read_only=read_only, cache_metadata=cache_metadata,
-                         cache_attrs=cache_attrs)
+    @staticmethod
+    def create_array(read_only=False, **kwargs):
+        path = mktemp(suffix='.dbm')
+        atexit.register(os.remove, path)
+        store = DBMStore(path, flag='n', open=bsddb3.btopen)
+        cache_metadata = kwargs.pop('cache_metadata', True)
+        cache_attrs = kwargs.pop('cache_attrs', True)
+        kwargs.setdefault('compressor', Zlib(1))
+        init_array(store, **kwargs)
+        return Array(store, read_only=read_only, cache_metadata=cache_metadata,
+                     cache_attrs=cache_attrs)
 
-        def test_nbytes_stored(self):
-            pass  # not implemented
+    def test_nbytes_stored(self):
+        pass  # not implemented
 
 
 try:
@@ -1277,41 +1276,42 @@ def test_nbytes_stored(self):
     lmdb = None
 
 
-if lmdb is not None:
+@unittest.skipIf(lmdb is None, 'lmdb is not installed')
+class TestArrayWithLMDBStore(TestArray):
 
-    class TestArrayWithLMDBStore(TestArray):
+    @staticmethod
+    def create_array(read_only=False, **kwargs):
+        path = mktemp(suffix='.lmdb')
+        atexit.register(atexit_rmtree, path)
+        store = LMDBStore(path, buffers=True)
+        cache_metadata = kwargs.pop('cache_metadata', True)
+        cache_attrs = kwargs.pop('cache_attrs', True)
+        kwargs.setdefault('compressor', Zlib(1))
+        init_array(store, **kwargs)
+        return Array(store, read_only=read_only, cache_metadata=cache_metadata,
+                     cache_attrs=cache_attrs)
 
-        @staticmethod
-        def create_array(read_only=False, **kwargs):
-            path = mktemp(suffix='.lmdb')
-            atexit.register(atexit_rmtree, path)
-            store = LMDBStore(path, buffers=True)
-            cache_metadata = kwargs.pop('cache_metadata', True)
-            cache_attrs = kwargs.pop('cache_attrs', True)
-            kwargs.setdefault('compressor', Zlib(1))
-            init_array(store, **kwargs)
-            return Array(store, read_only=read_only, cache_metadata=cache_metadata,
-                         cache_attrs=cache_attrs)
+    def test_nbytes_stored(self):
+        pass  # not implemented
 
-        def test_nbytes_stored(self):
-            pass  # not implemented
 
-    class TestArrayWithLMDBStoreNoBuffers(TestArray):
+@unittest.skipIf(lmdb is None, 'lmdb is not installed')
+class TestArrayWithLMDBStoreNoBuffers(TestArray):
 
-        @staticmethod
-        def create_array(read_only=False, **kwargs):
-            path = mktemp(suffix='.lmdb')
-            atexit.register(atexit_rmtree, path)
-            store = LMDBStore(path, buffers=False)
-            cache_metadata = kwargs.pop('cache_metadata', True)
-            cache_attrs = kwargs.pop('cache_attrs', True)
-            kwargs.setdefault('compressor', Zlib(1))
-            init_array(store, **kwargs)
-            return Array(store, read_only=read_only, cache_metadata=cache_metadata,
-                         cache_attrs=cache_attrs)
+    @staticmethod
+    def create_array(read_only=False, **kwargs):
+        path = mktemp(suffix='.lmdb')
+        atexit.register(atexit_rmtree, path)
+        store = LMDBStore(path, buffers=False)
+        cache_metadata = kwargs.pop('cache_metadata', True)
+        cache_attrs = kwargs.pop('cache_attrs', True)
+        kwargs.setdefault('compressor', Zlib(1))
+        init_array(store, **kwargs)
+        return Array(store, read_only=read_only, cache_metadata=cache_metadata,
+                     cache_attrs=cache_attrs)
 
-        def test_nbytes_stored(self):
-            pass  # not implemented
+    def test_nbytes_stored(self):
+        pass  # not implemented
 
 
 class TestArrayWithNoCompressor(TestArray):

zarr/tests/test_hierarchy.py

@@ -892,16 +892,15 @@ def create_store():
     bsddb3 = None
 
 
-if bsddb3 is not None:
+@unittest.skipIf(bsddb3 is None, 'bsddb3 is not installed')
+class TestGroupWithDBMStoreBerkeleyDB(TestGroup):
 
-    class TestGroupWithDBMStoreBerkeleyDB(TestGroup):
-
-        @staticmethod
-        def create_store():
-            path = tempfile.mktemp(suffix='.dbm')
-            atexit.register(os.remove, path)
-            store = DBMStore(path, flag='n', open=bsddb3.btopen)
-            return store, None
+    @staticmethod
+    def create_store():
+        path = tempfile.mktemp(suffix='.dbm')
+        atexit.register(os.remove, path)
+        store = DBMStore(path, flag='n', open=bsddb3.btopen)
+        return store, None
 
 
 try:
@@ -910,16 +909,15 @@ def create_store():
     lmdb = None
 
 
-if lmdb is not None:
+@unittest.skipIf(lmdb is None, 'lmdb is not installed')
+class TestGroupWithLMDBStore(TestGroup):
 
-    class TestGroupWithLMDBStore(TestGroup):
-
-        @staticmethod
-        def create_store():
-            path = tempfile.mktemp(suffix='.lmdb')
-            atexit.register(atexit_rmtree, path)
-            store = LMDBStore(path)
-            return store, None
+    @staticmethod
+    def create_store():
+        path = tempfile.mktemp(suffix='.lmdb')
+        atexit.register(atexit_rmtree, path)
+        store = LMDBStore(path)
+        return store, None
 
 
 class TestGroupWithChunkStore(TestGroup):

zarr/tests/test_storage.py

@@ -90,10 +90,10 @@ def test_writeable_values(self):
         store = self.create_store()
 
         # __setitem__ should accept any value that implements buffer interface
-        store['foo'] = b'bar'
-        store['foo'] = bytearray(b'bar')
-        store['foo'] = array.array('B', b'bar')
-        store['foo'] = np.frombuffer(b'bar', dtype='u1')
+        store['foo1'] = b'bar'
+        store['foo2'] = bytearray(b'bar')
+        store['foo3'] = array.array('B', b'bar')
+        store['foo4'] = np.frombuffer(b'bar', dtype='u1')
 
     def test_update(self):
         store = self.create_store()
@@ -777,32 +777,30 @@ def create_store(self):
     gdbm = None
 
 
-if gdbm is not None:
-
-    class TestDBMStoreGnu(TestDBMStore):
+@unittest.skipIf(gdbm is None, 'gdbm is not installed')
+class TestDBMStoreGnu(TestDBMStore):
 
-        def create_store(self):
-            path = tempfile.mktemp(suffix='.gdbm')
-            atexit.register(os.remove, path)
-            store = DBMStore(path, flag='n', open=gdbm.open, write_lock=False)
-            return store
+    def create_store(self):
+        path = tempfile.mktemp(suffix='.gdbm')
+        atexit.register(os.remove, path)
+        store = DBMStore(path, flag='n', open=gdbm.open, write_lock=False)
+        return store
 
 
-if not PY2:
+if not PY2:  # pragma: py2 no cover
     try:
         import dbm.ndbm as ndbm
     except ImportError:  # pragma: no cover
         ndbm = None
 
-    if ndbm is not None:
-
-        class TestDBMStoreNDBM(TestDBMStore):
+    @unittest.skipIf(ndbm is None, 'ndbm is not installed')
+    class TestDBMStoreNDBM(TestDBMStore):
 
-            def create_store(self):
-                path = tempfile.mktemp(suffix='.ndbm')
-                atexit.register(atexit_rmglob, path + '*')
-                store = DBMStore(path, flag='n', open=ndbm.open)
-                return store
+        def create_store(self):
+            path = tempfile.mktemp(suffix='.ndbm')
+            atexit.register(atexit_rmglob, path + '*')
+            store = DBMStore(path, flag='n', open=ndbm.open)
+            return store
 
 
 try:
@@ -811,15 +809,14 @@ def create_store(self):
     bsddb3 = None
 
 
-if bsddb3 is not None:
+@unittest.skipIf(bsddb3 is None, 'bsddb3 is not installed')
+class TestDBMStoreBerkeleyDB(TestDBMStore):
 
-    class TestDBMStoreBerkeleyDB(TestDBMStore):
-
-        def create_store(self):
-            path = tempfile.mktemp(suffix='.dbm')
-            atexit.register(os.remove, path)
-            store = DBMStore(path, flag='n', open=bsddb3.btopen, write_lock=False)
-            return store
+    def create_store(self):
+        path = tempfile.mktemp(suffix='.dbm')
+        atexit.register(os.remove, path)
+        store = DBMStore(path, flag='n', open=bsddb3.btopen, write_lock=False)
+        return store
 
 
 try:
@@ -828,27 +825,26 @@ def create_store(self):
     lmdb = None
 
 
-if lmdb is not None:
-
-    class TestLMDBStore(StoreTests, unittest.TestCase):
+@unittest.skipIf(lmdb is None, 'lmdb is not installed')
+class TestLMDBStore(StoreTests, unittest.TestCase):
 
-        def create_store(self):
-            path = tempfile.mktemp(suffix='.lmdb')
-            atexit.register(atexit_rmtree, path)
-            if PY2:  # pragma: py3 no cover
-                # don't use buffers, otherwise would have to rewrite tests as bytes and
-                # buffer don't compare equal in PY2
-                buffers = False
-            else:  # pragma: py2 no cover
-                buffers = True
-            store = LMDBStore(path, buffers=buffers)
-            return store
+    def create_store(self):
+        path = tempfile.mktemp(suffix='.lmdb')
+        atexit.register(atexit_rmtree, path)
+        if PY2:  # pragma: py3 no cover
+            # don't use buffers, otherwise would have to rewrite tests as bytes and
+            # buffer don't compare equal in PY2
+            buffers = False
+        else:  # pragma: py2 no cover
+            buffers = True
+        store = LMDBStore(path, buffers=buffers)
+        return store
 
-        def test_context_manager(self):
-            with self.create_store() as store:
-                store['foo'] = b'bar'
-                store['baz'] = b'qux'
-                assert 2 == len(store)
+    def test_context_manager(self):
+        with self.create_store() as store:
+            store['foo'] = b'bar'
+            store['baz'] = b'qux'
+            assert 2 == len(store)
 
 
 class TestLRUStoreCache(StoreTests, unittest.TestCase):