Merge pull request #1292 from ZedThree/unit-test-work-with-pytest

Allow running unit tests with pytest

Author: jswhit

pyproject.toml

@@ -72,3 +72,6 @@ where = ["src"]
 
 [tool.setuptools.package-data]
 "netCDF4.plugins" = ["lib__nc*"]
+
+[tool.pytest.ini_options]
+pythonpath = ["test"]

test/__init__.py

@@ -0,0 +1,20 @@
+from tempfile import NamedTemporaryFile
+from netCDF4 import (
+    Dataset,
+    __has_zstandard_support__,
+    __has_bzip2_support__,
+    __has_blosc_support__,
+    __has_szip_support__,
+)
+import os
+
+# True if plugins have been disabled
+no_plugins = os.getenv("NO_PLUGINS")
+
+
+with NamedTemporaryFile(suffix=".nc", delete=False) as tf:
+    with Dataset(tf.name, "w") as nc:
+        has_zstd_filter = __has_zstandard_support__ and nc.has_zstd_filter()
+        has_bzip2_filter = __has_bzip2_support__ and nc.has_bzip2_filter()
+        has_blosc_filter = __has_blosc_support__ and nc.has_blosc_filter()
+        has_szip_filter = __has_szip_support__ and nc.has_szip_filter()

test/filter_availability.py

@@ -1,66 +1,13 @@
 import glob, os, sys, unittest, struct, tempfile
-from netCDF4 import getlibversion,__hdf5libversion__,__netcdf4libversion__,__version__, Dataset
-from netCDF4 import __has_cdf5_format__, __has_nc_inq_path__, __has_nc_create_mem__, \
-                    __has_parallel4_support__, __has_pnetcdf_support__, \
-                    __has_zstandard_support__, __has_bzip2_support__, \
-                    __has_blosc_support__,__has_quantization_support__,\
-                    __has_szip_support__
-
+from netCDF4 import __hdf5libversion__,__netcdf4libversion__,__version__, Dataset
 # can also just run
 # python -m unittest discover . 'tst*py'
 
 # Find all test files.
-test_files = glob.glob('tst_*.py')
-if __netcdf4libversion__ < '4.2.1' or __has_parallel4_support__ or __has_pnetcdf_support__:
-    test_files.remove('tst_diskless.py')
-    sys.stdout.write('not running tst_diskless.py ...\n')
-if not __has_nc_inq_path__:
-    test_files.remove('tst_filepath.py')
-    sys.stdout.write('not running tst_filepath.py ...\n')
-if not __has_nc_create_mem__:
-    test_files.remove('tst_create_mem.py')
-    sys.stdout.write('not running tst_create_mem.py ...\n')
-if not __has_cdf5_format__ or struct.calcsize("P") < 8:
-    test_files.remove('tst_cdf5.py')
-    sys.stdout.write('not running tst_cdf5.py ...\n')
-if not __has_quantization_support__:
-    test_files.remove('tst_compression_quant.py')
-    sys.stdout.write('not running tst_compression_quant.py ...\n')
-filename = tempfile.NamedTemporaryFile(suffix='.nc', delete=False).name
-nc = Dataset(filename,'w')
-if not __has_zstandard_support__ or os.getenv('NO_PLUGINS') or not nc.has_zstd_filter():
-    test_files.remove('tst_compression_zstd.py')
-    sys.stdout.write('not running tst_compression_zstd.py ...\n')
-if not __has_bzip2_support__ or os.getenv('NO_PLUGINS') or not nc.has_bzip2_filter():
-    test_files.remove('tst_compression_bzip2.py')
-    sys.stdout.write('not running tst_compression_bzip2.py ...\n')
-if not __has_blosc_support__ or os.getenv('NO_PLUGINS') or not nc.has_blosc_filter():
-    test_files.remove('tst_compression_blosc.py')
-    sys.stdout.write('not running tst_compression_blosc.py ...\n')
-if not __has_szip_support__ or not nc.has_szip_filter():
-    test_files.remove('tst_compression_szip.py')
-    sys.stdout.write('not running tst_compression_szip.py ...\n')
-nc.close()
-os.remove(filename)
-
-# Don't run tests that require network connectivity
-if os.getenv('NO_NET'):
-    test_files.remove('tst_dap.py');
-    sys.stdout.write('not running tst_dap.py ...\n')
-else:
-    # run opendap test first (issue #856).
-    test_files.remove('tst_dap.py')
-    test_files.insert(0,'tst_dap.py')
-
-# Don't run CDL test (that requires ncdump/ncgen)
-if os.getenv('NO_CDL'):
-    test_files.remove('tst_cdl.py');
-    sys.stdout.write('not running tst_cdl.py ...\n')
-
-# Don't run computationally intensive test
-if not os.getenv('MEMORY_LEAK_TEST'):
-    test_files.remove('tst_multiple_open_close.py');
-    sys.stdout.write('not running tst_multiple_open_close.py ...\n')
+test_files = glob.glob('test_*.py')
+# run opendap test first (issue #856).
+test_files.remove('test_dap.py')
+test_files.insert(0,'test_dap.py')
 
 # Build the test suite from the tests found in the test files.
 testsuite = unittest.TestSuite()

