Select Git revision
test_base_table_exporter.py
Daniel Hornung authored
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_base_table_exporter.py 8.00 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
#
"""Test all functionality of the base exporter class that can be
tested without db connection.
"""
import json
import os
import caosdb as db
from pytest import raises
from caosadvancedtools import table_export as te
class DummyExporter(te.BaseTableExporter):
def __init__(self, export_dict, record=None,
raise_error_if_missing=False):
super().__init__(export_dict, record, raise_error_if_missing)
self.dummy = 27
def find_simple_test_entry(self):
return "bla"
def find_entry_with_strange_name(self):
return "blabla"
def find_function_with_error(self):
raise AttributeError("Can't find the value.")
def test_base():
"""test constructor of base class"""
empty = {}
my_exporter = te.BaseTableExporter(empty)
assert my_exporter.export_dict == {}
assert len(my_exporter.missing) == 0
assert my_exporter.info == {}
assert my_exporter.prepare_csv_export() == ""
def test_simple_record():
"""Test whether properties of simple record are found correctly."""
rec = db.Record(name="TestRecord")
rec.add_property(name="Test_Prop_1", value="bla")
rec.add_property(name="Test_Prop_2", value="blabla")
export_dict = {
"Test_Prop_1": {},
"Test_Prop_2": {"optional": True}
}
my_exporter = te.BaseTableExporter(export_dict=export_dict, record=rec)
assert my_exporter.record.name == rec.name
assert my_exporter.export_dict == export_dict
my_exporter.collect_information()
assert len(my_exporter.missing) == 0
assert my_exporter.info["Test_Prop_1"] == "bla"
assert my_exporter.info["Test_Prop_2"] == "blabla"
assert my_exporter.prepare_csv_export() == "bla,blabla"
assert my_exporter.prepare_csv_export(
delimiter='\t', print_header=True) == "Test_Prop_1\tTest_Prop_2\nbla\tblabla"
# remove optional entry from info
del my_exporter.info["Test_Prop_2"]
assert my_exporter.prepare_csv_export(skip_empty_optionals=True) == "bla"
assert my_exporter.prepare_csv_export(
delimiter='\t', print_header=True) == "Test_Prop_1\tTest_Prop_2\nbla\t"
# reload info, and delete mandatory entry
my_exporter.collect_information()
del my_exporter.info["Test_Prop_1"]
with raises(te.TableExportError) as exc:
my_exporter.prepare_csv_export()
assert "Test_Prop_1" in exc.value.msg
assert "Test_Prop_2" not in exc.value.msg
def test_broken_export_dicts():
# query but no record
export_dict = {
"Test_Prop_1": {"query": "SELECT Test_Prop_1 FROM Test_Record WITH ID={}"}
}
with raises(te.TableExportError):
my_exporter = te.BaseTableExporter(export_dict=export_dict)
# record without matching property or find function
export_dict = {"Test_Prop_1": {}}
rec = db.Record()
my_exporter = te.BaseTableExporter(export_dict=export_dict, record=rec)
with raises(te.TableExportError):
my_exporter.collect_information()
# query and function given
export_dict = {
"Test_Prop_1": {"query": "SELECT Test_Prop_1 FROM Test_Record WITH ID={}",
"find_func": "find_Test_Prop_1"}
}
with raises(te.TableExportError):
my_exporter = te.BaseTableExporter(export_dict=export_dict, record=rec)
# function not implemented
export_dict = {
"Test_Prop_1": {"find_func": "find_Test_Prop_1"}
}
with raises(te.TableExportError):
my_exporter = te.BaseTableExporter(export_dict)
# function not callable
export_dict = {
"Test_Prop_1": {"find_func": "dummy"}
}
with raises(te.TableExportError):
my_exporter = DummyExporter(export_dict)
# one good, one bad
export_dict = {
"Test_Prop_1": {"find_func": "find_simple_test_entry"},
"Test_Prop_2": {"find_func": "not_implemented"}
}
with raises(te.TableExportError) as exc:
my_exporter = DummyExporter(export_dict)
assert "Test_Prop_2" in exc.value.msg
assert "Test_Prop_1" not in exc.value.msg
def test_info_collection():
# guess correct function name
export_dict = {"simple_test_entry": {}}
my_exporter = DummyExporter(export_dict=export_dict,
raise_error_if_missing=True)
assert my_exporter.export_dict[
"simple_test_entry"]["find_func"] == "find_simple_test_entry"
my_exporter.collect_information()
assert my_exporter.info["simple_test_entry"] == "bla"
# use specific find function
export_dict["Not so simple"] = {
"find_func": "find_entry_with_strange_name"}
my_exporter = DummyExporter(export_dict=export_dict,
raise_error_if_missing=True)
my_exporter.collect_information()
assert my_exporter.info["simple_test_entry"] == "bla"
assert my_exporter.info["Not so simple"] == "blabla"
# mix functions and record
rec = db.Record()
rec.add_property(name="Test_Prop_1", value="blablabla")
export_dict["Test_Prop_1"] = {}
my_exporter = DummyExporter(export_dict=export_dict, record=rec,
raise_error_if_missing=True)
my_exporter.collect_information()
assert my_exporter.info["simple_test_entry"] == "bla"
assert my_exporter.info["Not so simple"] == "blabla"
assert my_exporter.info["Test_Prop_1"] == "blablabla"
# error in optional value
export_dict["optional_value"] = {
"find_func": "find_function_with_error",
"optional": True
}
my_exporter = DummyExporter(export_dict=export_dict, record=rec,
raise_error_if_missing=True)
my_exporter.collect_information()
assert "optional_value" not in my_exporter.info
# now error in mandatory value
del export_dict["optional_value"]
export_dict["mandatory_value"] = {
"find_func": "find_function_with_error"
}
my_exporter = DummyExporter(export_dict=export_dict, record=rec,
raise_error_if_missing=True)
with raises(te.TableExportError) as exc:
my_exporter.collect_information()
assert len(my_exporter.missing) == 1
assert "mandatory_value" in my_exporter.missing
assert exc.value.msg.split("\n")[1] == "mandatory_value"
# add explanation to error
export_dict["mandatory_value"]["error"] = "Explain the error"
my_exporter = DummyExporter(export_dict=export_dict, record=rec,
raise_error_if_missing=True)
with raises(te.TableExportError) as exc:
my_exporter.collect_information()
assert exc.value.msg.split(
"\n")[1] == "mandatory_value:\tExplain the error"
def test_json_import():
with open(os.path.join(os.path.dirname(__file__),
"dummy_export_dict.json")) as tmp:
export_dict = json.load(tmp)
rec = db.Record()
rec.add_property(name="Test_Prop_1", value="blablabla")
my_exporter = DummyExporter(export_dict=export_dict, record=rec,
raise_error_if_missing=True)
my_exporter.collect_information()
assert my_exporter.info["simple_test_entry"] == "bla"
assert my_exporter.info["Not so simple"] == "blabla"
assert my_exporter.info["Test_Prop_1"] == "blablabla"