Skip to content
Snippets Groups Projects
Select Git revision
  • d1ca617473ee9b28240c0f51708bec5ce1ae3525
  • main default protected
  • dev
  • f-docs-pylib
  • f-parse-value
  • f-compare
  • f-string-ids
  • f-217-set-special-property
  • f-filesystem-import
  • f-filesystem-link
  • f-filesystem-directory
  • f-filesystem-core
  • f-filesystem-cleanup
  • f-check-merge-entities
  • f-compare-enid
  • f-select-subproperties
  • v0.18.0
  • v0.17.0
  • v0.16.0
  • v0.15.1
  • v0.15.0
  • v0.14.0
  • v0.13.2
  • v0.13.1
  • v0.13.0
  • linkahead-rename-step-2
  • linkahead-rename-step-1
  • v0.12.0
  • v0.11.2
  • v0.11.1
  • v0.11.0
  • v0.10.0
  • v0.9.0
  • v0.8.0
  • v0.7.4
  • v0.7.3
36 results

setup.py

Blame
  • 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