test/run_all.py

@@ -0,0 +1,54 @@
+import unittest
+import netCDF4
+from numpy.testing import assert_array_equal
+import numpy as np
+import pathlib
+
+test_dir = pathlib.Path(__file__).parent
+
+
+class Test_Unsigned(unittest.TestCase):
+    """
+    Test autoconversion to unsigned ints when _Unsigned attribute is True.
+    This attribute is is set by netcdf-java to designate unsigned
+    integer data stored with a signed integer type in netcdf-3.
+    If _Unsigned=True, a view to the data as unsigned integers is returned.
+    set_autoscale can be used to turn this off (default is on)
+    See issue #656 (pull request #658).
+    """
+    def test_unsigned(self):
+        with netCDF4.Dataset(test_dir / "ubyte.nc") as f:
+            data = f['ub'][:]
+            assert data.dtype.str[1:] == 'u1'
+            assert_array_equal(data,np.array([0,255],np.uint8))
+            f.set_auto_scale(False)
+            data2 = f['ub'][:]
+            assert data2.dtype.str[1:] == 'i1'
+            assert_array_equal(data2,np.array([0,-1],np.int8))
+            data = f['sb'][:]
+            assert data.dtype.str[1:] == 'i1'
+            # issue #1232 _Unsigned='false' is same as not having _Unsigned set.
+            data = f['sb2'][:]
+            assert data.dtype.str[1:] == 'i1'
+
+        # issue 671
+        with netCDF4.Dataset(test_dir / "issue671.nc") as f:
+            data1 = f['soil_moisture'][:]
+            assert(np.ma.isMA(data1))
+            f.set_auto_scale(False)
+            data2 = f['soil_moisture'][:]
+            assert(data1.mask.sum() == data2.mask.sum())
+
+        # issue 794
+        # test that valid_min/valid_max/_FillValue are
+        # treated as unsigned integers.
+        with netCDF4.Dataset(test_dir / "20171025_2056.Cloud_Top_Height.nc") as f:
+            data = f['HT'][:]
+            assert(data.mask.sum() == 57432)
+            assert(int(data.max()) == 15430)
+            assert(int(data.min()) == 0)
+            assert(data.dtype == np.float32)
+
+
+if __name__ == '__main__':
+    unittest.main()

test/test_Unsigned.py

@@ -5,6 +5,7 @@
 import os
 import tempfile
 import warnings
+import pathlib
 
 import numpy as np
 from collections import OrderedDict
@@ -219,9 +220,8 @@ def runTest(self):
             assert getattr(v1,'nonexistantatt',None) == None
 
         # issue 915 empty string attribute (ncdump reports 'NIL')
-        f = netCDF4.Dataset('test_gold.nc')
-        assert f['RADIANCE'].VAR_NOTES == ""
-        f.close()
+        with netCDF4.Dataset(pathlib.Path(__file__).parent / "test_gold.nc") as f:
+            assert f['RADIANCE'].VAR_NOTES == ""
 
 if __name__ == '__main__':
     unittest.main()

