1import numpy as np 2import h5py 3 4from .common import ut, TestCase 5 6 7@ut.skipUnless(h5py.version.hdf5_version_tuple < (1, 9, 178), 'SWMR is available. Skipping backwards compatible tests') 8class TestSwmrNotAvailable(TestCase): 9 """ Test backwards compatibility behaviour when using SWMR functions with 10 an older version of HDF5 which does not have this feature available. 11 Skip this test if SWMR features *are* available in the HDF5 library. 12 """ 13 14 def setUp(self): 15 TestCase.setUp(self) 16 self.data = np.arange(13).astype('f') 17 self.dset = self.f.create_dataset('data', chunks=(13,), maxshape=(None,), data=self.data) 18 19 def test_open_swmr_raises(self): 20 fname = self.f.filename 21 self.f.close() 22 23 with self.assertRaises(ValueError): 24 self.f = h5py.File(fname, 'r', swmr=True) 25 26 def test_refresh_raises(self): 27 """ If the SWMR feature is not available then Dataset.refresh() should throw an AttributeError 28 """ 29 with self.assertRaises(AttributeError): 30 self.dset.refresh() 31 32 def test_flush_raises(self): 33 """ If the SWMR feature is not available the Dataset.flush() should 34 throw an AttributeError 35 """ 36 with self.assertRaises(AttributeError): 37 self.dset.flush() 38 39 def test_swmr_mode_false(self): 40 """ The SWMR getter should just be False 41 """ 42 assert not self.f.swmr_mode 43 44 def test_set_swmr_mode_raises(self): 45 """ If the SWMR feature is not available, setting swmr_mode = True 46 should raise a RuntimeError 47 """ 48 with self.assertRaises(RuntimeError): 49 self.f.swmr_mode = True 50 assert not self.f.swmr_mode 51 52@ut.skipUnless(h5py.version.hdf5_version_tuple >= (1, 9, 178), 'SWMR requires HDF5 >= 1.9.178') 53class TestDatasetSwmrRead(TestCase): 54 """ Testing SWMR functions when reading a dataset. 55 Skip this test if the HDF5 library does not have the SWMR features. 56 """ 57 58 def setUp(self): 59 TestCase.setUp(self) 60 self.data = np.arange(13).astype('f') 61 self.dset = self.f.create_dataset('data', chunks=(13,), maxshape=(None,), data=self.data) 62 fname = self.f.filename 63 self.f.close() 64 65 self.f = h5py.File(fname, 'r', swmr=True) 66 self.dset = self.f['data'] 67 68 def test_initial_swmr_mode_on(self): 69 """ Verify that the file is initially in SWMR mode""" 70 self.assertTrue(self.f.swmr_mode) 71 72 def test_read_data(self): 73 self.assertArrayEqual(self.dset, self.data) 74 75 def test_refresh(self): 76 self.dset.refresh() 77 78 def test_force_swmr_mode_on_raises(self): 79 """ Verify when reading a file cannot be forcibly switched to swmr mode. 80 When reading with SWMR the file must be opened with swmr=True.""" 81 with self.assertRaises(Exception): 82 self.f.swmr_mode = True 83 self.assertTrue(self.f.swmr_mode) 84 85 def test_force_swmr_mode_off_raises(self): 86 """ Switching SWMR write mode off is only possible by closing the file. 87 Attempts to forcibly switch off the SWMR mode should raise a ValueError. 88 """ 89 with self.assertRaises(ValueError): 90 self.f.swmr_mode = False 91 self.assertTrue(self.f.swmr_mode) 92 93@ut.skipUnless(h5py.version.hdf5_version_tuple >= (1, 9, 178), 'SWMR requires HDF5 >= 1.9.178') 94class TestDatasetSwmrWrite(TestCase): 95 """ Testing SWMR functions when reading a dataset. 96 Skip this test if the HDF5 library does not have the SWMR features. 97 """ 98 99 def setUp(self): 100 """ First setup a file with a small chunked and empty dataset. 101 No data written yet. 102 """ 103 104 # Note that when creating the file, the swmr=True is not required for 105 # write, but libver='latest' is required. 106 self.f = h5py.File(self.mktemp(), 'w', libver='latest') 107 108 self.data = np.arange(4).astype('f') 109 self.dset = self.f.create_dataset('data', shape=(0,), dtype=self.data.dtype, chunks=(2,), maxshape=(None,)) 110 111 112 def test_initial_swmr_mode_off(self): 113 """ Verify that the file is not initially in SWMR mode""" 114 self.assertFalse(self.f.swmr_mode) 115 116 def test_switch_swmr_mode_on(self): 117 """ Switch to SWMR mode and verify """ 118 self.f.swmr_mode = True 119 self.assertTrue(self.f.swmr_mode) 120 121 def test_switch_swmr_mode_off_raises(self): 122 """ Switching SWMR write mode off is only possible by closing the file. 123 Attempts to forcibly switch off the SWMR mode should raise a ValueError. 124 """ 125 self.f.swmr_mode = True 126 self.assertTrue(self.f.swmr_mode) 127 with self.assertRaises(ValueError): 128 self.f.swmr_mode = False 129 self.assertTrue(self.f.swmr_mode) 130 131 def test_extend_dset(self): 132 """ Extend and flush a SWMR dataset 133 """ 134 self.f.swmr_mode = True 135 self.assertTrue(self.f.swmr_mode) 136 137 self.dset.resize( self.data.shape ) 138 self.dset[:] = self.data 139 self.dset.flush() 140 141 # Refresh and read back data for assertion 142 self.dset.refresh() 143 self.assertArrayEqual(self.dset, self.data) 144 145 def test_extend_dset_multiple(self): 146 147 self.f.swmr_mode = True 148 self.assertTrue(self.f.swmr_mode) 149 150 self.dset.resize( (4,) ) 151 self.dset[0:] = self.data 152 self.dset.flush() 153 154 # Refresh and read back 1st data block for assertion 155 self.dset.refresh() 156 self.assertArrayEqual(self.dset, self.data) 157 158 self.dset.resize( (8,) ) 159 self.dset[4:] = self.data 160 self.dset.flush() 161 162 # Refresh and read back 1st data block for assertion 163 self.dset.refresh() 164 self.assertArrayEqual(self.dset[0:4], self.data) 165 self.assertArrayEqual(self.dset[4:8], self.data) 166