test_json_schema_datamodel_parser.py

#
# This file is a part of the LinkAhead Project.
#
# Copyright (C) 2022 IndiScale GmbH <info@indiscale.com>
# Copyright (C) 2022 Florian Spreckelsen <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/>.
#

import os

import caosdb as db
from caosadvancedtools.models.parser import parse_model_from_json_schema


def _clear_db():
    ents = db.execute_query("FIND ENTITY WITH ID>99")
    if ents:
        ents.delete()


def setup_module():
    _clear_db()


def teardown_module():
    _clear_db()


def _load_and_sync(fname):
    """Load datamodel from json schema in fname and synchronize it without asking.

    """
    # @author Florian Spreckelsen
    # @date 2022-03-23
    fpath = os.path.join(os.path.dirname(os.path.abspath(__file__)), fname)
    model = parse_model_from_json_schema(fpath)
    model.sync_data_model(noquestion=True)


def test_json_parsed_datamodel():
    # @author Florian Spreckelsen
    # @date 2022-03-23

    _load_and_sync("test_datamodel.schema.json")

    # RecordType with atomic properties
    rt1 = db.execute_query(
        "FIND RECORDTYPE TestTypeWithAtomicProps", unique=True)
    assert rt1.description == "RecordType with scalar atomic properties"
    assert rt1.get_property("simple_text_prop") is not None
    assert rt1.get_property("simple_text_prop").datatype == db.TEXT
    assert rt1.get_importance("simple_text_prop") == db.OBLIGATORY

    assert rt1.get_property("IntegerProperty") is not None
    assert rt1.get_property("IntegerProperty").datatype == db.INTEGER
    assert rt1.get_importance("IntegerProperty") == db.RECOMMENDED

    assert rt1.get_property("double_prop") is not None
    assert rt1.get_property("double_prop").datatype == db.DOUBLE
    assert rt1.get_importance("double_prop") == db.OBLIGATORY
    assert (db.Property(name="double_prop").retrieve().description ==
            "Some generic double-valued property")

    further_props = [
        ("bool_prop", db.BOOLEAN),
        ("datetime_prop", db.DATETIME),
        ("date_prop", db.DATETIME)
    ]
    for name, dtype in further_props:
        assert rt1.get_property(name) is not None
        assert rt1.get_property(name).datatype == dtype
        assert rt1.get_importance(name) == db.RECOMMENDED

    # RecordType with references and enums
    rt2 = db.execute_query(
        "FIND RECORDTYPE TestTypeWithReferencesAndEnum", unique=True)
    assert rt2.get_property(rt1.name) is not None
    assert rt2.get_property(rt1.name).is_reference()
    assert rt2.get_property(rt1.name).name == rt1.name
    assert rt2.get_property(rt1.name).id == rt1.id

    other_ref_type = db.execute_query(
        "FIND RECORDTYPE OtherReference", unique=True)
    assert rt2.get_property(other_ref_type.name) is not None
    assert rt2.get_property(other_ref_type.name).is_reference()
    assert rt2.get_property(other_ref_type.name).name == other_ref_type.name
    assert rt2.get_property(other_ref_type.name).id == other_ref_type.id
    assert other_ref_type.description == "Some generic refernced RecordType"
    assert len(other_ref_type.properties) == 0

    named_ref_type = db.execute_query(
        "FIND RECORDTYPE NamedReference", unique=True)
    assert rt2.get_property(named_ref_type.name) is not None
    assert rt2.get_property(named_ref_type.name).is_reference()
    assert rt2.get_property(named_ref_type.name).name == named_ref_type.name
    assert rt2.get_property(named_ref_type.name).id == named_ref_type.id
    assert named_ref_type.get_property("simple_text_prop") is not None
    assert (named_ref_type.get_property("simple_text_prop").id ==
            rt1.get_property("simple_text_prop").id)
    assert (named_ref_type.get_property("simple_text_prop").datatype ==
            rt1.get_property("simple_text_prop").datatype)

    enums = {
        "string_enum": ["StringEnumA", "StringEnumB", "StringEnumC"],
        "NamedEnum": ["NameA", "NameB", "NameC"]
    }
    for enum_type_name, enum_names in enums.items():
        enum_type = db.execute_query(
            f"FIND RECORDTYPE {enum_type_name}", unique=True)
        assert len(enum_type.properties) == 0
        enum_records = db.execute_query(f"FIND RECORD {enum_type_name}")
        assert len(enum_records) == len(enum_names)
        for rec in enum_records:
            assert rec.name in enum_names
        assert rt2.get_property(enum_type_name) is not None
        assert rt2.get_property(enum_type_name).is_reference()
        assert rt2.get_property(enum_type_name).name == enum_type.name
        assert rt2.get_property(enum_type_name).id == enum_type.id

    # Recordtype with lists
    rt3 = db.execute_query("FIND RECORDTYPE TestTypeWithLists", unique=True)
    assert rt3.get_property("string_list") is not None
    assert rt3.get_property("string_list").datatype == db.LIST(db.TEXT)
    string_list_prop = db.Property(name="string_list").retrieve()
    assert string_list_prop.description == "A list of words"
    assert string_list_prop.datatype == db.LIST(db.TEXT)
    assert string_list_prop.id == rt3.get_property("string_list").id

    assert rt3.get_property("NamedIntList") is not None
    assert rt3.get_property("NamedIntList").datatype == db.LIST(db.INTEGER)

    # This is a list of a plain references to a specific type
    list_rt = db.execute_query("FIND RECORDTYPE ListRecordType", unique=True)
    assert len(list_rt.properties) == 0
    assert rt3.get_property(list_rt.name) is not None
    assert rt3.get_property(list_rt.name).is_reference()
    assert rt3.get_property(list_rt.name).datatype == db.LIST(list_rt)
    assert rt3.get_property(list_rt.name).id == list_rt.id

    # This is a list property of its own, referencing another separate RT
    referenced_list_rt = db.execute_query(
        "FIND RECORDTYPE ReferencedListTypeWithName", unique=True)
    assert referenced_list_rt.description == "Referenced by a named list-of-references property"
    assert referenced_list_rt.get_property("double_prop") is not None
    assert (referenced_list_rt.get_property("double_prop").id ==
            rt1.get_property("double_prop").id)
    assert rt3.get_property("NamedReferenceList") is not None
    assert rt3.get_property("NamedReferenceList").is_reference()
    assert rt3.get_property(
        "NamedReferenceList").datatype == db.LIST(referenced_list_rt)
    assert rt3.get_property("NamedReferenceList").id != referenced_list_rt.id

    enum_type = db.execute_query("FIND RECORDTYPE ListNumberEnum", unique=True)
    assert len(enum_type.properties) == 0
    enum_names = ["1.1", "2.2", "3.3"]
    enum_records = db.execute_query("FIND RECORD ListNumberEnum")
    assert len(enum_records) == len(enum_names)
    for rec in enum_records:
        assert rec.name in enum_names
    assert rt3.get_property(enum_type.name) is not None
    assert rt3.get_property(enum_type.name).datatype == db.LIST(enum_type)
    assert rt3.get_property(enum_type.name).id == enum_type.id