test/tst_alignment.py renamed to test/test_alignment.py

@@ -1,15 +1,17 @@
-from netCDF4 import Dataset
+from netCDF4 import Dataset, __has_cdf5_format__
 import numpy as np
 import sys, os, unittest, tempfile
+import struct
 from numpy.testing import assert_array_equal
 
 FILE_NAME = tempfile.NamedTemporaryFile(suffix='.nc', delete=False).name
 dimsize = np.iinfo(np.int32).max*2 # only allowed in CDF5
 ndim = 100
 arrdata = np.random.randint(np.iinfo(np.uint8).min,np.iinfo(np.uint8).max,size=ndim)
 
-class test_cdf5(unittest.TestCase):
 
+@unittest.skipIf(not __has_cdf5_format__ or struct.calcsize("P") < 8, "no CDF5 support")
+class test_cdf5(unittest.TestCase):
     def setUp(self):
         self.netcdf_file = FILE_NAME
         nc = Dataset(self.netcdf_file,'w',format='NETCDF3_64BIT_DATA')

test/tst_atts.py renamed to test/test_atts.py

@@ -0,0 +1,79 @@
+import unittest
+import netCDF4
+import os
+import pathlib
+
+test_ncdump="""netcdf ubyte {
+dimensions:
+	d = 2 ;
+variables:
+	byte ub(d) ;
+		ub:_Unsigned = "true" ;
+	byte sb(d) ;
+	byte sb2(d) ;
+		sb2:_Unsigned = "false" ;
+
+// global attributes:
+		:_Format = "classic" ;
+}
+"""
+test_ncdump2="""netcdf ubyte {
+dimensions:
+	d = 2 ;
+variables:
+	byte ub(d) ;
+		ub:_Unsigned = "true" ;
+	byte sb(d) ;
+	byte sb2(d) ;
+		sb2:_Unsigned = "false" ;
+
+// global attributes:
+		:_Format = "classic" ;
+data:
+
+ ub = 0, -1 ;
+
+ sb = -128, 127 ;
+
+ sb2 = -127, -127 ;
+}
+"""
+
+
+ubyte_filename = pathlib.Path(__file__).parent / "ubyte.nc"
+
+
+@unittest.skipIf(os.getenv("NO_CDL"), "CDL test disabled")
+class Test_CDL(unittest.TestCase):
+    """
+    Test import/export of CDL
+    """
+
+    def setUp(self):
+        with netCDF4.Dataset(ubyte_filename) as f:
+            f.tocdl(outfile="ubyte.cdl", data=True)
+
+    def test_tocdl(self):
+        # treated as unsigned integers.
+        with netCDF4.Dataset(ubyte_filename) as f:
+            assert f.tocdl() == test_ncdump
+            assert f.tocdl(data=True) == test_ncdump2
+
+    def test_fromcdl(self):
+        with netCDF4.Dataset.fromcdl("ubyte.cdl", ncfilename="ubyte2.nc") as f1:
+            with netCDF4.Dataset(ubyte_filename) as f2:
+                assert f1.variables.keys() == f2.variables.keys()
+                assert f1.filepath() == "ubyte2.nc"
+                assert f1.dimensions.keys() == f2.dimensions.keys()
+                assert len(f1.dimensions["d"]) == len(f2.dimensions["d"])
+                assert (f1["ub"][:] == f2["ub"][:]).all()
+                assert (f1["sb"][:] == f2["sb"][:]).all()
+
+        os.remove("ubyte2.nc")
+
+    def tearDown(self):
+        # Remove the temporary files
+        os.remove('ubyte.cdl')
+
+if __name__ == '__main__':
+    unittest.main()

test/tst_cdf5.py renamed to test/test_cdf5.py

@@ -2,6 +2,8 @@
 from netCDF4 import Dataset
 from numpy.testing import assert_almost_equal
 import os, tempfile, unittest, sys
+from filter_availability import no_plugins, has_blosc_filter
+
 
 ndim = 100000
 iblosc_shuffle=2
