Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
filename = self.getFileName("query_dset")
f = h5py.File(filename, "w")
count = 100
dt = np.dtype([('a', np.int), ('b', np.int)])
dset = f.create_dataset('dset', (count,), dtype=dt)
elem = dset[0]
for i in range(count):
elem['a'] = i // 10
elem['b'] = i % 10
dset[i] = elem
# select from dset1
if h5py.__name__ == "h5pyd":
count = 0
for row in dset.read_where("b>4"):
self.assertTrue(row[1] > 4)
count += 1
self.assertEqual(count, 50)
f.close()
def test_create(self):
filename = self.getFileName("new_file")
print("filename:", filename)
now = time.time()
f = h5py.File(filename, 'w')
self.assertEqual(f.filename, filename)
self.assertEqual(f.name, "/")
self.assertTrue(f.id.id is not None)
self.assertEqual(len(f.keys()), 0)
self.assertEqual(f.mode, 'r+')
is_hsds = False
if isinstance(f.id.id, str) and f.id.id.startswith("g-"):
is_hsds = True # HSDS has different permission defaults
if h5py.__name__ == "h5pyd":
self.assertTrue(f.id.http_conn.endpoint.startswith("http"))
self.assertTrue(f.id.id is not None)
self.assertTrue('/' in f)
# Check domain's timestamps
if h5py.__name__ == "h5pyd" and is_hsds:
# TBD: remove is_hsds when h5serv timestamp changed to float
#print("modified:", datetime.fromtimestamp(f.modified), f.modified)
#print("created: ", datetime.fromtimestamp(f.created), f.created)
#print("now: ", datetime.fromtimestamp(now), now)
# verify the timestamps make sense
# we add a 30-sec margin to account for possible time skew
# between client and server
self.assertTrue(f.created - 30.0 < now)
self.assertTrue(f.created + 30.0 > now)
self.assertTrue(f.modified - 30.0 < now)
self.assertTrue(f.modified + 30.0 > now)
self.assertTrue(isinstance(val, np.int64))
# try will ellipsis
val = dset[...]
self.assertEqual(val, 24)
self.assertTrue(isinstance(val, np.ndarray))
# try setting value using tuple
dset[()] = 99
val = dset[()]
self.assertEqual(val, 99)
self.assertEqual(dset.file.filename, filename)
# Check dataset's last modified time
if h5py.__name__ == "h5pyd":
self.assertTrue(isinstance(dset.modified, datetime))
f.close()
# create group using nested path
g2 = r['g2']
r['g1/g1.3'] = g2
self.assertEqual(len(r), 4)
# try creating a link with a space in the name
r["a space"] = g2
self.assertEqual(len(r), 5)
# re-create softlink
r['mysoftlink'] = h5py.SoftLink('/g1/g1.1')
# Check group's last modified time
if h5py.__name__ == "h5pyd":
self.assertTrue(isinstance(g1.modified, datetime))
#self.assertEqual(g1.modified.tzname(), six.u('UTC'))
f.close()
# re-open file in read-only mode
f = h5py.File(filename, 'r')
self.assertEqual(len(f), 6)
for name in ("g1", "g2", "g4", "g1.1", "a space", "mysoftlink"):
self.assertTrue(name in f)
self.assertTrue("/g1/g1.1" in f)
g1_1 = f["/g1/g1.1"]
linkee_class = r.get('mysoftlink', getclass=True)
self.assertEqual(linkee_class, h5py.Group)
link_class = r.get('mysoftlink', getclass=True, getlink=True)
self.assertEqual(link_class, h5py.SoftLink)
def test_serverinfo(self):
if h5py.__name__ == "h5pyd":
info = h5py.getServerInfo()
self.assertTrue("greeting" in info)
self.assertTrue("name" in info)
self.assertTrue("about" in info)
self.assertTrue("endpoint" in info)
self.assertTrue("username" in info)
self.assertTrue("password" in info)
self.assertEqual(str(dset.dtype), 'float32')
self.assertTrue(isinstance(dset.maxshape, tuple))
self.assertEqual(len(dset.maxshape), 2)
self.assertEqual(dset.maxshape[0], 40)
self.assertEqual(dset.maxshape[1], 80)
self.assertEqual(dset[0,0], 0)
dset_ref = f['/simple_dset']
self.assertTrue(dset_ref is not None)
if not config.get("use_h5py"):
# obj ids should be the same with h5pyd (but not h5py)
self.assertEqual(dset.id.id, dset_ref.id.id)
# Check dataset's last modified time
self.assertTrue(isinstance(dset.modified, datetime))
if h5py.__name__ == "h5pyd":
# test h5pyd extensions
if not config.get('use_h5py') and isinstance(f.id.id, str) and f.id.id.startswith("g-"):
print("test h5pyd extensions")
self.assertEqual(dset.num_chunks, 0)
self.assertEqual(dset.allocated_size, 0)
# try with chunk=True
dset_chunked = f.create_dataset('chunked_dset', dims, dtype='f4', chunks=True)
if config.get('use_h5py') or (isinstance(f.id.id, str) and f.id.id.startswith("g-")):
self.assertTrue(dset_chunked.chunks)
else:
# h5serv not reporting chunks
self.assertTrue(dset_chunked.chunks is None)
f.close()