import unittest
from tempfile import NamedTemporaryFile
import linkahead as db
import linkahead.apiutils
import h5py
import numpy as np
from caosadvancedtools.cfoods import h5
from caosadvancedtools.cfoods.h5 import h5_attr_to_property
from create_dummy_hdf5file import create_hdf5_file
ENTS = {
101: db.Record(id=101),
102: db.Record(id=102),
103: db.Record(id=103).add_property("test", value=101,
datatype=db.REFERENCE),
}
def dummy_get(eid):
return ENTS[eid]
class H5CFoodTest(unittest.TestCase):
def setUp(self):
self.h5file = NamedTemporaryFile(delete=False, suffix=".h5")
self.h5file.close()
create_hdf5_file(self.h5file.name)
self.h5obj = h5py.File(self.h5file.name, mode="a")
def test_create_record_records(self):
result = h5.H5CFood.create_structure(self.h5obj)
record_list = []
parents = ['group_level1_a', 'group_level1_b', 'group_level1_c', 'root_integers']
for i in parents:
record_list.append(db.Record().add_parent(name=i))
found_parents = []
for ent in [p.value for p in result.properties]:
if ent.parents[0].name == 'group_level1_a':
found_parents.append('group_level1_a')
self.assertTrue(ent.get_property("group_level2_aa") is not None)
self.assertTrue(ent.get_property("group_level1_a") is None)
elif ent.parents[0].name == 'group_level1_b':
found_parents.append('group_level1_b')
pass
elif ent.parents[0].name == 'group_level1_c':
found_parents.append('group_level1_c')
pass
elif ent.parents[0].name == 'root_integers':
found_parents.append('root_integers')
pass
for p in parents:
self.assertTrue(p in found_parents)
for i in range(len(result.properties)):
for j in result.properties[i].value.get_parents():
for k in record_list[i].get_parents():
self.assertEqual(j.name, k.name)
result1 = h5.H5CFood.create_structure(self.h5obj["group_level1_a"])
for i in result1.get_parents():
self.assertEqual(i.name, "group_level1_a")
result2 = h5.H5CFood.create_structure(self.h5obj["group_level1_a/group_level2_aa"])
for i in result2.get_parents():
self.assertEqual(i.name, "group_level2_aa")
def test_collect_existing_structure(self):
# TODO this does probably break the code: The function will not be
# restored correctly.
# Change it to use the BaseMockUpTest
real_retrieve = linkahead.apiutils.retrieve_entity_with_id
linkahead.apiutils.retrieve_entity_with_id = dummy_get
# should run without problem
h5.collect_existing_structure(db.Record(), db.Record(id=234), h5.EntityMapping())
# test with retrieval: both Records have one test Property with one
# value -> The referenced Entities are matched
r_exist = db.Record(id=234)
r_exist.add_property("test", value=101, datatype=db.REFERENCE)
r_target = db.Record()
r_child = db.Record()
r_target.add_property("test", value=r_child, datatype=db.REFERENCE)
em = h5.EntityMapping()
h5.collect_existing_structure(r_target, r_exist, em)
self.assertTrue(em.to_existing[r_child._cuid] is ENTS[101])
self.assertTrue(em.to_target[101] is r_child)
# test with retrieval: the existing Record has another Property
# -> The referenced Entities are matched
r_exist = db.Record(id=234)
r_exist.add_property("test_other", value=101, datatype=db.REFERENCE)
r_target = db.Record()
r_child = db.Record()
r_target.add_property("test", value=r_child, datatype=db.REFERENCE)
em = h5.EntityMapping()
h5.collect_existing_structure(r_target, r_exist, em)
self.assertEqual(em.to_existing, {})
self.assertEqual(em.to_target, {})
# test with retrieval: both Records have one test Property; the
# existing is missing the value -> The referenced Entities are matched
r_exist = db.Record(id=234)
r_exist.add_property("test", value=None, datatype=db.REFERENCE)
r_target = db.Record()
r_child = db.Record()
r_target.add_property("test", value=r_child, datatype=db.REFERENCE)
em = h5.EntityMapping()
h5.collect_existing_structure(r_target, r_exist, em)
self.assertEqual(em.to_existing, {})
self.assertEqual(em.to_target, {})
# test with retrieval: both Records have one test Property with
# multiple values -> The referenced Entities are matched
r_exist = db.Record(id=234)
r_exist.add_property("test", value=[101, 102], datatype=db.LIST(db.REFERENCE))
r_target = db.Record()
r_child = db.Record()
r_child2 = db.Record()
r_target.add_property("test", value=[r_child, r_child2],
datatype=db.LIST(db.REFERENCE))
em = h5.EntityMapping()
h5.collect_existing_structure(r_target, r_exist, em)
self.assertEqual(em.to_existing[r_child._cuid], ENTS[101])
self.assertEqual(em.to_existing[r_child2._cuid], ENTS[102])
self.assertEqual(em.to_target[101], r_child)
self.assertEqual(em.to_target[102], r_child2)
# test with retrieval: both Records have one test Property with one
# value; Add another recursion level -> The referenced Entities are matched
r_exist = db.Record(id=234)
r_exist.add_property("test", value=103, datatype=db.REFERENCE)
r_target = db.Record()
r_child = db.Record()
r_child2 = db.Record()
r_target.add_property("test", value=r_child, datatype=db.REFERENCE)
r_child.add_property("test", value=r_child2, datatype=db.REFERENCE)
em = h5.EntityMapping()
h5.collect_existing_structure(r_target, r_exist, em)
self.assertEqual(em.to_existing[r_child._cuid], ENTS[103])
self.assertEqual(em.to_target[103], r_child)
self.assertEqual(em.to_existing[r_child2._cuid], ENTS[101])
self.assertEqual(em.to_target[101], r_child2)
linkahead.apiutils.retrieve_entity_with_id = real_retrieve
def test_h5_attr_to_property(self):
test_int: int = 1
test_integer = np.int_(1)
test_float = np.float_(1.0)
test_str = "Test"
test_complex: complex = 2+3j
self.assertRaises(NotImplementedError, h5_attr_to_property,
test_int) # only numpy-integers processed?
self.assertTupleEqual((1, db.INTEGER), h5_attr_to_property(test_integer))
self.assertTupleEqual((1.0, db.DOUBLE), h5_attr_to_property(test_float))
self.assertTupleEqual(("Test", db.TEXT), h5_attr_to_property(test_str))
self.assertTupleEqual((2+3j, db.TEXT), h5_attr_to_property(test_complex))
# strings are often represented using a binary format
self.assertTupleEqual(("yeti", db.TEXT), h5_attr_to_property(
np.array(["yeti"], dtype=h5py.string_dtype(r'utf-8', 8))[0]))
test_integer_1d = np.arange(10)
test_float_1d = np.arange(0, 1, 0.1)
test_str_1d = np.array(["a", "b", "c"])
self.assertTrue((np.arange(10) == h5_attr_to_property(test_integer_1d)[0]).all())
self.assertTrue(db.LIST(db.INTEGER) == h5_attr_to_property(test_integer_1d)[1])
self.assertTrue((np.arange(0, 1, 0.1) == h5_attr_to_property(test_float_1d)[0]).all())
self.assertTrue(db.LIST(db.DOUBLE) == h5_attr_to_property(test_float_1d)[1])
self.assertTrue((np.array(["a", "b", "c"]) == h5_attr_to_property(test_str_1d)[0]).all())
self.assertTrue(db.LIST(db.TEXT) == h5_attr_to_property(test_str_1d)[1])
test_integers_2d = np.diag(np.arange(100))
test_floats_2d = np.eye(100)
self.assertTupleEqual((None, None), h5_attr_to_property(test_integers_2d))
self.assertTupleEqual((None, None), h5_attr_to_property(test_floats_2d))
# Test scalar values given as np.array
self.assertTupleEqual((1, db.INTEGER), h5_attr_to_property(np.array(1)))
self.assertTupleEqual((1.123, db.DOUBLE), h5_attr_to_property(np.array(1.123)))
self.assertTupleEqual(('Hello World', db.TEXT),
h5_attr_to_property(np.array("Hello World")))