Select Git revision
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_base_table_exporter_integration.py 5.58 KiB
#!/usr/bin/env python3
# encoding: utf-8
#
# ** header v3.0
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2020 Indiscale GmbH <info@indiscale.com>
# Copyright (C) 2020 Florian Sprecklelsen <f.spreckelsen@indiscale.com>
#
# 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
#
import caosdb as db
import pytest
from caosadvancedtools import table_export as te
class IntegrationExporter(te.BaseTableExporter):
def __init__(self, export_dict, rec_id,
raise_error_if_missing=False):
self.record = db.execute_query(
"FIND Record WITH ID={}".format(rec_id), unique=True)
super().__init__(export_dict, record=self.record,
raise_error_if_missing=raise_error_if_missing)
def find_Test_Property_1(self):
tmp = db.execute_query(
"FIND Record WITH ID={}".format(self.record.id),
unique=True)
return tmp.get_property("Test_Property_1").value
def find_more_complicated_value(self):
tp1 = self.record.get_property("Test_Property_1").value
tp2 = db.execute_query(
"SELECT Test_Property_2 FROM Test_Type_2 WHICH IS"
" REFERENCED BY A Test_Type_1 WITH ID={}".format(
self.record.id),
unique=True).get_property_values("Test_Property_2")[0]
return tp1+tp2
def insert_entities():
"""Insert four test records and the corresponding properties and
record types.
"""
rt1 = db.RecordType(name="Test_Type_1").insert()
rt2 = db.RecordType(name="Test_Type_2").insert()
prop1 = db.Property(name="Test_Property_1", datatype=db.DOUBLE).insert()
prop2 = db.Property(name="Test_Property_2", datatype=db.DOUBLE).insert()
rec1 = db.Record(name="Test_Record_1").add_parent(rt1)
rec1.add_property(name="Test_Property_1", value=1.0)
rec2 = db.Record(name="Test_Record_2").add_parent(rt2)
rec2.add_property(name="Test_Property_2", value=2.0).insert()
rec1.add_property(name="Test_Record_2", value=rec2.id,
datatype=db.REFERENCE).insert()
rec3 = db.Record(name="Test_Record_3").add_parent(rt1)
rec3.add_property(name="Test_Property_1", value=11.0)
rec4 = db.Record(name="Test_Record_4").add_parent(rt2)
rec4.add_property(name="Test_Property_2", value=12.0).insert()
rec3.add_property(name="Test_Record_2", value=rec4.id,
datatype=db.REFERENCE).insert()
return rec1, rec2, rec3, rec4
def setup_module():
"""Clear all test entities"""
try:
db.execute_query("FIND ENTITY Test*").delete()
except BaseException:
pass
@pytest.fixture(autouse=True)
def setup():
"""Same as module setup."""
setup_module()
yield None
setup_module()
def teardown():
"""Delete everything again."""
setup_module()
def test_find_functions():
rec1, rec2, rec3, rec4 = insert_entities()
export_dict = {
"Test_Property_1": {},
"Other value": {
"find_func": "find_more_complicated_value"
}
}
my_exporter1 = IntegrationExporter(
export_dict, rec1.id, raise_error_if_missing=True)
assert rec1.name == my_exporter1.record.name
my_exporter1.collect_information()
assert my_exporter1.info["Test_Property_1"] == rec1.get_property(
"Test_Property_1").value
assert my_exporter1.info["Other value"] == 3
assert not my_exporter1.missing
# again with other record
my_exporter2 = IntegrationExporter(
export_dict, rec3.id, raise_error_if_missing=True)
my_exporter2.collect_information()
assert my_exporter2.info["Test_Property_1"] == rec3.get_property(
"Test_Property_1").value
assert my_exporter2.info["Other value"] == 23
assert not my_exporter2.missing
def test_queries():
rec1, rec2, _, _ = insert_entities()
# no explicit functions since only `query` key is used,
# so works with BaseTableExporter
export_dict = {
"Test_Property_1": {},
"Test_Property_2": {
"query": "SELECT Test_Property_2 FROM Test_Type_2 WHICH IS REFERENCED BY A Test_Type_1 WITH ID={}",
"selector": "Test_Property_2"
}
}
my_exporter = te.BaseTableExporter(
export_dict=export_dict, record=rec1, raise_error_if_missing=True)
my_exporter.collect_information()
assert my_exporter.info["Test_Property_1"] == rec1.get_property(
"Test_Property_1").value
assert my_exporter.info["Test_Property_2"] == rec2.get_property(
"Test_Property_2").value
# test guessing of selector
del (export_dict["Test_Property_2"]["selector"])
my_exporter = te.BaseTableExporter(
export_dict=export_dict, record=rec1, raise_error_if_missing=True)
assert my_exporter.export_dict["Test_Property_2"]["selector"] == "Test_Property_2"
my_exporter.collect_information()
assert my_exporter.info["Test_Property_1"] == rec1.get_property(
"Test_Property_1").value
assert my_exporter.info["Test_Property_2"] == rec2.get_property(
"Test_Property_2").value