# 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)