test_h5f.py 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108
  1. # This file is part of h5py, a Python interface to the HDF5 library.
  2. #
  3. # http://www.h5py.org
  4. #
  5. # Copyright 2008-2013 Andrew Collette and contributors
  6. #
  7. # License: Standard 3-clause BSD; see "license.txt" for full license terms
  8. # and contributor agreement.
  9. import tempfile
  10. import shutil
  11. import os
  12. import numpy as np
  13. from h5py import File, special_dtype
  14. from h5py._hl.files import direct_vfd
  15. from .common import ut, TestCase
  16. class TestFileID(TestCase):
  17. def test_descriptor_core(self):
  18. with File(self.mktemp(), driver='core',
  19. backing_store=False, mode='x') as f:
  20. assert isinstance(f.id.get_vfd_handle(), int)
  21. def test_descriptor_sec2(self):
  22. dn_tmp = tempfile.mkdtemp('h5py.lowtest.test_h5f.TestFileID.test_descriptor_sec2')
  23. fn_h5 = os.path.join(dn_tmp, 'test.h5')
  24. try:
  25. with File(fn_h5, driver='sec2', mode='x') as f:
  26. descriptor = f.id.get_vfd_handle()
  27. self.assertNotEqual(descriptor, 0)
  28. os.fsync(descriptor)
  29. finally:
  30. shutil.rmtree(dn_tmp)
  31. @ut.skipUnless(direct_vfd,
  32. "DIRECT driver is supported on Linux if hdf5 is "
  33. "built with the appriorate flags.")
  34. def test_descriptor_direct(self):
  35. dn_tmp = tempfile.mkdtemp('h5py.lowtest.test_h5f.TestFileID.test_descriptor_direct')
  36. fn_h5 = os.path.join(dn_tmp, 'test.h5')
  37. try:
  38. with File(fn_h5, driver='direct', mode='x') as f:
  39. descriptor = f.id.get_vfd_handle()
  40. self.assertNotEqual(descriptor, 0)
  41. os.fsync(descriptor)
  42. finally:
  43. shutil.rmtree(dn_tmp)
  44. class TestCacheConfig(TestCase):
  45. def test_simple_gets(self):
  46. dn_tmp = tempfile.mkdtemp('h5py.lowtest.test_h5f.TestFileID.TestCacheConfig.test_simple_gets')
  47. fn_h5 = os.path.join(dn_tmp, 'test.h5')
  48. try:
  49. with File(fn_h5, mode='x') as f:
  50. hit_rate = f._id.get_mdc_hit_rate()
  51. mdc_size = f._id.get_mdc_size()
  52. finally:
  53. shutil.rmtree(dn_tmp)
  54. def test_hitrate_reset(self):
  55. dn_tmp = tempfile.mkdtemp('h5py.lowtest.test_h5f.TestFileID.TestCacheConfig.test_hitrate_reset')
  56. fn_h5 = os.path.join(dn_tmp, 'test.h5')
  57. try:
  58. with File(fn_h5, mode='x') as f:
  59. hit_rate = f._id.get_mdc_hit_rate()
  60. f._id.reset_mdc_hit_rate_stats()
  61. hit_rate = f._id.get_mdc_hit_rate()
  62. assert hit_rate == 0
  63. finally:
  64. shutil.rmtree(dn_tmp)
  65. def test_mdc_config_get(self):
  66. dn_tmp = tempfile.mkdtemp('h5py.lowtest.test_h5f.TestFileID.TestCacheConfig.test_mdc_config_get')
  67. fn_h5 = os.path.join(dn_tmp, 'test.h5')
  68. try:
  69. with File(fn_h5, mode='x') as f:
  70. conf = f._id.get_mdc_config()
  71. f._id.set_mdc_config(conf)
  72. finally:
  73. shutil.rmtree(dn_tmp)
  74. class TestVlenData(TestCase):
  75. def test_vlen_strings(self):
  76. # Create file with dataset containing vlen arrays of vlen strings
  77. dn_tmp = tempfile.mkdtemp('h5py.lowtest.test_h5f.TestVlenStrings.test_vlen_strings')
  78. fn_h5 = os.path.join(dn_tmp, 'test.h5')
  79. try:
  80. with File(fn_h5, mode='w') as h:
  81. vlen_str = special_dtype(vlen=str)
  82. vlen_vlen_str = special_dtype(vlen=vlen_str)
  83. ds = h.create_dataset('/com', (2,), dtype=vlen_vlen_str)
  84. ds[0] = (np.array(["a", "b", "c"], dtype=vlen_vlen_str))
  85. ds[1] = (np.array(["d", "e", "f","g"], dtype=vlen_vlen_str))
  86. with File(fn_h5, "r") as h:
  87. ds = h["com"]
  88. assert ds[0].tolist() == [b'a', b'b', b'c']
  89. assert ds[1].tolist() == [b'd', b'e', b'f', b'g']
  90. finally:
  91. shutil.rmtree(dn_tmp)