-
Alexander Schlemmer authoredAlexander Schlemmer authored
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_high_level_api.py 10.04 KiB
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2018 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
# Copyright (C) 2022 Alexander Schlemmer <alexander.schlemmer@ds.mpg.de>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
# ** end header
#
# Test high level api module
# A. Schlemmer, 02/2022
import caosdb as db
from caosdb.high_level_api import (convert_to_entity, convert_to_python_object)
from caosdb.high_level_api import (CaosDBPythonUnresolvedParent,
CaosDBPythonUnresolvedReference,
CaosDBPythonRecord, CaosDBPythonFile)
from caosdb.apiutils import compare_entities
import pytest
from lxml import etree
import os
import tempfile
import pickle
import sys, traceback, pdb
@pytest.fixture
def testrecord():
parser = etree.XMLParser(remove_comments=True)
testrecord = db.Record._from_xml(
db.Record(),
etree.parse(os.path.join(os.path.dirname(__file__), "test_record.xml"),
parser).getroot())
return testrecord
def test_convert_object(testrecord):
r2 = convert_to_python_object(testrecord)
assert r2.species == "Rabbit"
def test_pickle_object(testrecord):
r2 = convert_to_python_object(testrecord)
with tempfile.TemporaryFile() as f:
pickle.dump(r2, f)
f.seek(0)
rn2 = pickle.load(f)
assert r2.date == rn2.date
def test_convert_record():
"""
Test the high level python API.
"""
r = db.Record()
r.add_parent("bla")
r.add_property(name="a", value=42)
r.add_property(name="b", value="test")
obj = convert_to_python_object(r)
assert obj.a == 42
assert obj.b == "test"
# There is no such property
with pytest.raises(AttributeError):
assert obj.c == 18
assert obj.has_parent("bla") is True
assert obj.has_parent(CaosDBPythonUnresolvedParent(name="bla")) is True
# Check the has_parent function:
assert obj.has_parent("test") is False
assert obj.has_parent(CaosDBPythonUnresolvedParent(name="test")) is False
# duplicate parent
with pytest.raises(RuntimeError):
obj.add_parent("bla")
# add parent with just an id:
obj.add_parent(CaosDBPythonUnresolvedParent(id=225))
assert obj.has_parent(225) is True
assert obj.has_parent(CaosDBPythonUnresolvedParent(id=225)) is True
assert obj.has_parent(226) is False
assert obj.has_parent(CaosDBPythonUnresolvedParent(id=228)) is False
# same with just a name:
obj.add_parent(CaosDBPythonUnresolvedParent(name="another"))
assert obj.has_parent("another") is True
def test_convert_with_references():
r_ref = db.Record()
r_ref.add_property(name="a", value=42)
r = db.Record()
r.add_property(name="ref", value=r_ref)
# try:
obj = convert_to_python_object(r)
# except:
# extype, value, tb = sys.exc_info()
# traceback.print_exc()
# pdb.post_mortem(tb)
assert obj.ref.a == 42
# With datatype:
r_ref = db.Record()
r_ref.add_parent("bla")
r_ref.add_property(name="a", value=42)
r = db.Record()
r.add_property(name="ref", value=r_ref)
obj = convert_to_python_object(r)
assert obj.ref.a == 42
# Parent does not automatically lead to a datatype:
assert obj.get_property_metadata("ref").datatype is None
assert obj.ref.has_parent("bla") is True
# Add datatype explicitely:
r_ref = db.Record()
r_ref.add_parent("bla")
r_ref.add_property(name="a", value=42)
r = db.Record()
r.add_property(name="ref", value=r_ref, datatype="bla")
obj = convert_to_python_object(r)
assert obj.ref.a == 42
# Parent does not automatically lead to a datatype:
assert obj.get_property_metadata("ref").datatype is "bla"
assert obj.ref.has_parent("bla") is True
# Unresolved Reference:
r = db.Record()
r.add_property(name="ref", value=27, datatype="bla")
obj = convert_to_python_object(r)
# Parent does not automatically lead to a datatype:
assert obj.get_property_metadata("ref").datatype is "bla"
assert isinstance(obj.ref, CaosDBPythonUnresolvedReference)
assert obj.ref.id == 27
def test_resolve_references():
r = db.Record()
r.add_property(name="ref", value=27, datatype="bla")
r.add_property(name="ref_false", value=27) # this should be interpreted as integer property
obj = convert_to_python_object(r)
ref = db.Record(id=27)
ref.add_property(name="a", value=57)
unused_ref1 = db.Record(id=28)
unused_ref2 = db.Record(id=29)
unused_ref3 = db.Record(name="bla")
references = db.Container().extend([
unused_ref1, ref, unused_ref2, unused_ref3])
# Nothing is going to be resolved:
obj.resolve_references(False, db.Container())
assert isinstance(obj.ref, CaosDBPythonUnresolvedReference)
assert obj.ref.id == 27
assert obj.ref_false == 27
# deep == True does not help:
obj.resolve_references(True, db.Container())
assert isinstance(obj.ref, CaosDBPythonUnresolvedReference)
assert obj.ref.id == 27
# But adding the reference container will do:
obj.resolve_references(False, references)
assert not isinstance(obj.ref, CaosDBPythonUnresolvedReference)
assert isinstance(obj.ref, CaosDBPythonRecord)
assert obj.ref.id == 27
assert obj.ref.a == 57
# Datatypes will not automatically be set:
assert obj.ref.get_property_metadata("a").datatype is None
# Test deep resolve:
ref2 = db.Record(id=225)
ref2.add_property(name="c", value="test")
ref.add_property(name="ref", value=225, datatype="bla")
obj = convert_to_python_object(r)
assert isinstance(obj.ref, CaosDBPythonUnresolvedReference)
obj.resolve_references(False, references)
assert not isinstance(obj.ref, CaosDBPythonUnresolvedReference)
assert isinstance(obj.ref.ref, CaosDBPythonUnresolvedReference)
assert obj.ref.ref.id == 225
# Will not help, because ref2 is missing in container:
obj.resolve_references(True, references)
assert not isinstance(obj.ref, CaosDBPythonUnresolvedReference)
assert isinstance(obj.ref.ref, CaosDBPythonUnresolvedReference)
assert obj.ref.ref.id == 225
references.append(ref2)
obj.resolve_references(False, references)
assert not isinstance(obj.ref, CaosDBPythonUnresolvedReference)
assert isinstance(obj.ref.ref, CaosDBPythonUnresolvedReference)
assert obj.ref.ref.id == 225
obj.resolve_references(True, references)
assert not isinstance(obj.ref, CaosDBPythonUnresolvedReference)
assert not isinstance(obj.ref.ref, CaosDBPythonUnresolvedReference)
assert obj.ref.ref.c == "test"
# Test circular dependencies:
ref2.add_property(name="ref", value=27, datatype="bla")
obj = convert_to_python_object(r)
obj.resolve_references(True, references)
assert obj.ref.ref.ref == obj.ref
def equal_entities(r1, r2):
res = compare_entities(r1, r2)
if len(res) != 2:
return False
for i in range(2):
if len(res[i]["parents"]) != 0 or len(res[i]["properties"]) != 0:
return False
return True
def test_conversion_to_entity():
r = db.Record()
r.add_parent("bla")
r.add_property(name="a", value=42)
r.add_property(name="b", value="test")
obj = convert_to_python_object(r)
rconv = convert_to_entity(obj)
assert equal_entities(r, rconv)
# With a reference:
r_ref = db.Record()
r_ref.add_parent("bla")
r_ref.add_property(name="a", value=42)
r = db.Record()
r.add_property(name="ref", value=r_ref)
obj = convert_to_python_object(r)
rconv = convert_to_entity(obj)
assert (rconv.get_property("ref").value.get_property("a").value
== r.get_property("ref").value.get_property("a").value)
# TODO: add more tests here
def test_empty():
r = db.Record()
obj = convert_to_python_object(r)
assert isinstance(obj, CaosDBPythonRecord)
assert len(obj.get_properties()) == 0
assert len(obj.get_parents()) == 0
rconv = convert_to_entity(obj)
assert len(rconv.properties) == 0
def test_wrong_entity_for_file():
r = db.Record()
r.path = "test.dat"
r.file = "/local/path/test.dat"
assert r.path == "test.dat"
assert r.file == "/local/path/test.dat"
with pytest.raises(RuntimeError):
obj = convert_to_python_object(r)
def test_serialization():
r = db.Record(id=5, name="test", description="ok")
r.add_property(name="v", value=15, datatype=db.INTEGER, unit="kpx",
importance="RECOMMENDED")
obj = convert_to_python_object(r)
text = str(obj)
teststrs = ["description: ok", "id: 5", "datatype: INTEGER",
"importance: RECOMMENDED", "unit: kpx", "name: test", "v: 15"]
for teststr in teststrs:
assert teststr in text
def test_files():
# empty file:
r = db.File()
obj = convert_to_python_object(r)
print(type(obj))
assert isinstance(obj, CaosDBPythonFile)
assert len(obj.get_properties()) == 0
assert len(obj.get_parents()) == 0
rconv = convert_to_entity(obj)
assert len(rconv.properties) == 0
r.path = "test.dat"
r.file = "/local/path/test.dat"
obj = convert_to_python_object(r)
assert r.path == "test.dat"
assert r.file == "/local/path/test.dat"
assert isinstance(obj, CaosDBPythonFile)
assert obj.path == "test.dat"
assert obj.file == "/local/path/test.dat"
print(obj)
assert "path: test.dat" in str(obj)
assert "file: /local/path/test.dat" in str(obj)