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