import os import sys import queue import shutil import platform import tempfile import warnings import threading import subprocess from pathlib import Path import tables try: import multiprocessing as mp multiprocessing_imported = True except ImportError: multiprocessing_imported = False import numpy as np import tables as tb from tables.tests import common class OpenFileFailureTestCase(common.PyTablesTestCase): def setUp(self): super().setUp() self.N = len(tb.file._open_files) self.open_files = tb.file._open_files def test01_open_file(self): """Checking opening of a non existing file.""" h5fname = tempfile.mktemp(".h5") with self.assertRaises(IOError): h5file = tb.open_file(h5fname) h5file.close() self.assertEqual(self.N, len(self.open_files)) def test02_open_file(self): """Checking opening of an existing non HDF5 file.""" # create a dummy file h5fname = tempfile.mktemp(".h5") Path(h5fname).write_text("") # Try to open the dummy file try: with self.assertRaises(tb.HDF5ExtError): h5file = tb.open_file(h5fname) h5file.close() self.assertEqual(self.N, len(self.open_files)) finally: Path(h5fname).unlink() def test03_open_file(self): """Checking opening of an existing file with invalid mode.""" # See gh-318 # create a dummy file h5fname = tempfile.mktemp(".h5") h5file = tb.open_file(h5fname, "w") h5file.close() try: # Try to open the dummy file self.assertRaises(ValueError, tb.open_file, h5fname, "ab") finally: Path(h5fname).unlink() class OpenFileTestCase(common.TempFileMixin, common.PyTablesTestCase): def setUp(self): super().setUp() self.populateFile() def populateFile(self): root = self.h5file.root # Create an array self.h5file.create_array(root, "array", [1, 2], title="Array example") self.h5file.create_table( root, "table", {"var1": tb.IntCol()}, "Table example" ) root._v_attrs.testattr = 41 # Create another array object self.h5file.create_array(root, "anarray", [1], "Array title") self.h5file.create_table( root, "atable", {"var1": tb.IntCol()}, "Table title" ) # Create a group object group = self.h5file.create_group(root, "agroup", "Group title") group._v_attrs.testattr = 42 # Create a some objects there array1 = self.h5file.create_array( group, "anarray1", [1, 2, 3, 4, 5, 6, 7], "Array title 1" ) array1.attrs.testattr = 42 self.h5file.create_array(group, "anarray2", [2], "Array title 2") self.h5file.create_table( group, "atable1", {"var1": tb.IntCol()}, "Table title 1" ) ra = np.rec.array([(1, 11, "a")], formats="u1,f4,S1") self.h5file.create_table(group, "atable2", ra, "Table title 2") # Create a lonely group in first level self.h5file.create_group(root, "agroup2", "Group title 2") # Create a new group in the second level group3 = self.h5file.create_group(group, "agroup3", "Group title 3") # Create a new group in the third level self.h5file.create_group(group3, "agroup4", "Group title 4") # Create an array in the root with the same name as one in 'agroup' self.h5file.create_array( root, "anarray1", [1, 2], title="Array example" ) def test00_newFile(self): """Checking creation of a new file.""" self.h5file.create_array( self.h5file.root, "array_new", [1, 2], title="Array example" ) # Get the CLASS attribute of the arr object class_ = self.h5file.root.array.attrs.CLASS self.assertEqual(class_.capitalize(), "Array") def test00_newFile_unicode_filename(self): temp_dir = tempfile.mkdtemp() try: h5fname = str(Path(temp_dir) / "test.h5") with tb.open_file(h5fname, "w") as h5file: self.assertTrue(h5file, tb.File) finally: shutil.rmtree(temp_dir) def test00_newFile_numpy_str_filename(self): temp_dir = tempfile.mkdtemp() try: h5fname = np.str_(Path(temp_dir) / "test.h5") with tb.open_file(h5fname, "w") as h5file: self.assertTrue(h5file, tb.File) finally: shutil.rmtree(temp_dir) def test00_newFile_numpy_unicode_filename(self): temp_dir = tempfile.mkdtemp() try: h5fname = np.str_(Path(temp_dir) / "test.h5") with tb.open_file(h5fname, "w") as h5file: self.assertTrue(h5file, tb.File) finally: shutil.rmtree(temp_dir) def test01_openFile(self): """Checking opening of an existing file.""" # Open the old HDF5 file self._reopen(node_cache_slots=self.node_cache_slots) # Get the CLASS attribute of the arr object title = self.h5file.root.array.get_attr("TITLE") self.assertEqual(title, "Array example") def test01_open_file_pathlib(self): """Checking opening of an existing file.""" self.h5file.close() h5fname = Path(self.h5fname) with tables.open_file(h5fname) as h5file: title = h5file.root.array.get_attr("TITLE") self.assertEqual(title, "Array example") def test02_appendFile(self): """Checking appending objects to an existing file.""" # Append a new array to the existing file self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.create_array( self.h5file.root, "array2", [3, 4], title="Title example 2" ) # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) # Get the CLASS attribute of the arr object title = self.h5file.root.array2.get_attr("TITLE") self.assertEqual(title, "Title example 2") def test02b_appendFile2(self): """Checking appending objects to an existing file ("a" version)""" # Append a new array to the existing file self._reopen(mode="a", node_cache_slots=self.node_cache_slots) self.h5file.create_array( self.h5file.root, "array2", [3, 4], title="Title example 2" ) # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) # Get the CLASS attribute of the arr object title = self.h5file.root.array2.get_attr("TITLE") self.assertEqual(title, "Title example 2") # Begin to raise errors... def test03_appendErrorFile(self): """Checking appending objects to an existing file in "w" mode.""" # Append a new array to the existing file but in write mode # so, the existing file should be deleted! self._reopen(mode="w", node_cache_slots=self.node_cache_slots) self.h5file.create_array( self.h5file.root, "array2", [3, 4], title="Title example 2" ) # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) with self.assertRaises(LookupError): # Try to get the 'array' object in the old existing file self.h5file.root.array def test04a_openErrorFile(self): """Checking opening a non-existing file for reading""" with self.assertRaises(IOError): tb.open_file( "nonexistent.h5", mode="r", node_cache_slots=self.node_cache_slots, ) def test04b_alternateRootFile(self): """Checking alternate root access to the object tree.""" # Open the existent HDF5 file self._reopen( root_uep="/agroup", node_cache_slots=self.node_cache_slots ) # Get the CLASS attribute of the arr object if common.verbose: print("\nFile tree dump:", self.h5file) title = self.h5file.root.anarray1.get_attr("TITLE") # Get the node again, as this can trigger errors in some situations anarray1 = self.h5file.root.anarray1 self.assertIsNotNone(anarray1) self.assertEqual(title, "Array title 1") # This test works well, but HDF5 emits a series of messages that # may lose the user. It is better to deactivate it. def notest04c_alternateRootFile(self): """Checking non-existent alternate root access to the object tree""" with self.assertRaises(RuntimeError): self._reopen( root_uep="/nonexistent", node_cache_slots=self.node_cache_slots ) def test05a_removeGroupRecursively(self): """Checking removing a group recursively.""" # Delete a group with leafs self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) with self.assertRaises(tb.NodeError): self.h5file.remove_node(self.h5file.root.agroup) # This should work now self.h5file.remove_node(self.h5file.root, "agroup", recursive=1) # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) # Try to get the removed object with self.assertRaises(LookupError): self.h5file.root.agroup # Try to get a child of the removed object with self.assertRaises(LookupError): self.h5file.get_node("/agroup/agroup3") def test05b_removeGroupRecursively(self): """Checking removing a group recursively and access to it immediately.""" if common.verbose: print("\n", "-=" * 30) print( "Running %s.test05b_removeGroupRecursively..." % self.__class__.__name__ ) # Delete a group with leafs self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) with self.assertRaises(tb.NodeError): self.h5file.remove_node(self.h5file.root, "agroup") # This should work now self.h5file.remove_node(self.h5file.root, "agroup", recursive=1) # Try to get the removed object with self.assertRaises(LookupError): self.h5file.root.agroup # Try to get a child of the removed object with self.assertRaises(LookupError): self.h5file.get_node("/agroup/agroup3") def test06_removeNodeWithDel(self): """Checking removing a node using ``__delattr__()``""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) with self.assertRaises(AttributeError): # This should fail because there is no *Python attribute* # called ``agroup``. del self.h5file.root.agroup def test06a_removeGroup(self): """Checking removing a lonely group from an existing file.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.remove_node(self.h5file.root, "agroup2") # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) # Try to get the removed object with self.assertRaises(LookupError): self.h5file.root.agroup2 def test06b_removeLeaf(self): """Checking removing Leaves from an existing file.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.remove_node(self.h5file.root, "anarray") # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) # Try to get the removed object with self.assertRaises(LookupError): self.h5file.root.anarray def test06c_removeLeaf(self): """Checking removing Leaves and access it immediately.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.remove_node(self.h5file.root, "anarray") # Try to get the removed object with self.assertRaises(LookupError): self.h5file.root.anarray def test06d_removeLeaf(self): """Checking removing a non-existent node""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # Try to get the removed object with self.assertRaises(LookupError): self.h5file.remove_node(self.h5file.root, "nonexistent") def test06e_removeTable(self): """Checking removing Tables from an existing file.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.remove_node(self.h5file.root, "atable") # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) # Try to get the removed object with self.assertRaises(LookupError): self.h5file.root.atable def test07_renameLeaf(self): """Checking renaming a leave and access it after a close/open.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.rename_node(self.h5file.root.anarray, "anarray2") # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) # Ensure that the new name exists array_ = self.h5file.root.anarray2 self.assertEqual(array_.name, "anarray2") self.assertEqual(array_._v_pathname, "/anarray2") self.assertEqual(array_._v_depth, 1) # Try to get the previous object with the old name with self.assertRaises(LookupError): self.h5file.root.anarray def test07b_renameLeaf(self): """Checking renaming Leaves and accesing them immediately.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.rename_node(self.h5file.root.anarray, "anarray2") # Ensure that the new name exists array_ = self.h5file.root.anarray2 self.assertEqual(array_.name, "anarray2") self.assertEqual(array_._v_pathname, "/anarray2") self.assertEqual(array_._v_depth, 1) # Try to get the previous object with the old name with self.assertRaises(LookupError): self.h5file.root.anarray def test07c_renameLeaf(self): """Checking renaming Leaves and modify attributes after that.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.rename_node(self.h5file.root.anarray, "anarray2") array_ = self.h5file.root.anarray2 array_.attrs.TITLE = "hello" # Ensure that the new attribute has been written correctly self.assertEqual(array_.title, "hello") self.assertEqual(array_.attrs.TITLE, "hello") def test07d_renameLeaf(self): """Checking renaming a Group under a nested group.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.rename_node(self.h5file.root.agroup.anarray2, "anarray3") # Ensure that we can access n attributes in the new group node = self.h5file.root.agroup.anarray3 self.assertEqual(node._v_title, "Array title 2") def test08_renameToExistingLeaf(self): """Checking renaming a node to an existing name.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # Try to get the previous object with the old name with self.assertRaises(tb.NodeError): self.h5file.rename_node(self.h5file.root.anarray, "array") # Now overwrite the destination node. anarray = self.h5file.root.anarray self.h5file.rename_node(anarray, "array", overwrite=True) self.assertNotIn("/anarray", self.h5file) self.assertIs(self.h5file.root.array, anarray) def test08b_renameToNotValidNaturalName(self): """Checking renaming a node to a non-valid natural name""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) with warnings.catch_warnings(): warnings.filterwarnings("error", category=tb.NaturalNameWarning) # Try to get the previous object with the old name with self.assertRaises(tb.NaturalNameWarning): self.h5file.rename_node(self.h5file.root.anarray, "array 2") def test09_renameGroup(self): """Checking renaming a Group and access it after a close/open.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.rename_node(self.h5file.root.agroup, "agroup3") # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) # Ensure that the new name exists group = self.h5file.root.agroup3 self.assertEqual(group._v_name, "agroup3") self.assertEqual(group._v_pathname, "/agroup3") # The children of this group also must be accessible through the # new name path group2 = self.h5file.get_node("/agroup3/agroup3") self.assertEqual(group2._v_name, "agroup3") self.assertEqual(group2._v_pathname, "/agroup3/agroup3") # Try to get the previous object with the old name with self.assertRaises(LookupError): self.h5file.root.agroup # Try to get a child with the old pathname with self.assertRaises(LookupError): self.h5file.get_node("/agroup/agroup3") def test09b_renameGroup(self): """Checking renaming a Group and access it immediately.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.rename_node(self.h5file.root.agroup, "agroup3") # Ensure that the new name exists group = self.h5file.root.agroup3 self.assertEqual(group._v_name, "agroup3") self.assertEqual(group._v_pathname, "/agroup3") # The children of this group also must be accessible through the # new name path group2 = self.h5file.get_node("/agroup3/agroup3") self.assertEqual(group2._v_name, "agroup3") self.assertEqual(group2._v_pathname, "/agroup3/agroup3") # Try to get the previous object with the old name with self.assertRaises(LookupError): self.h5file.root.agroup # Try to get a child with the old pathname with self.assertRaises(LookupError): self.h5file.get_node("/agroup/agroup3") def test09c_renameGroup(self): """Checking renaming a Group and modify attributes afterwards.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.rename_node(self.h5file.root.agroup, "agroup3") # Ensure that we can modify attributes in the new group group = self.h5file.root.agroup3 group._v_attrs.TITLE = "Hello" self.assertEqual(group._v_title, "Hello") self.assertEqual(group._v_attrs.TITLE, "Hello") def test09d_renameGroup(self): """Checking renaming a Group under a nested group.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) self.h5file.rename_node(self.h5file.root.agroup.agroup3, "agroup4") # Ensure that we can access n attributes in the new group group = self.h5file.root.agroup.agroup4 self.assertEqual(group._v_title, "Group title 3") def test09e_renameGroup(self): """Checking renaming a Group with nested groups in the LRU cache.""" # This checks for ticket #126. self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # Load intermediate groups and keep a nested one alive. g = self.h5file.root.agroup.agroup3.agroup4 self.assertIsNotNone(g) self.h5file.rename_node("/", name="agroup", newname="agroup_") # see ticket #126 self.assertNotIn("/agroup_/agroup4", self.h5file) self.assertNotIn("/agroup", self.h5file) for newpath in [ "/agroup_", "/agroup_/agroup3", "/agroup_/agroup3/agroup4", ]: self.assertIn(newpath, self.h5file) self.assertEqual( newpath, self.h5file.get_node(newpath)._v_pathname ) def test10_moveLeaf(self): """Checking moving a leave and access it after a close/open.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) newgroup = self.h5file.create_group("/", "newgroup") self.h5file.move_node(self.h5file.root.anarray, newgroup, "anarray2") # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) # Ensure that the new name exists array_ = self.h5file.root.newgroup.anarray2 self.assertEqual(array_.name, "anarray2") self.assertEqual(array_._v_pathname, "/newgroup/anarray2") self.assertEqual(array_._v_depth, 2) # Try to get the previous object with the old name with self.assertRaises(LookupError): self.h5file.root.anarray def test10b_moveLeaf(self): """Checking moving a leave and access it without a close/open.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) newgroup = self.h5file.create_group("/", "newgroup") self.h5file.move_node(self.h5file.root.anarray, newgroup, "anarray2") # Ensure that the new name exists array_ = self.h5file.root.newgroup.anarray2 self.assertEqual(array_.name, "anarray2") self.assertEqual(array_._v_pathname, "/newgroup/anarray2") self.assertEqual(array_._v_depth, 2) # Try to get the previous object with the old name with self.assertRaises(LookupError): self.h5file.root.anarray def test10c_moveLeaf(self): """Checking moving Leaves and modify attributes after that.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) newgroup = self.h5file.create_group("/", "newgroup") self.h5file.move_node(self.h5file.root.anarray, newgroup, "anarray2") array_ = self.h5file.root.newgroup.anarray2 array_.attrs.TITLE = "hello" # Ensure that the new attribute has been written correctly self.assertEqual(array_.title, "hello") self.assertEqual(array_.attrs.TITLE, "hello") def test10d_moveToExistingLeaf(self): """Checking moving a leaf to an existing name.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # Try to get the previous object with the old name with self.assertRaises(tb.NodeError): self.h5file.move_node( self.h5file.root.anarray, self.h5file.root, "array" ) def test10_2_moveTable(self): """Checking moving a table and access it after a close/open.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) newgroup = self.h5file.create_group("/", "newgroup") self.h5file.move_node(self.h5file.root.atable, newgroup, "atable2") # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) # Ensure that the new name exists table_ = self.h5file.root.newgroup.atable2 self.assertEqual(table_.name, "atable2") self.assertEqual(table_._v_pathname, "/newgroup/atable2") self.assertEqual(table_._v_depth, 2) # Try to get the previous object with the old name with self.assertRaises(LookupError): self.h5file.root.atable def test10_2b_moveTable(self): """Checking moving a table and access it without a close/open.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) newgroup = self.h5file.create_group("/", "newgroup") self.h5file.move_node(self.h5file.root.atable, newgroup, "atable2") # Ensure that the new name exists table_ = self.h5file.root.newgroup.atable2 self.assertEqual(table_.name, "atable2") self.assertEqual(table_._v_pathname, "/newgroup/atable2") self.assertEqual(table_._v_depth, 2) # Try to get the previous object with the old name with self.assertRaises(LookupError): self.h5file.root.atable def test10_2b_bis_moveTable(self): """Checking moving a table and use cached row without a close/open.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) newgroup = self.h5file.create_group("/", "newgroup") # Cache the Row attribute prior to the move row = self.h5file.root.atable.row self.h5file.move_node(self.h5file.root.atable, newgroup, "atable2") # Ensure that the new name exists table_ = self.h5file.root.newgroup.atable2 self.assertEqual(table_.name, "atable2") self.assertEqual(table_._v_pathname, "/newgroup/atable2") self.assertEqual(table_._v_depth, 2) # Ensure that cache Row attribute has been updated row = table_.row self.assertEqual(table_._v_pathname, row.table._v_pathname) nrows = table_.nrows # Add a new row just to make sure that this works row.append() table_.flush() self.assertEqual(table_.nrows, nrows + 1) def test10_2c_moveTable(self): """Checking moving tables and modify attributes after that.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) newgroup = self.h5file.create_group("/", "newgroup") self.h5file.move_node(self.h5file.root.atable, newgroup, "atable2") table_ = self.h5file.root.newgroup.atable2 table_.attrs.TITLE = "hello" # Ensure that the new attribute has been written correctly self.assertEqual(table_.title, "hello") self.assertEqual(table_.attrs.TITLE, "hello") def test10_2d_moveToExistingTable(self): """Checking moving a table to an existing name.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # Try to get the previous object with the old name with self.assertRaises(tb.NodeError): self.h5file.move_node( self.h5file.root.atable, self.h5file.root, "table" ) def test10_2e_moveToExistingTableOverwrite(self): """Checking moving a table to an existing name, overwriting it.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) srcNode = self.h5file.root.atable self.h5file.move_node( srcNode, self.h5file.root, "table", overwrite=True ) dstNode = self.h5file.root.table self.assertIs(srcNode, dstNode) def test11_moveGroup(self): """Checking moving a Group and access it after a close/open.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) newgroup = self.h5file.create_group(self.h5file.root, "newgroup") self.h5file.move_node(self.h5file.root.agroup, newgroup, "agroup3") # Open this file in read-only mode self._reopen(node_cache_slots=self.node_cache_slots) # Ensure that the new name exists group = self.h5file.root.newgroup.agroup3 self.assertEqual(group._v_name, "agroup3") self.assertEqual(group._v_pathname, "/newgroup/agroup3") self.assertEqual(group._v_depth, 2) # The children of this group must also be accessible through the # new name path group2 = self.h5file.get_node("/newgroup/agroup3/agroup3") self.assertEqual(group2._v_name, "agroup3") self.assertEqual(group2._v_pathname, "/newgroup/agroup3/agroup3") self.assertEqual(group2._v_depth, 3) # Try to get the previous object with the old name with self.assertRaises(LookupError): self.h5file.root.agroup # Try to get a child with the old pathname with self.assertRaises(LookupError): self.h5file.get_node("/agroup/agroup3") def test11b_moveGroup(self): """Checking moving a Group and access it immediately.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) newgroup = self.h5file.create_group(self.h5file.root, "newgroup") self.h5file.move_node(self.h5file.root.agroup, newgroup, "agroup3") # Ensure that the new name exists group = self.h5file.root.newgroup.agroup3 self.assertEqual(group._v_name, "agroup3") self.assertEqual(group._v_pathname, "/newgroup/agroup3") self.assertEqual(group._v_depth, 2) # The children of this group must also be accessible through the # new name path group2 = self.h5file.get_node("/newgroup/agroup3/agroup3") self.assertEqual(group2._v_name, "agroup3") self.assertEqual(group2._v_pathname, "/newgroup/agroup3/agroup3") self.assertEqual(group2._v_depth, 3) # Try to get the previous object with the old name with self.assertRaises(LookupError): self.h5file.root.agroup # Try to get a child with the old pathname with self.assertRaises(LookupError): self.h5file.get_node("/agroup/agroup3") def test11c_moveGroup(self): """Checking moving a Group and modify attributes afterwards.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) newgroup = self.h5file.create_group(self.h5file.root, "newgroup") self.h5file.move_node(self.h5file.root.agroup, newgroup, "agroup3") # Ensure that we can modify attributes in the new group group = self.h5file.root.newgroup.agroup3 group._v_attrs.TITLE = "Hello" group._v_attrs.hola = "Hello" self.assertEqual(group._v_title, "Hello") self.assertEqual(group._v_attrs.TITLE, "Hello") self.assertEqual(group._v_attrs.hola, "Hello") def test11d_moveToExistingGroup(self): """Checking moving a group to an existing name.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # Try to get the previous object with the old name with self.assertRaises(tb.NodeError): self.h5file.move_node( self.h5file.root.agroup, self.h5file.root, "agroup2" ) def test11e_moveToExistingGroupOverwrite(self): """Checking moving a group to an existing name, overwriting it.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # agroup2 -> agroup srcNode = self.h5file.root.agroup2 self.h5file.move_node( srcNode, self.h5file.root, "agroup", overwrite=True ) dstNode = self.h5file.root.agroup self.assertIs(srcNode, dstNode) def test12a_moveNodeOverItself(self): """Checking moving a node over itself.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # array -> array srcNode = self.h5file.root.array self.h5file.move_node(srcNode, self.h5file.root, "array") dstNode = self.h5file.root.array self.assertIs(srcNode, dstNode) def test12b_moveGroupIntoItself(self): """Checking moving a group into itself.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) with self.assertRaises(tb.NodeError): # agroup2 -> agroup2/ self.h5file.move_node( self.h5file.root.agroup2, self.h5file.root.agroup2 ) def test13a_copyLeaf(self): """Copying a leaf.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # array => agroup2/ new_node = self.h5file.copy_node( self.h5file.root.array, self.h5file.root.agroup2 ) dstNode = self.h5file.root.agroup2.array self.assertIs(new_node, dstNode) def test13b_copyGroup(self): """Copying a group.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # agroup2 => agroup/ new_node = self.h5file.copy_node( self.h5file.root.agroup2, self.h5file.root.agroup ) dstNode = self.h5file.root.agroup.agroup2 self.assertIs(new_node, dstNode) def test13c_copyGroupSelf(self): """Copying a group into itself.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # agroup2 => agroup2/ new_node = self.h5file.copy_node( self.h5file.root.agroup2, self.h5file.root.agroup2 ) dstNode = self.h5file.root.agroup2.agroup2 self.assertIs(new_node, dstNode) def test13d_copyGroupRecursive(self): """Recursively copying a group.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # agroup => agroup2/ new_node = self.h5file.copy_node( self.h5file.root.agroup, self.h5file.root.agroup2, recursive=True ) dstNode = self.h5file.root.agroup2.agroup self.assertIs(new_node, dstNode) dstChild1 = dstNode.anarray1 self.assertIsNotNone(dstChild1) dstChild2 = dstNode.anarray2 self.assertIsNotNone(dstChild2) dstChild3 = dstNode.agroup3 self.assertIsNotNone(dstChild3) def test13e_copyRootRecursive(self): """Recursively copying the root group into the root of another file.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) h5fname2 = tempfile.mktemp(".h5") h5file2 = tb.open_file( h5fname2, mode="w", node_cache_slots=self.node_cache_slots ) try: # h5file.root => h5file2.root new_node = self.h5file.copy_node( self.h5file.root, h5file2.root, recursive=True ) dstNode = h5file2.root self.assertIs(new_node, dstNode) self.assertIn("/agroup", h5file2) self.assertIn("/agroup/anarray1", h5file2) self.assertIn("/agroup/agroup3", h5file2) finally: h5file2.close() Path(h5fname2).unlink() def test13f_copyRootRecursive(self): """Recursively copying the root group into a group in another file.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) h5fname2 = tempfile.mktemp(".h5") h5file2 = tb.open_file( h5fname2, mode="w", node_cache_slots=self.node_cache_slots ) try: h5file2.create_group("/", "agroup2") # fileh.root => h5file2.root.agroup2 new_node = self.h5file.copy_node( self.h5file.root, h5file2.root.agroup2, recursive=True ) dstNode = h5file2.root.agroup2 self.assertIs(new_node, dstNode) self.assertIn("/agroup2/agroup", h5file2) self.assertIn("/agroup2/agroup/anarray1", h5file2) self.assertIn("/agroup2/agroup/agroup3", h5file2) finally: h5file2.close() Path(h5fname2).unlink() def test13g_copyRootItself(self): """Recursively copying the root group into itself.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) agroup2 = self.h5file.root self.assertIsNotNone(agroup2) # h5file.root => h5file.root self.assertRaises( IOError, self.h5file.copy_node, self.h5file.root, self.h5file.root, recursive=True, ) def test14a_copyNodeExisting(self): """Copying over an existing node.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) with self.assertRaises(tb.NodeError): # agroup2 => agroup self.h5file.copy_node(self.h5file.root.agroup2, newname="agroup") def test14b_copyNodeExistingOverwrite(self): """Copying over an existing node, overwriting it.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # agroup2 => agroup new_node = self.h5file.copy_node( self.h5file.root.agroup2, newname="agroup", overwrite=True ) dstNode = self.h5file.root.agroup self.assertIs(new_node, dstNode) def test14b2_copyNodeExistingOverwrite(self): """Copying over an existing node in other file, overwriting it.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) h5fname2 = tempfile.mktemp(".h5") h5file2 = tb.open_file( h5fname2, mode="w", node_cache_slots=self.node_cache_slots ) try: # file1:/anarray1 => h5fname2:/anarray1 new_node = self.h5file.copy_node( self.h5file.root.agroup.anarray1, newparent=h5file2.root ) # file1:/ => h5fname2:/ new_node = self.h5file.copy_node( self.h5file.root, h5file2.root, overwrite=True, recursive=True ) dstNode = h5file2.root self.assertIs(new_node, dstNode) finally: h5file2.close() Path(h5fname2).unlink() def test14c_copyNodeExistingSelf(self): """Copying over self.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) with self.assertRaises(tb.NodeError): # agroup => agroup self.h5file.copy_node(self.h5file.root.agroup, newname="agroup") def test14d_copyNodeExistingOverwriteSelf(self): """Copying over self, trying to overwrite.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) with self.assertRaises(tb.NodeError): # agroup => agroup self.h5file.copy_node( self.h5file.root.agroup, newname="agroup", overwrite=True ) def test14e_copyGroupSelfRecursive(self): """Recursively copying a group into itself.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) with self.assertRaises(tb.NodeError): # agroup => agroup/ self.h5file.copy_node( self.h5file.root.agroup, self.h5file.root.agroup, recursive=True, ) def test15a_oneStepMove(self): """Moving and renaming a node in a single action.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # anarray1 -> agroup/array srcNode = self.h5file.root.anarray1 self.h5file.move_node(srcNode, self.h5file.root.agroup, "array") dstNode = self.h5file.root.agroup.array self.assertIs(srcNode, dstNode) def test15b_oneStepCopy(self): """Copying and renaming a node in a single action.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # anarray1 => agroup/array new_node = self.h5file.copy_node( self.h5file.root.anarray1, self.h5file.root.agroup, "array" ) dstNode = self.h5file.root.agroup.array self.assertIs(new_node, dstNode) def test16a_fullCopy(self): """Copying full data and user attributes.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # agroup => groupcopy srcNode = self.h5file.root.agroup new_node = self.h5file.copy_node( srcNode, newname="groupcopy", recursive=True ) dstNode = self.h5file.root.groupcopy self.assertIs(new_node, dstNode) self.assertEqual(srcNode._v_attrs.testattr, dstNode._v_attrs.testattr) self.assertEqual( srcNode.anarray1.attrs.testattr, dstNode.anarray1.attrs.testattr ) self.assertEqual(srcNode.anarray1.read(), dstNode.anarray1.read()) def test16b_partialCopy(self): """Copying partial data and no user attributes.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) # agroup => groupcopy srcNode = self.h5file.root.agroup new_node = self.h5file.copy_node( srcNode, newname="groupcopy", recursive=True, copyuserattrs=False, start=0, stop=5, step=2, ) dstNode = self.h5file.root.groupcopy self.assertIs(new_node, dstNode) self.assertFalse(hasattr(dstNode._v_attrs, "testattr")) self.assertFalse(hasattr(dstNode.anarray1.attrs, "testattr")) self.assertEqual( srcNode.anarray1.read()[0:5:2], dstNode.anarray1.read() ) def test16c_fullCopy(self): """Copying full data and user attributes (from file to file).""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) h5fname2 = tempfile.mktemp(".h5") h5file2 = tb.open_file( h5fname2, mode="w", node_cache_slots=self.node_cache_slots ) try: # file1:/ => h5fname2:groupcopy srcNode = self.h5file.root new_node = self.h5file.copy_node( srcNode, h5file2.root, newname="groupcopy", recursive=True ) dstNode = h5file2.root.groupcopy self.assertIs(new_node, dstNode) self.assertEqual( srcNode._v_attrs.testattr, dstNode._v_attrs.testattr ) self.assertEqual( srcNode.agroup.anarray1.attrs.testattr, dstNode.agroup.anarray1.attrs.testattr, ) self.assertEqual( srcNode.agroup.anarray1.read(), dstNode.agroup.anarray1.read() ) finally: h5file2.close() Path(h5fname2).unlink() def test17a_CopyChunkshape(self): """Copying dataset with a chunkshape.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) srcTable = self.h5file.root.table newTable = self.h5file.copy_node( srcTable, newname="tablecopy", chunkshape=11 ) self.assertEqual(newTable.chunkshape, (11,)) self.assertNotEqual(srcTable.chunkshape, newTable.chunkshape) def test17b_CopyChunkshape(self): """Copying dataset with a chunkshape with 'keep' value.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) srcTable = self.h5file.root.table newTable = self.h5file.copy_node( srcTable, newname="tablecopy", chunkshape="keep" ) self.assertEqual(srcTable.chunkshape, newTable.chunkshape) def test17c_CopyChunkshape(self): """Copying dataset with a chunkshape with 'auto' value.""" self._reopen(mode="r+", node_cache_slots=self.node_cache_slots) srcTable = self.h5file.root.table newTable = self.h5file.copy_node( srcTable, newname="tablecopy", chunkshape=11 ) newTable2 = self.h5file.copy_node( newTable, newname="tablecopy2", chunkshape="auto" ) self.assertEqual(srcTable.chunkshape, newTable2.chunkshape) def test18_closedRepr(self): """Representing a closed node as a string.""" self._reopen(node_cache_slots=self.node_cache_slots) for node in [self.h5file.root.agroup, self.h5file.root.anarray]: node._f_close() self.assertIn("closed", str(node)) self.assertIn("closed", repr(node)) def test19_fileno(self): """Checking that the 'fileno()' method works.""" # Open the old HDF5 file self._reopen(mode="r", node_cache_slots=self.node_cache_slots) # Get the file descriptor for this file fd = self.h5file.fileno() if common.verbose: print("Value of fileno():", fd) self.assertGreaterEqual(fd, 0) class NodeCacheOpenFile(OpenFileTestCase): node_cache_slots = tb.parameters.NODE_CACHE_SLOTS open_kwargs = dict(node_cache_slots=node_cache_slots) class NoNodeCacheOpenFile(OpenFileTestCase): node_cache_slots = 0 open_kwargs = dict(node_cache_slots=node_cache_slots) class DictNodeCacheOpenFile(OpenFileTestCase): node_cache_slots = -tb.parameters.NODE_CACHE_SLOTS open_kwargs = dict(node_cache_slots=node_cache_slots) class CheckFileTestCase(common.TempFileMixin, common.PyTablesTestCase): def setUp(self): super().setUp() # Create a regular (text) file self.txtfile = tempfile.mktemp(".h5") self.fileh = open(self.txtfile, "w") self.fileh.write("Hello!") self.fileh.close() def tearDown(self): self.fileh.close() Path(self.txtfile).unlink() super().tearDown() def test00_isHDF5File(self): """Checking tables.is_hdf5_file function (TRUE case)""" # Create a PyTables file (and by so, an HDF5 file) self.h5file.create_array( self.h5file.root, "array", [1, 2], title="Title example" ) # For this method to run, it needs a closed file self.h5file.close() # When file has an HDF5 format, always returns 1 if common.verbose: print( "\nisHDF5File(%s) ==> %d" % (self.h5fname, tb.is_hdf5_file(self.h5fname)) ) self.assertEqual(tb.is_hdf5_file(self.h5fname), 1) def test01_isHDF5File(self): """Checking tables.is_hdf5_file function (FALSE case)""" version = tb.is_hdf5_file(self.txtfile) # When file is not an HDF5 format, always returns 0 or # negative value self.assertLessEqual(version, 0) def test01x_isHDF5File_nonexistent(self): """Identifying a nonexistent HDF5 file.""" self.assertRaises(IOError, tb.is_hdf5_file, "nonexistent") @common.unittest.skipUnless( hasattr(os, "getuid") and os.getuid() != 0, "no UID" ) def test01x_isHDF5File_unreadable(self): """Identifying an unreadable HDF5 file.""" self.h5file.close() Path(self.h5fname).chmod(0) # no permissions at all self.assertRaises(IOError, tb.is_hdf5_file, self.h5fname) def test02_isPyTablesFile(self): """Checking is_pytables_file function (TRUE case)""" # Create a PyTables h5fname self.h5file.create_array( self.h5file.root, "array", [1, 2], title="Title example" ) # For this method to run, it needs a closed h5fname self.h5file.close() version = tb.is_pytables_file(self.h5fname) # When h5fname has a PyTables format, always returns "1.0" string or # greater if common.verbose: print() print("\nPyTables format version number ==> %s" % version) self.assertGreaterEqual(version, "1.0") def test03_isPyTablesFile(self): """Checking is_pytables_file function (FALSE case)""" version = tb.is_pytables_file(self.txtfile) # When file is not a PyTables format, always returns 0 or # negative value if common.verbose: print() print("\nPyTables format version number ==> %s" % version) self.assertIsNone(version) def test04_openGenericHDF5File(self): """Checking opening of a generic HDF5 file.""" # Open an existing generic HDF5 file h5fname = common.test_filename("ex-noattr.h5") with tb.open_file(h5fname, mode="r") as h5file: # Check for some objects inside # A group columns = h5file.get_node("/columns", classname="Group") self.assertEqual(columns._v_name, "columns") # An Array array_ = h5file.get_node(columns, "TDC", classname="Array") self.assertEqual(array_._v_name, "TDC") # The new LRU code defers the appearance of a warning to this point # Here comes an Array of H5T_ARRAY type ui = h5file.get_node(columns, "pressure", classname="Array") self.assertEqual(ui._v_name, "pressure") if common.verbose: print("Array object with type H5T_ARRAY -->", repr(ui)) print("Array contents -->", ui[:]) # A Table table = h5file.get_node("/detector", "table", classname="Table") self.assertEqual(table._v_name, "table") def test04b_UnImplementedOnLoading(self): """Checking failure loading resulting in an ``UnImplemented`` node.""" # ############## Note for developers ############################## # This test fails if you have the line: # # ##return ChildClass(self, childname) # uncomment for debugging # # uncommented in Group.py! # # ################################################################# h5fname = common.test_filename("smpl_unsupptype.h5") with tb.open_file(h5fname) as h5file: with self.assertWarns(UserWarning): node = h5file.get_node("/CompoundChunked") self.assertIsInstance(node, tb.UnImplemented) def test04c_UnImplementedScalar(self): """Checking opening of HDF5 files containing scalar dataset of UnImplemented type.""" with tb.open_file(common.test_filename("scalar.h5")) as h5file: with self.assertWarns(UserWarning): node = h5file.get_node("/variable length string") self.assertIsInstance(node, tb.UnImplemented) def test05_copyUnimplemented(self): """Checking that an UnImplemented object cannot be copied.""" # Open an existing generic HDF5 file h5fname = common.test_filename("smpl_unsupptype.h5") with tb.open_file(h5fname, mode="r") as h5file: self.assertWarns(UserWarning, h5file.get_node, "/CompoundChunked") with warnings.catch_warnings(): warnings.simplefilter("ignore") ui = h5file.get_node("/CompoundChunked") self.assertEqual(ui._v_name, "CompoundChunked") if common.verbose: print("UnImplement object -->", repr(ui)) # Check that it cannot be copied to another file: self.assertWarns(UserWarning, ui.copy, self.h5file.root, "newui") # The next can be used to check the copy of Array objects with H5T_ARRAY # in the future def _test05_copyUnimplemented(self): """Checking that an UnImplemented object cannot be copied.""" # Open an existing generic HDF5 file # We don't need to wrap this in a try clause because # it has already been tried and the warning will not happen again h5fname2 = common.test_filename("ex-noattr.h5") with tb.open_file(h5fname2, mode="r") as h5file2: # An unsupported object (the deprecated H5T_ARRAY type in # Array, from pytables 0.8 on) ui = h5file2.get_node(h5file2.root.columns, "pressure") self.assertEqual(ui._v_name, "pressure") if common.verbose: print("UnImplement object -->", repr(ui)) # Check that it cannot be copied to another file with warnings.catch_warnings(): # Force the userwarning to issue an error warnings.filterwarnings("error", category=UserWarning) with self.assertRaises(UserWarning): ui.copy(self.h5file.root, "newui") @common.unittest.skipIf( tb.file._FILE_OPEN_POLICY == "strict", 'FILE_OPEN_POLICY = "strict"' ) class ThreadingTestCase(common.TempFileMixin, common.PyTablesTestCase): def setUp(self): super().setUp() self.h5file.create_carray( "/", "test_array", tb.Int64Atom(), (200, 300) ) self.h5file.close() def test(self): lock = threading.Lock() def syncronized_open_file(*args, **kwargs): with lock: return tb.open_file(*args, **kwargs) def syncronized_close_file(self, *args, **kwargs): with lock: return self.close(*args, **kwargs) filename = self.h5fname def run(filename, q): try: f = syncronized_open_file(filename, mode="r") arr = f.root.test_array[8:12, 18:22] assert arr.max() == arr.min() == 0 syncronized_close_file(f) except Exception: q.put(sys.exc_info()) else: q.put("OK") threads = [] q = queue.Queue() for i in range(10): t = threading.Thread(target=run, args=(filename, q)) t.start() threads.append(t) for i in range(10): self.assertEqual(q.get(), "OK") for t in threads: t.join() class PythonAttrsTestCase(common.TempFileMixin, common.PyTablesTestCase): """Test interactions of Python attributes and child nodes.""" def test00_attrOverChild(self): """Setting a Python attribute over a child node.""" root = self.h5file.root # Create ``/test`` and overshadow it with ``root.test``. child = self.h5file.create_array(root, "test", [1]) attr = "foobar" self.assertWarns(tb.NaturalNameWarning, setattr, root, "test", attr) self.assertIs(root.test, attr) self.assertIs(root._f_get_child("test"), child) # Now bring ``/test`` again to light. del root.test self.assertIs(root.test, child) # Now there is no *attribute* named ``test``. self.assertRaises(AttributeError, delattr, root, "test") def test01_childUnderAttr(self): """Creating a child node under a Python attribute.""" h5file = self.h5file root = h5file.root # Create ``root.test`` and an overshadowed ``/test``. attr = "foobar" root.test = attr self.assertWarns( tb.NaturalNameWarning, h5file.create_array, root, "test", [1] ) child = h5file.get_node("/test") self.assertIs(root.test, attr) self.assertIs(root._f_get_child("test"), child) # Now bring ``/test`` again to light. del root.test self.assertIs(root.test, child) # Now there is no *attribute* named ``test``. self.assertRaises(AttributeError, delattr, root, "test") def test02_nodeAttrInLeaf(self): """Assigning a ``Node`` value as an attribute to a ``Leaf``.""" h5file = self.h5file array1 = h5file.create_array("/", "array1", [1]) array2 = h5file.create_array("/", "array2", [1]) # This may make the garbage collector work a little. array1.array2 = array2 array2.array1 = array1 # Check the assignments. self.assertIs(array1.array2, array2) self.assertIs(array2.array1, array1) self.assertRaises( tb.NoSuchNodeError, # ``/array1`` is not a group h5file.get_node, "/array1/array2", ) self.assertRaises( tb.NoSuchNodeError, # ``/array2`` is not a group h5file.get_node, "/array2/array3", ) def test03_nodeAttrInGroup(self): """Assigning a ``Node`` value as an attribute to a ``Group``.""" h5file = self.h5file root = h5file.root array = h5file.create_array("/", "array", [1]) # Assign the array to a pair of attributes, # one of them overshadowing the original. root.arrayAlias = array self.assertWarns(tb.NaturalNameWarning, setattr, root, "array", array) # Check the assignments. self.assertIs(root.arrayAlias, array) self.assertIs(root.array, array) self.assertRaises(tb.NoSuchNodeError, h5file.get_node, "/arrayAlias") self.assertIs(h5file.get_node("/array"), array) # Remove the attribute overshadowing the child. del root.array # Now there is no *attribute* named ``array``. self.assertRaises(AttributeError, delattr, root, "array") class StateTestCase(common.TempFileMixin, common.PyTablesTestCase): """Test that ``File`` and ``Node`` operations check their state (open or closed, readable or writable) before proceeding.""" def test00_fileCopyFileClosed(self): """Test copying a closed file.""" self.h5file.close() h5cfname = tempfile.mktemp(suffix=".h5") try: self.assertRaises( tb.ClosedFileError, self.h5file.copy_file, h5cfname ) finally: if Path(h5cfname).is_file(): Path(h5cfname).unlink() def test01_fileCloseClosed(self): """Test closing an already closed file.""" self.h5file.close() try: self.h5file.close() except tb.ClosedFileError: self.fail("could not close an already closed file") def test02_fileFlushClosed(self): """Test flushing a closed file.""" self.h5file.close() self.assertRaises(tb.ClosedFileError, self.h5file.flush) def test03_fileFlushRO(self): """Flushing a read-only file.""" self._reopen("r") try: self.h5file.flush() except tb.FileModeError: self.fail("could not flush a read-only file") def test04_fileCreateNodeClosed(self): """Test creating a node in a closed file.""" self.h5file.close() self.assertRaises( tb.ClosedFileError, self.h5file.create_group, "/", "test" ) def test05_fileCreateNodeRO(self): """Test creating a node in a read-only file.""" self._reopen("r") self.assertRaises( tb.FileModeError, self.h5file.create_group, "/", "test" ) def test06_fileRemoveNodeClosed(self): """Test removing a node from a closed file.""" self.h5file.create_group("/", "test") self.h5file.close() self.assertRaises( tb.ClosedFileError, self.h5file.remove_node, "/", "test" ) def test07_fileRemoveNodeRO(self): """Test removing a node from a read-only file.""" self.h5file.create_group("/", "test") self._reopen("r") self.assertRaises( tb.FileModeError, self.h5file.remove_node, "/", "test" ) def test08_fileMoveNodeClosed(self): """Test moving a node in a closed file.""" self.h5file.create_group("/", "test1") self.h5file.create_group("/", "test2") self.h5file.close() self.assertRaises( tb.ClosedFileError, self.h5file.move_node, "/test1", "/", "test2" ) def test09_fileMoveNodeRO(self): """Test moving a node in a read-only file.""" self.h5file.create_group("/", "test1") self.h5file.create_group("/", "test2") self._reopen("r") self.assertRaises( tb.FileModeError, self.h5file.move_node, "/test1", "/", "test2" ) def test10_fileCopyNodeClosed(self): """Test copying a node in a closed file.""" self.h5file.create_group("/", "test1") self.h5file.create_group("/", "test2") self.h5file.close() self.assertRaises( tb.ClosedFileError, self.h5file.copy_node, "/test1", "/", "test2" ) def test11_fileCopyNodeRO(self): """Test copying a node in a read-only file.""" self.h5file.create_group("/", "test1") self._reopen("r") self.assertRaises( tb.FileModeError, self.h5file.copy_node, "/test1", "/", "test2" ) def test13_fileGetNodeClosed(self): """Test getting a node from a closed file.""" self.h5file.create_group("/", "test") self.h5file.close() self.assertRaises(tb.ClosedFileError, self.h5file.get_node, "/test") def test14_fileWalkNodesClosed(self): """Test walking a closed file.""" self.h5file.create_group("/", "test1") self.h5file.create_group("/", "test2") self.h5file.close() self.assertRaises(tb.ClosedFileError, next, self.h5file.walk_nodes()) def test15_fileAttrClosed(self): """Test setting and deleting a node attribute in a closed file.""" self.h5file.create_group("/", "test") self.h5file.close() self.assertRaises( tb.ClosedFileError, self.h5file.set_node_attr, "/test", "foo", "bar", ) self.assertRaises( tb.ClosedFileError, self.h5file.del_node_attr, "/test", "foo" ) def test16_fileAttrRO(self): """Test setting and deleting a node attribute in a read-only file.""" self.h5file.create_group("/", "test") self.h5file.set_node_attr("/test", "foo", "foo") self._reopen("r") self.assertRaises( tb.FileModeError, self.h5file.set_node_attr, "/test", "foo", "bar" ) self.assertRaises( tb.FileModeError, self.h5file.del_node_attr, "/test", "foo" ) def test17_fileUndoClosed(self): """Test undo operations in a closed file.""" self.h5file.enable_undo() self.h5file.create_group("/", "test2") self.h5file.close() self.assertRaises(tb.ClosedFileError, self.h5file.is_undo_enabled) self.assertRaises(tb.ClosedFileError, self.h5file.get_current_mark) self.assertRaises(tb.ClosedFileError, self.h5file.undo) self.assertRaises(tb.ClosedFileError, self.h5file.disable_undo) def test18_fileUndoRO(self): """Test undo operations in a read-only file.""" self.h5file.enable_undo() self.h5file.create_group("/", "test") self._reopen("r") self.assertEqual(self.h5file._undoEnabled, False) # self.assertRaises(FileModeError, self.h5file.undo) # self.assertRaises(FileModeError, self.h5file.disable_undo) def test19a_getNode(self): """Test getting a child of a closed node.""" g1 = self.h5file.create_group("/", "g1") g2 = self.h5file.create_group("/g1", "g2") # Close this *object* so that it should not be used. g1._f_close() self.assertRaises(tb.ClosedNodeError, g1._f_get_child, "g2") # Getting a node by its closed object is not allowed. self.assertRaises(tb.ClosedNodeError, self.h5file.get_node, g1) # Going through that *node* should reopen it automatically. try: g2_ = self.h5file.get_node("/g1/g2") except tb.ClosedNodeError: self.fail("closed parent group has not been reopened") # Already open nodes should be closed now, but not the new ones. self.assertIs( g2._v_isopen, False, "open child of closed group has not been closed", ) self.assertIs( g2_._v_isopen, True, "open child of closed group has not been closed", ) # And existing closed ones should remain closed, but not the new ones. g1_ = self.h5file.get_node("/g1") self.assertIs( g1._v_isopen, False, "already closed group is not closed anymore" ) self.assertIs( g1_._v_isopen, True, "newly opened group is still closed" ) def test19b_getNode(self): """Test getting a node that does not start with a slash ('/').""" # Create an array in the root self.h5file.create_array("/", "array", [1, 2], title="Title example") # Get the array without specifying a leading slash self.assertRaises(NameError, self.h5file.get_node, "array") def test20_removeNode(self): """Test removing a closed node.""" # This test is a little redundant once we know that ``File.get_node()`` # will reload a closed node, but anyway... group = self.h5file.create_group("/", "group") array = self.h5file.create_array("/group", "array", [1]) # The closed *object* can not be used. group._f_close() self.assertRaises(tb.ClosedNodeError, group._f_remove) self.assertRaises(tb.ClosedNodeError, self.h5file.remove_node, group) # Still, the *node* is reloaded when necessary. try: self.h5file.remove_node("/group", recursive=True) except tb.ClosedNodeError: self.fail("closed node has not been reloaded") # Objects of descendent removed nodes # should have been automatically closed when removed. self.assertRaises(tb.ClosedNodeError, array._f_remove) self.assertNotIn("/group/array", self.h5file) # just in case self.assertNotIn("/group", self.h5file) # just in case def test21_attrsOfNode(self): """Test manipulating the attributes of a closed node.""" node = self.h5file.create_group("/", "test") nodeAttrs = node._v_attrs nodeAttrs.test = attr = "foo" node._f_close() self.assertRaises(tb.ClosedNodeError, getattr, node, "_v_attrs") # The design of ``AttributeSet`` does not yet allow this test. # self.assertRaises(ClosedNodeError, getattr, nodeAttrs, 'test') self.assertEqual(self.h5file.get_node_attr("/test", "test"), attr) def test21b_attrsOfNode(self): """Test manipulating the attributes of a node in a read-only file.""" self.h5file.create_group("/", "test") self.h5file.set_node_attr("/test", "test", "foo") self._reopen("r") self.assertRaises( tb.FileModeError, self.h5file.set_node_attr, "/test", "test", "bar" ) def test22_fileClosesNode(self): """Test node closing because of file closing.""" node = self.h5file.create_group("/", "test") self.h5file.close() self.assertRaises(tb.ClosedNodeError, getattr, node, "_v_attrs") def test23_reopenFile(self): """Testing reopening a file and closing it several times.""" self.h5file.create_array("/", "test", [1, 2, 3]) self.h5file.close() with tb.open_file(self.h5fname, "r") as h5file1: if tb.file._FILE_OPEN_POLICY == "strict": self.assertRaises(ValueError, tb.open_file, self.h5fname, "r") else: with tb.open_file(self.h5fname, "r") as h5file2: if common.verbose: print("(h5file1) test[1]:", h5file1.root.test[1]) self.assertEqual(h5file1.root.test[1], 2) h5file1.close() if common.verbose: print("(h5file2) test[1]:", h5file2.root.test[1]) self.assertEqual(h5file2.root.test[1], 2) class FlavorTestCase(common.TempFileMixin, common.PyTablesTestCase): """Test that setting, getting and changing the ``flavor`` attribute of a leaf works as expected.""" array_data = np.arange(10) scalar_data = np.int32(10) def _reopen(self, mode="r"): super()._reopen(mode) self.array = self.h5file.get_node("/array") self.scalar = self.h5file.get_node("/scalar") return True def setUp(self): super().setUp() self.array = self.h5file.create_array("/", "array", self.array_data) self.scalar = self.h5file.create_array("/", "scalar", self.scalar_data) def test00_invalid(self): """Setting an invalid flavor.""" self.assertRaises(tb.FlavorError, setattr, self.array, "flavor", "foo") def test01_readonly(self): """Setting a flavor in a read-only file.""" self._reopen(mode="r") self.assertRaises( tb.FileModeError, setattr, self.array, "flavor", tb.flavor.internal_flavor, ) def test02_change(self): """Changing the flavor and reading data.""" for flavor in tb.flavor.all_flavors: self.array.flavor = flavor self.assertEqual(self.array.flavor, flavor) idata = tb.flavor.array_of_flavor(self.array_data, flavor) odata = self.array[:] self.assertTrue(common.allequal(odata, idata, flavor)) def test03_store(self): """Storing a changed flavor.""" for flavor in tb.flavor.all_flavors: self.array.flavor = flavor self.assertEqual(self.array.flavor, flavor) self._reopen(mode="r+") self.assertEqual(self.array.flavor, flavor) def test04_missing(self): """Reading a dataset of a missing flavor.""" flavor = self.array.flavor # default is internal self.array._v_attrs.FLAVOR = "foobar" # breaks flavor self._reopen(mode="r") idata = tb.flavor.array_of_flavor(self.array_data, flavor) with self.assertWarns(tb.FlavorWarning): odata = self.array.read() self.assertTrue(common.allequal(odata, idata, flavor)) def test05_delete(self): """Deleting the flavor of a dataset.""" self.array.flavor = "python" # non-default self.assertEqual(self.array.flavor, "python") self.assertEqual(self.array.attrs.FLAVOR, "python") del self.array.flavor self.assertEqual(self.array.flavor, tb.flavor.internal_flavor) self.assertRaises(AttributeError, getattr, self.array.attrs, "FLAVOR") def test06_copyDeleted(self): """Copying a node with a deleted flavor (see #100).""" snames = [node._v_name for node in [self.array, self.scalar]] dnames = ["%s_copy" % name for name in snames] for name in snames: node = self.h5file.get_node("/", name) del node.flavor # Check the copied flavors right after copying and after reopening. for fmode in ["r+", "r"]: self._reopen(fmode) for sname, dname in zip(snames, dnames): if fmode == "r+": snode = self.h5file.get_node("/", sname) node = snode.copy("/", dname) elif fmode == "r": node = self.h5file.get_node("/", dname) self.assertEqual( node.flavor, tb.flavor.internal_flavor, "flavor of node ``%s`` is not internal: %r" % (node._v_pathname, node.flavor), ) def test07_restrict_flavors(self): # regression test for gh-163 all_flavors = list(tb.flavor.all_flavors) alias_map = tb.flavor.alias_map.copy() converter_map = tb.flavor.converter_map.copy() identifier_map = tb.flavor.identifier_map.copy() description_map = tb.flavor.description_map.copy() try: tb.flavor.restrict_flavors(keep=[]) self.assertLess(len(tb.flavor.alias_map), len(alias_map)) self.assertLess(len(tb.flavor.converter_map), len(converter_map)) finally: tb.flavor.all_flavors[:] = all_flavors[:] tb.flavor.alias_map.update(alias_map) tb.flavor.converter_map.update(converter_map) tb.flavor.identifier_map.update(identifier_map) tb.flavor.description_map.update(description_map) # @common.unittest.skipIf( # sys.getfilesystemencoding() != "utf-8", # "need utf-8 file-system encoding", # ) @common.unittest.skipIf( sys.platform == "win32", "no unicode filenames on windows" ) class UnicodeFilename(common.TempFileMixin, common.PyTablesTestCase): unicode_prefix = "para\u0140lel" def _getTempFileName(self): return tempfile.mktemp(prefix=self.unicode_prefix, suffix=".h5") def setUp(self): super().setUp() self.test = self.h5file.create_array("/", "test", [1, 2]) # So as to check the reading self._reopen() def test01(self): """Checking creating a filename with Unicode chars.""" test = self.h5file.root.test if common.verbose: print("Filename:", self.h5fname) print("Array:", test[:]) print("Should look like:", [1, 2]) self.assertEqual(test[:], [1, 2], "Values does not match.") def test02(self): """Checking tables.is_hdf5_file with a Unicode filename.""" self.h5file.close() if common.verbose: print("Filename:", self.h5fname) print(" tables.is_hdf5_file?:", tb.is_hdf5_file(self.h5fname)) self.assertTrue(tb.is_hdf5_file(self.h5fname)) def test03(self): """Checking is_pytables_file with a Unicode filename.""" self.h5file.close() if common.verbose: print("Filename:", self.h5fname) print("is_pytables_file?:", tb.is_pytables_file(self.h5fname)) self.assertNotEqual(tb.is_pytables_file(self.h5fname), False) @staticmethod def _store_carray(name, data, group): atom = tb.Atom.from_dtype(data.dtype) node = tb.CArray(group, name, shape=data.shape, atom=atom) node[:] = data def test_store_and_load_with_non_ascii_attributes(self): self.h5file.close() self.h5file = tb.open_file(self.h5fname, "a") root = self.h5file.root group = self.h5file.create_group(root, "face_data") array_name = "data at 40\N{DEGREE SIGN}C" data = np.sinh(np.linspace(-1.4, 1.4, 500)) with warnings.catch_warnings(): warnings.simplefilter("ignore", tb.NaturalNameWarning) self._store_carray(array_name, data, group) group = self.h5file.create_group(root, "vertex_data") @common.unittest.skipIf( sys.version_info < (3, 6), "PEP 519 was implemented in Python 3.6" ) class PathLikeFilename(common.TempFileMixin, common.PyTablesTestCase): def _getTempFileName(self): from pathlib import Path return Path(tempfile.mktemp(suffix=".h5")) def setUp(self): super().setUp() self.test = self.h5file.create_array("/", "test", [1, 2]) # So as to check the reading self._reopen() def test01(self): """Checking creating a file with a PathLike object as the filename.""" test = self.h5file.root.test if common.verbose: print("Filename:", self.h5fname) print("Array:", test[:]) print("Should look like:", [1, 2]) self.assertEqual(test[:], [1, 2], "Values does not match.") def test02(self): """Checking tables.is_hdf5_file with a PathLike object as the filename.""" self.h5file.close() if common.verbose: print("Filename:", self.h5fname) print(" tables.is_hdf5_file?:", tb.is_hdf5_file(self.h5fname)) self.assertTrue(tb.is_hdf5_file(self.h5fname)) def test03(self): """Checking is_pytables_file with a PathLike object as the filename.""" self.h5file.close() if common.verbose: print("Filename:", self.h5fname) print("is_pytables_file?:", tb.is_pytables_file(self.h5fname)) self.assertNotEqual(tb.is_pytables_file(self.h5fname), False) def test04_str(self): str(self.h5file) class FilePropertyTestCase(common.PyTablesTestCase): def setUp(self): super().setUp() self.h5fname = tempfile.mktemp(".h5") self.h5file = None def tearDown(self): if self.h5file: self.h5file.close() if Path(self.h5fname).is_file(): Path(self.h5fname).unlink() super().tearDown() def test_get_filesize(self): data = np.zeros((2000, 2000)) datasize = np.prod(data.shape) * data.dtype.itemsize self.h5file = tb.open_file(self.h5fname, mode="w") self.h5file.create_array(self.h5file.root, "array", data) h5_filesize = self.h5file.get_filesize() self.h5file.close() fs_filesize = Path(self.h5fname).stat().st_size self.assertGreaterEqual(h5_filesize, datasize) self.assertEqual(h5_filesize, fs_filesize) def test01_null_userblock_size(self): self.h5file = tb.open_file(self.h5fname, mode="w") self.h5file.create_array(self.h5file.root, "array", [1, 2]) self.assertEqual(self.h5file.get_userblock_size(), 0) def test02_null_userblock_size(self): self.h5file = tb.open_file(self.h5fname, mode="w") self.h5file.create_array(self.h5file.root, "array", [1, 2]) self.h5file.close() self.h5file = tb.open_file(self.h5fname, mode="r") self.assertEqual(self.h5file.get_userblock_size(), 0) def test03_null_userblock_size(self): USER_BLOCK_SIZE = 0 self.h5file = tb.open_file( self.h5fname, mode="w", user_block_size=USER_BLOCK_SIZE ) self.h5file.create_array(self.h5file.root, "array", [1, 2]) self.assertEqual(self.h5file.get_userblock_size(), 0) def test01_userblock_size(self): USER_BLOCK_SIZE = 512 self.h5file = tb.open_file( self.h5fname, mode="w", user_block_size=USER_BLOCK_SIZE ) self.h5file.create_array(self.h5file.root, "array", [1, 2]) self.assertEqual(self.h5file.get_userblock_size(), USER_BLOCK_SIZE) def test02_userblock_size(self): USER_BLOCK_SIZE = 512 self.h5file = tb.open_file( self.h5fname, mode="w", user_block_size=USER_BLOCK_SIZE ) self.h5file.create_array(self.h5file.root, "array", [1, 2]) self.h5file.close() self.h5file = tb.open_file(self.h5fname, mode="r") self.assertEqual(self.h5file.get_userblock_size(), USER_BLOCK_SIZE) def test_small_userblock_size(self): USER_BLOCK_SIZE = 12 self.assertRaises( ValueError, tb.open_file, self.h5fname, mode="w", user_block_size=USER_BLOCK_SIZE, ) def test_invalid_userblock_size(self): USER_BLOCK_SIZE = 1025 self.assertRaises( ValueError, tb.open_file, self.h5fname, mode="w", user_block_size=USER_BLOCK_SIZE, ) # Test for reading a file that uses Blosc and created on a big-endian platform @common.unittest.skipIf(not common.blosc_avail, "Blosc not available") class BloscBigEndian(common.TestFileMixin, common.PyTablesTestCase): h5fname = common.test_filename("blosc_bigendian.h5") def test00_bigendian(self): """Checking compatibility with Blosc on big-endian machines.""" # Check that we can read the contents without problems (nor warnings!) for dset_name in ("i1", "i2", "i4", "i8"): a = np.arange(10, dtype=dset_name) dset = self.h5file.get_node("/" + dset_name) self.assertTrue( common.allequal(a, dset[:]), "Error in big-endian data!" ) # Case test for Blosc and subprocesses (via multiprocessing module) # The worker function for the subprocess (needs to be here because Windows # has problems pickling nested functions with the multiprocess module :-/) def _worker(fn, qout=None): fp = tb.open_file(fn) if common.verbose: print("About to load: ", fn) rows = fp.root.table.where("(f0 < 10)") if common.verbose: print("Got the iterator, about to iterate") next(rows) if common.verbose: print("Succeeded in one iteration\n") fp.close() if qout is not None: qout.put("Done") # From: Yaroslav Halchenko # Subject: Skip the unittest on kFreeBSD and Hurd -- locking seems to # be N/A # # on kfreebsd /dev/shm is N/A # on Hurd -- inter-process semaphore locking is N/A @common.unittest.skipIf( not multiprocessing_imported, "multiprocessing module not available" ) @common.unittest.skipIf( platform.system().lower() in ("gnu", "gnu/kfreebsd"), "multiprocessing module is not " "supported on Hurd/kFreeBSD", ) @common.unittest.skipIf(not common.blosc_avail, "Blosc not available") class BloscSubprocess(common.PyTablesTestCase): def test_multiprocess(self): # Create a relatively large table with Blosc level 9 (large blocks) h5fname = tempfile.mktemp(prefix="multiproc-blosc9-", suffix=".h5") try: size = 300_000 sa = np.fromiter( ((i, i**2, i // 3) for i in range(size)), "i4,i8,f8" ) with tb.open_file(h5fname, "w") as h5file: h5file.create_table( h5file.root, "table", sa, filters=tb.Filters(complevel=9, complib="blosc"), chunkshape=(size // 3,), ) if common.verbose: print("**** Running from main process:") _worker(h5fname) if common.verbose: print("**** Running from subprocess:") try: qout = mp.Queue() except OSError: print("Permission denied due to /dev/shm settings") else: ps = mp.Process( target=_worker, args=( h5fname, qout, ), ) ps.daemon = True ps.start() result = qout.get() if common.verbose: print(result) ps.join() # Avoid warnings with later tests forking subprocesses. ps.terminate() finally: Path(h5fname).unlink() class HDF5ErrorHandling(common.PyTablesTestCase): def setUp(self): super().setUp() self._old_policy = tb.HDF5ExtError.DEFAULT_H5_BACKTRACE_POLICY def tearDown(self): tb.HDF5ExtError.DEFAULT_H5_BACKTRACE_POLICY = self._old_policy super().tearDown() def test_silence_messages(self): code = """ import tables as tb tb.silence_hdf5_messages(False) tb.silence_hdf5_messages() try: tb.open_file(r'%s') except tb.HDF5ExtError, e: pass """ filename = tempfile.mktemp(prefix="hdf5-error-handling-", suffix=".py") try: with open(filename, "w") as fp: fp.write(code % filename) p = subprocess.Popen( [sys.executable, filename], stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) (stdout, stderr) = p.communicate() self.assertNotIn("HDF5-DIAG", stderr.decode("ascii")) finally: Path(filename).unlink() # This test is a bit flaky and in some situations it fails # E.g. on Mac OSX (arm64), I am getting this: # FAIL: None (tables.tests.test_basics.HDF5ErrorHandling) # ---------------------------------------------------------------------- # Traceback (most recent call last): # File "/Users/faltet/software/PyTables-upstream/tables/tests/test_basics.py", # line 2231, in test_enable_messages # self.assertIn("HDF5-DIAG", stderr.decode('ascii')) # AssertionError: 'HDF5-DIAG' not found in 'Traceback (most recent call last):\n # symbol not found in flat namespace \'_blosc2_cbuffer_sizes\'\n' # As the fix is not clear to me, I prefer to disable it until a more robust # path is found. def _test_enable_messages(self): code = """ import tables as tb tb.silence_hdf5_messages() tb.silence_hdf5_messages(False) try: tb.open_file(r'%s') except tb.HDF5ExtError as e: pass """ filename = tempfile.mktemp(prefix="hdf5-error-handling-", suffix=".py") try: Path(filename).write_text(code % filename) p = subprocess.Popen( [sys.executable, filename], stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) (stdout, stderr) = p.communicate() self.assertIn("HDF5-DIAG", stderr.decode("ascii")) finally: Path(filename).unlink() def _raise_exterror(self): h5fname = tempfile.mktemp(".h5") Path(h5fname).write_text("") try: h5file = tb.open_file(h5fname) h5file.close() finally: Path(h5fname).unlink() def test_h5_backtrace_quiet(self): tb.HDF5ExtError.DEFAULT_H5_BACKTRACE_POLICY = True with self.assertRaises(tb.HDF5ExtError) as cm: self._raise_exterror() self.assertIsNotNone(cm.exception.h5backtrace) def test_h5_backtrace_verbose(self): tb.HDF5ExtError.DEFAULT_H5_BACKTRACE_POLICY = "VERBOSE" with self.assertRaises(tb.HDF5ExtError) as cm: self._raise_exterror() self.assertIsNotNone(cm.exception.h5backtrace) msg = str(cm.exception) self.assertIn(cm.exception.h5backtrace[-1][-1], msg) def test_h5_backtrace_ignore(self): tb.HDF5ExtError.DEFAULT_H5_BACKTRACE_POLICY = False with self.assertRaises(tb.HDF5ExtError) as cm: self._raise_exterror() self.assertIsNone(cm.exception.h5backtrace) class TestDescription(common.PyTablesTestCase): def test_isdescription_inheritance(self): # Regression test for gh-65 class TestDescParent(tb.IsDescription): c = tb.Int32Col() class TestDesc(TestDescParent): pass self.assertIn("c", TestDesc.columns) def test_descr_from_dtype(self): t = np.dtype([("col1", "int16"), ("col2", float)]) descr, byteorder = tb.description.descr_from_dtype(t) self.assertIn("col1", descr._v_colobjects) self.assertIn("col2", descr._v_colobjects) self.assertEqual(len(descr._v_colobjects), 2) self.assertIsInstance(descr._v_colobjects["col1"], tb.Col) self.assertIsInstance(descr._v_colobjects["col2"], tb.Col) self.assertEqual(descr._v_colobjects["col1"].dtype, np.int16) self.assertEqual(descr._v_colobjects["col2"].dtype, float) def test_descr_from_dtype_rich_dtype(self): header = [ (("timestamp", "t"), "u4"), (("unit (cluster) id", "unit"), "u2"), ] t = np.dtype(header) descr, byteorder = tb.description.descr_from_dtype(t) self.assertEqual(len(descr._v_names), 2) self.assertEqual(sorted(descr._v_names), ["t", "unit"]) def test_descr_from_dtype_comp_01(self): d1 = np.dtype([("x", "int16"), ("y", "int16")]) d_comp = np.dtype([("time", "float64"), ("value", d1)]) descr, byteorder = tb.description.descr_from_dtype(d_comp) self.assertTrue(descr._v_is_nested) self.assertIn("time", descr._v_colobjects) self.assertIn("value", descr._v_colobjects) self.assertEqual(len(descr._v_colobjects), 2) self.assertIsInstance(descr._v_colobjects["time"], tb.Col) self.assertTrue( isinstance(descr._v_colobjects["value"], tb.Description) ) self.assertEqual(descr._v_colobjects["time"].dtype, np.float64) def test_descr_from_dtype_comp_02(self): d1 = np.dtype([("x", "int16"), ("y", "int16")]) d_comp = np.dtype([("time", "float64"), ("value", (d1, (1,)))]) with self.assertWarns(UserWarning): descr, byteorder = tb.description.descr_from_dtype(d_comp) self.assertTrue(descr._v_is_nested) self.assertIn("time", descr._v_colobjects) self.assertIn("value", descr._v_colobjects) self.assertEqual(len(descr._v_colobjects), 2) self.assertIsInstance(descr._v_colobjects["time"], tb.Col) self.assertTrue( isinstance(descr._v_colobjects["value"], tb.Description) ) self.assertEqual(descr._v_colobjects["time"].dtype, np.float64) def test_dtype_from_descr_is_description(self): # See gh-152 class TestDescParent(tb.IsDescription): col1 = tb.Int16Col() col2 = tb.FloatCol() dtype = np.dtype([("col1", "int16"), ("col2", float)]) t = tb.description.dtype_from_descr(TestDescParent) self.assertEqual(t, dtype) def test_dtype_from_descr_is_description_instance(self): # See gh-152 class TestDescParent(tb.IsDescription): col1 = tb.Int16Col() col2 = tb.FloatCol() dtype = np.dtype([("col1", "int16"), ("col2", float)]) t = tb.description.dtype_from_descr(TestDescParent()) self.assertEqual(t, dtype) def test_dtype_from_descr_description_instance(self): # See gh-152 class TestDescParent(tb.IsDescription): col1 = tb.Int16Col() col2 = tb.FloatCol() dtype = np.dtype([("col1", "int16"), ("col2", float)]) desctiption = tb.Description(TestDescParent().columns) t = tb.description.dtype_from_descr(desctiption) self.assertEqual(t, dtype) def test_dtype_from_descr_dict(self): # See gh-152 dtype = np.dtype([("col1", "int16"), ("col2", float)]) t = tb.description.dtype_from_descr( {"col1": tb.Int16Col(), "col2": tb.FloatCol()} ) self.assertEqual(t, dtype) def test_dtype_from_descr_invalid_type(self): # See gh-152 self.assertRaises(ValueError, tb.description.dtype_from_descr, []) def test_dtype_from_descr_byteorder(self): # See gh-152 class TestDescParent(tb.IsDescription): col1 = tb.Int16Col() col2 = tb.FloatCol() t = tb.description.dtype_from_descr(TestDescParent, byteorder=">") self.assertEqual(t["col1"].byteorder, ">") self.assertEqual(t["col2"].byteorder, ">") def test_str_names(self): # see gh-42 d = {"name": tb.Int16Col()} descr = tb.Description(d) self.assertEqual(sorted(descr._v_names), sorted(d.keys())) self.assertIsInstance(descr._v_dtype, np.dtype) self.assertTrue(sorted(descr._v_dtype.fields), sorted(d.keys())) class TestAtom(common.PyTablesTestCase): def test_atom_attributes01(self): shape = (10, 10) a = tb.Float64Atom(shape=shape) self.assertEqual(a.dflt, 0.0) self.assertEqual(a.dtype, np.dtype((np.float64, shape))) self.assertEqual(a.itemsize, a.dtype.base.itemsize) self.assertEqual(a.kind, "float") self.assertEqual(a.ndim, len(shape)) # self.assertEqual(a.recarrtype, ) self.assertEqual(a.shape, shape) self.assertEqual(a.size, a.itemsize * np.prod(shape)) self.assertEqual(a.type, "float64") def test_atom_copy01(self): shape = (10, 10) a = tb.Float64Atom(shape=shape) aa = a.copy() self.assertEqual(aa.shape, shape) def test_atom_copy02(self): dflt = 2.0 a = tb.Float64Atom(dflt=dflt) aa = a.copy() self.assertEqual(aa.dflt, dflt) def test_atom_copy_override(self): shape = (10, 10) dflt = 2.0 a = tb.Float64Atom(shape=shape, dflt=dflt) aa = a.copy(dflt=-dflt) self.assertEqual(aa.shape, shape) self.assertNotEqual(aa.dflt, dflt) self.assertEqual(aa.dflt, -dflt) class TestCol(common.PyTablesTestCase): def test_col_copy01(self): shape = (10, 10) c = tb.Float64Col(shape=shape) cc = c.copy() self.assertEqual(cc.shape, shape) def test_col_copy02(self): dflt = 2.0 c = tb.Float64Col(dflt=dflt) cc = c.copy() self.assertEqual(cc.dflt, dflt) def test_col_copy_override(self): shape = (10, 10) dflt = 2.0 pos = 3 c = tb.Float64Col(shape=shape, dflt=dflt, pos=pos) cc = c.copy(pos=2) self.assertEqual(cc.shape, shape) self.assertEqual(cc.dflt, dflt) self.assertNotEqual(cc._v_pos, pos) self.assertEqual(cc._v_pos, 2) class TestSysattrCompatibility(common.PyTablesTestCase): def test_open_python2(self): h5fname = common.test_filename("python2.h5") with tb.open_file(h5fname, "r") as h5file: self.assertTrue(h5file.isopen) def test_open_python3(self): h5fname = common.test_filename("python3.h5") with tb.open_file(h5fname, "r") as h5file: self.assertTrue(h5file.isopen) def suite(): theSuite = common.unittest.TestSuite() niter = 1 for i in range(niter): theSuite.addTest(common.make_suite(OpenFileFailureTestCase)) theSuite.addTest(common.make_suite(NodeCacheOpenFile)) theSuite.addTest(common.make_suite(NoNodeCacheOpenFile)) theSuite.addTest(common.make_suite(DictNodeCacheOpenFile)) theSuite.addTest(common.make_suite(CheckFileTestCase)) theSuite.addTest(common.make_suite(ThreadingTestCase)) theSuite.addTest(common.make_suite(PythonAttrsTestCase)) theSuite.addTest(common.make_suite(StateTestCase)) theSuite.addTest(common.make_suite(FlavorTestCase)) theSuite.addTest(common.make_suite(UnicodeFilename)) theSuite.addTest(common.make_suite(PathLikeFilename)) theSuite.addTest(common.make_suite(FilePropertyTestCase)) theSuite.addTest(common.make_suite(BloscBigEndian)) theSuite.addTest(common.make_suite(BloscSubprocess)) theSuite.addTest(common.make_suite(HDF5ErrorHandling)) theSuite.addTest(common.make_suite(TestDescription)) theSuite.addTest(common.make_suite(TestAtom)) theSuite.addTest(common.make_suite(TestCol)) theSuite.addTest(common.make_suite(TestSysattrCompatibility)) return theSuite if __name__ == "__main__": common.parse_argv(sys.argv) common.print_versions() common.unittest.main(defaultTest="suite")