@@ -31,8 +33,9 @@ def write_netcdf(filename,dtype='f8',blosc_shuffle=1,complevel=6):
     foo_zstd[:] = datarr
     nc.close()
 
-class CompressionTestCase(unittest.TestCase):
 
+@unittest.skipIf(no_plugins or not has_blosc_filter, "blosc filter not available")
+class CompressionTestCase(unittest.TestCase):
     def setUp(self):
         self.filename = filename
         write_netcdf(self.filename,complevel=iblosc_complevel,blosc_shuffle=iblosc_shuffle)
@@ -73,10 +76,6 @@ def runTest(self):
         assert f.variables['data_zstd'].filters() == dtest
         f.close()
 
+
 if __name__ == '__main__':
-    nc = Dataset(filename,'w')
-    if not nc.has_blosc_filter():
-        sys.stdout.write('blosc filter not available, skipping tests ...\n')
-    else:
-        nc.close()
-        unittest.main()
+    unittest.main()

test/test_cdl.py

@@ -2,6 +2,7 @@
 from netCDF4 import Dataset
 from numpy.testing import assert_almost_equal
 import os, tempfile, unittest, sys
+from filter_availability import no_plugins, has_bzip2_filter
 
 ndim = 100000
 filename1 = tempfile.NamedTemporaryFile(suffix='.nc', delete=False).name
@@ -16,8 +17,9 @@ def write_netcdf(filename,dtype='f8',complevel=6):
     foo[:] = array
     nc.close()
 
-class CompressionTestCase(unittest.TestCase):
 
+@unittest.skipIf(no_plugins or not has_bzip2_filter, "bzip2 filter not available")
+class CompressionTestCase(unittest.TestCase):
     def setUp(self):
         self.filename1 = filename1
         self.filename2 = filename2
@@ -48,10 +50,6 @@ def runTest(self):
         assert(size < 0.96*uncompressed_size)
         f.close()
 
+
 if __name__ == '__main__':
-    nc = Dataset(filename1,'w')
-    if not nc.has_bzip2_filter():
-        sys.stdout.write('bzip2 filter not available, skipping tests ...\n')
-    else:
-        nc.close()
-        unittest.main()
+    unittest.main()

test/tst_chunk_cache.py renamed to test/test_chunk_cache.py

@@ -1,5 +1,5 @@
 from numpy.random.mtrand import uniform
-from netCDF4 import Dataset
+from netCDF4 import Dataset, __has_quantization_support__
 from numpy.testing import assert_almost_equal
 import numpy as np
 import os, tempfile, unittest
@@ -25,8 +25,9 @@ def write_netcdf(filename,zlib,significant_digits,data,dtype='f8',shuffle=False,
     data = file.variables['data'][:]
     file.close()
 
-class CompressionTestCase(unittest.TestCase):
 
+@unittest.skipIf(not __has_quantization_support__, "missing quantisation support")
+class CompressionTestCase(unittest.TestCase):
     def setUp(self):
         self.files = files
         # no compression

test/tst_compound_alignment.py renamed to test/test_compound_alignment.py

@@ -2,6 +2,7 @@
 from netCDF4 import Dataset
 from numpy.testing import assert_almost_equal
 import os, tempfile, unittest, sys
+from filter_availability import has_szip_filter
 
 ndim = 100000
 filename = tempfile.NamedTemporaryFile(suffix='.nc', delete=False).name
@@ -18,8 +19,9 @@ def write_netcdf(filename,dtype='f8'):
     foo_szip[:] = datarr
     nc.close()
 
-class CompressionTestCase(unittest.TestCase):
 
+@unittest.skipIf(not has_szip_filter, "szip filter not available")
+class CompressionTestCase(unittest.TestCase):
     def setUp(self):
         self.filename = filename
         write_netcdf(self.filename)
@@ -38,10 +40,6 @@ def runTest(self):
         assert f.variables['data_szip'].filters() == dtest
         f.close()
 
+
 if __name__ == '__main__':
-    nc = Dataset(filename,'w')
-    if not nc.has_szip_filter():
-        sys.stdout.write('szip filter not available, skipping tests ...\n')
-    else:
-        nc.close()
-        unittest.main()
+    unittest.main()