test_select.py

# encoding: utf-8
#
# ** header v3.0
# 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) 2021 IndiScale GmbH <info@indiscale.com>
# Copyright (C) 2021 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/>.
#
# ** end header
#
"""Created on 11.10.2016.

@author: tf
"""
from pytest import mark, raises

import caosdb as db


def setup_module():
    teardown_module()
    db.Property(name="TestPropertyOne", datatype=db.TEXT).insert()
    db.Property(
        name="TestPropertyTwo",
        description="Desc2",
        datatype=db.TEXT).insert()
    db.RecordType(
        name="TestRecordType",
        description="DescRecTy").add_property(
        name="TestPropertyOne", value="v1").add_property(
            name="TestPropertyTwo", value="v2").insert()
    rt_house = db.RecordType("TestHouse", description="TestHouseDesc").insert()
    db.RecordType("TestHousePart", description="TestHousePartDesc").insert()
    db.RecordType("TestWindow").add_parent("TestHousePart").insert()
    db.RecordType("TestDoor").add_parent("TestHousePart").insert()
    rt_person = db.RecordType("TestPerson",
                              description="TestPersonDesc").insert()
    db.RecordType("TestParty", description="TestPartyDesc").insert()
    db.Property("TestHouseProperty", datatype=db.TEXT).insert()
    db.Property("TestHeight", description="TestHeightDesc", datatype=db.DOUBLE,
                unit="ft").add_parent("TestHouseProperty").insert()
    db.Property("TestDate", description="TestDateDesc",
                datatype=db.DATETIME).insert()

    door = db.Record("Door1",
                     description="Door1Desc").add_parent("TestDoor")
    door.add_property("TestHeight", "21.5", unit="ft")
    door.insert()
    window = db.Record("Window1",
                       description="Window1Desc").add_parent("TestWindow")
    window.add_property("TestHeight", 20.5, unit="ft")
    window.insert()

    owner = db.Record("The Queen").add_parent("TestPerson").insert()

    house = db.Record("Buckingham Palace")
    house.description = "A rather large house"
    house.add_parent("TestHouse")
    house.add_property(rt_person, name="TestOwner", value=owner)
    house.add_property("TestWindow", [window], datatype=db.LIST("TestWindow"))
    house.add_property("TestDoor", door)
    house.insert()

    g1 = db.Record().add_parent("TestPerson").insert()
    g2 = db.Record().add_parent("TestPerson").insert()
    g3 = db.Record().add_parent("TestPerson").insert()

    party = db.Record(
        "Diamond Jubilee of Elizabeth II").add_parent("TestParty")
    party.add_property(rt_house, name="Location", value=house)
    party.add_property("TestDate", "2012-02-06")
    party.add_property(rt_person, datatype=db.LIST(rt_person), name="Guests",
                       value=[g1, g2, g3])
    party.insert()


def teardown_module():
    d = db.execute_query("FIND Test*")

    if len(d) > 0:
        d.delete()


def test_id1():
    p1 = db.execute_query("FIND TestPropertyOne", unique=True)
    assert p1.is_valid() is True
    assert p1.name is not None
    assert p1.datatype is not None
    assert p1.description is None

    p1_c = db.execute_query("SELECT id FROM TestPropertyOne", unique=True)
    assert p1_c.is_valid() is True
    assert p1_c.id == p1.id
    assert p1_c.name is not None, "Name is always included"
    assert p1_c.datatype is None
    assert p1_c.description is None


def test_id2():
    p2 = db.execute_query("FIND TestPropertyTwo", unique=True)
    assert p2.is_valid() is True
    assert p2.name is not None
    assert p2.datatype is not None
    assert p2.description is not None

    p2_c = db.execute_query("SELECT id FROM TestPropertyTwo", unique=True)
    assert p2_c.is_valid() is True
    assert p2_c.id == p2.id
    assert p2_c.name is not None, "Name is always included"
    assert p2_c.datatype is None
    assert p2_c.description is None


def test_id3():
    p3s = db.execute_query("SELECT description FROM TestProperty*")
    assert len(p3s) == 2

    for e in p3s:
        assert e.id is not None


def test_name1():
    p3s = db.execute_query("SELECT description FROM TestProperty*")
    assert len(p3s) == 2

    for e in p3s:
        assert e.name is not None


def test_name2():
    p3s = db.execute_query("SELECT name FROM TestProperty*")
    assert len(p3s) == 2

    for e in p3s:
        assert e.name is not None
        assert e.description is None


def test_name_from_entity():
    db.administration.set_server_property(
        "QUERY_FILTER_ENTITIES_WITHOUT_RETRIEVE_PERMISSIONS",
        "FALSE")
    p3s = db.execute_query("SELECT name FROM ENTITY", flags={"cache": "false"})
    db.administration.set_server_property(
        "QUERY_FILTER_ENTITIES_WITHOUT_RETRIEVE_PERMISSIONS",
        "TRUE")
    p3s = db.execute_query("SELECT name FROM ENTITY", flags={"cache": "false"})


def test_multi1():
    p1 = db.execute_query(
        "SELECT id, name, description FROM TestPropertyOne",
        unique=True)
    assert p1.id is not None
    assert p1.name == "TestPropertyOne"
    assert p1.description is None

    p2 = db.execute_query(
        "SELECT id, name, description FROM TestPropertyTwo",
        unique=True)
    assert p2.id is not None
    assert p2.name == "TestPropertyTwo"
    assert p2.description == "Desc2"


def test_sub1():
    rt = db.execute_query("FIND TestRecordType", unique=True)
    assert rt.id is not None
    assert rt.name is not None
    assert rt.get_property("TestPropertyOne") is not None

    rt = db.execute_query(
        "SELECT TestPropertyOne FROM TestRecordType",
        unique=True)
    assert rt.id is not None
    assert rt.name is not None
    assert rt.get_property("TestPropertyOne") is not None
    assert rt.get_property("TestPropertyOne").value == "v1"
    assert rt.get_property("TestPropertyTwo") is None


def test_sub2():
    rt = db.execute_query(
        "SELECT TestPropertyTwo.description FROM TestRecordType",
        unique=True)
    assert rt.id is not None
    assert rt.name is not None
    assert rt.get_property("TestPropertyTwo") is not None
    assert rt.get_property("TestPropertyTwo").description == "Desc2"
    assert rt.get_property("TestPropertyTwo").datatype is None
    assert rt.get_property("TestPropertyTwo").value is None


def test_subref():
    door = db.execute_query("FIND RECORD TestDoor", unique=True)
    s = db.execute_query("SELECT name, TestDoor.TestHeight.value, "
                         "TestDoor.TestHeight.unit FROM RECORD TestHouse")

    assert len(s) == 1

    row = s.get_property_values("name", "TestDoor")[0]
    assert row[0] == "Buckingham Palace"
    assert row[1] == door.id

    row = s.get_property_values("name", ("TestDoor", "TestHeight"))[0]
    assert row[0] == "Buckingham Palace"
    assert row[1] == 21.5

    row = s.get_property_values(
        "name", ("TestDoor", "TestHeight", "unit"))[0]
    assert row[0] == "Buckingham Palace"
    assert row[1] == "ft"


def test_subref_list():
    window = db.execute_query("FIND RECORD TestWindow", unique=True)
    s = db.execute_query("SELECT name, TestWindow.TestHeight.value, "
                         "TestWindow.TestHeight.unit FROM RECORD TestHouse")

    assert len(s) == 1

    row = s.get_property_values("name", "TestWindow")[0]
    assert row[0] == "Buckingham Palace"
    assert row[1][0].id == window.id

    # current limitation of get_property_values - no lists
    # row = s.get_property_values("name", ("TestWindow", "TestHeight"))[0]
    # assert row[0] == "Buckingham Palace"
    # assert row[1] == 20.5
    assert row[1][0].get_property("TestHeight").value == 20.5

    # current limitation of get_property_values - no lists
    # row = s.get_property_values(
    # "name", ("TestWindow", "TestHeight", "unit"))[0]
    # assert row[0] == "Buckingham Palace"
    # assert row[1] == "ft"
    assert row[1][0].get_property("TestHeight").unit == "ft"


def test_subref_deep():
    p = db.execute_query(
        "SELECT name, Testdate, location, location.TestDoor.Testheight FROM "
        "RECORD TestParty", unique=True)
    row = p.get_property_values("name", "Testdate",
                                ("location", "Testdoor", "Testheight"))
    assert row == ("Diamond Jubilee of Elizabeth II", "2012-02-06", 21.5)


def test_subref_deep_list():
    p = db.execute_query(
        "SELECT name, Testdate, location, location.TestWindow.Testheight FROM "
        "RECORD TestParty", unique=True)

    # current limitation of get_property_values - no lists
    # row = p.get_property_values("name", "Testdate",
    # ("location", "Testwindow", "Testheight"))
    # assert row == ("Diamond Jubilee of Elizabeth II", "2012-02-06", 20.5)
    assert p.get_property_values(("location", "TestWindow"))[
        0][0].get_property("TestHeight").value == 20.5


def test_select_list():
    guests = db.execute_query(
        "FIND RECORD TestPerson WHICH IS REFERENCED BY TestParty")
    s = db.execute_query("SELECT guests FROM RECORD TestParty", unique=True)

    column = s.get_property_values("guests")[0]
    print(s)
    assert len(column) == len(guests)

    for eid in [e.id for e in guests]:
        assert eid in column


def test_select_unit():
    s = db.execute_query("SELECT unit FROM RECORD TestHouse", unique=True)
    column = s.get_property_values("unit")
    assert column == (None,)

    s = db.execute_query("SELECT unit FROM PROPERTY TestHeight", unique=True)
    column = s.get_property_values("unit")
    assert column == ("ft",)

    s = db.execute_query("SELECT TestDoor.TestHeight.unit FROM "
                         "RECORD TestHouse", unique=True)
    column = s.get_property_values(("TestDoor", "TestHeight", "unit"))
    assert column == ("ft",)

    s = db.execute_query("SELECT TestHeight.unit.TestDoor FROM "
                         "RECORD TestDoor", unique=True)
    column = s.get_property_values(("TestHeight", "unit", "TestDoor"))
    assert column == (None,)


def test_select_unit_list():
    s = db.execute_query("SELECT unit FROM RECORD TestHouse", unique=True)
    column = s.get_property_values("unit")
    assert column == (None,)

    s = db.execute_query("SELECT unit FROM PROPERTY TestHeight", unique=True)
    column = s.get_property_values("unit")
    assert column == ("ft",)

    s = db.execute_query("SELECT TestWindow.TestHeight.unit FROM "
                         "RECORD TestHouse", unique=True)
    # current limitation of get_property_values - no lists
    # column = s.get_property_values(("TestWindow", "TestHeight", "unit"))
    # assert column == ("ft",)
    assert s.get_property_values(("TestWindow"))[
        0][0].get_property("TestHeight").unit == "ft"


def test_select_description():
    s = db.execute_query("SELECT description FROM RECORD TestPerson")
    column = s.get_property_values("description")
    assert column == [(None,), (None,), (None,), (None,)]

    s = db.execute_query("SELECT description"
                         "FROM RECORD TestHouse", unique=True)
    column = s.get_property_values(("description"))
    assert column == ("A rather large house",)

    s = db.execute_query("SELECT location.description"
                         "FROM RECORD TestParty", unique=True)
    column = s.get_property_values(("location", "description"))
    assert column == ("A rather large house",)

    s = db.execute_query("SELECT TestHeight.description FROM "
                         "RECORD TestDoor", unique=True)
    column = s.get_property_values(("TestHeight", "description"))
    assert column == ('TestHeightDesc',)

    s = db.execute_query("SELECT TestDoor.TestHeight.description FROM "
                         "RECORD TestHouse", unique=True)
    column = s.get_property_values(("TestDoor", "TestHeight", "description"))
    assert column == ('TestHeightDesc',)

    s = db.execute_query("SELECT TestHeight.description.TestDoor FROM "
                         "RECORD TestDoor", unique=True)
    column = s.get_property_values(("TestHeight", "description", "TestDoor"))
    assert column == (None,)


def test_select_description_list():
    s = db.execute_query("SELECT description FROM RECORD TestPerson")
    column = s.get_property_values("description")
    assert column == [(None,), (None,), (None,), (None,)]

    s = db.execute_query("SELECT description"
                         "FROM RECORD TestHouse", unique=True)
    column = s.get_property_values(("description"))
    assert column == ("A rather large house",)

    s = db.execute_query("SELECT location.description"
                         "FROM RECORD TestParty", unique=True)
    column = s.get_property_values(("location", "description"))
    assert column == ("A rather large house",)

    s = db.execute_query("SELECT TestHeight.description FROM "
                         "RECORD TestWindow", unique=True)
    column = s.get_property_values(("TestHeight", "description"))
    assert column == ('TestHeightDesc',)

    s = db.execute_query("SELECT TestWindow.TestHeight.description FROM "
                         "RECORD TestHouse", unique=True)
    # current limitation of get_property_values - no lists
    # column = s.get_property_values(("TestWindow", "TestHeight", "description"))
    # assert column == ('TestHeightDesc',)
    assert s.get_property_values(("TestWindow"))[0][0].get_property(
        "TestHeight").description == "TestHeightDesc"


def test_select_id():
    house_id = db.execute_query("FIND RECORD TestHouse", unique=True).id
    s = db.execute_query("SELECT id FROM RECORD TestHouse", unique=True)
    column = s.get_property_values("id")
    assert column == (house_id,)

    s = db.execute_query(
        "SELECT location.id FROM RECORD TestHouse",
        unique=True)
    column = s.get_property_values("id")
    assert column == (house_id,)

    height_id = db.execute_query("FIND PROPERTY TestHeight", unique=True).id
    s = db.execute_query("SELECT id FROM PROPERTY TestHeight", unique=True)
    column = s.get_property_values("id")
    assert column == (height_id,)

    s = db.execute_query("SELECT TestDoor.TestHeight.id FROM "
                         "RECORD TestHouse", unique=True)
    column = s.get_property_values(("TestDoor", "TestHeight", "id"))
    assert column == (height_id,)

    s = db.execute_query("SELECT TestHeight.id.TestDoor FROM "
                         "RECORD TestDoor", unique=True)
    column = s.get_property_values(("TestHeight", "id", "TestDoor"))
    assert column == (None,)


def test_select_id_list():
    house_id = db.execute_query("FIND RECORD TestHouse", unique=True).id
    s = db.execute_query("SELECT id FROM RECORD TestHouse", unique=True)
    column = s.get_property_values("id")
    assert column == (house_id,)

    s = db.execute_query(
        "SELECT location.id FROM RECORD TestHouse",
        unique=True)
    column = s.get_property_values("id")
    assert column == (house_id,)

    height_id = db.execute_query("FIND PROPERTY TestHeight", unique=True).id
    s = db.execute_query("SELECT id FROM PROPERTY TestHeight", unique=True)
    column = s.get_property_values("id")
    assert column == (height_id,)

    s = db.execute_query("SELECT TestWindow.TestHeight.id FROM "
                         "RECORD TestHouse", unique=True)
    # current limitation of get_property_values - no lists
    # column = s.get_property_values(("TestWindow", "TestHeight", "id"))
    # assert column == (height_id,)
    assert s.get_property_values(("TestWindow"))[0][0].get_property(
        "TestHeight").id == height_id


def test_select_name():
    s = db.execute_query("SELECT name FROM RECORD TestHouse", unique=True)
    column = s.get_property_values("name")
    assert column == ("Buckingham Palace",)

    s = db.execute_query("SELECT location.name FROM RECORD TestHouse",
                         unique=True)
    column = s.get_property_values("name")
    assert column == ("Buckingham Palace",)

    s = db.execute_query("SELECT name FROM PROPERTY TestHeight", unique=True)
    column = s.get_property_values("name")
    assert column == ("TestHeight",)

    s = db.execute_query("SELECT TestDoor.TestHeight.name FROM "
                         "RECORD TestHouse", unique=True)
    column = s.get_property_values(("TestDoor", "TestHeight", "name"))
    assert column == ("TestHeight",)

    s = db.execute_query("SELECT TestHeight.name.TestDoor FROM "
                         "RECORD TestDoor", unique=True)
    column = s.get_property_values(("TestHeight", "name", "TestDoor"))
    assert column == (None,)


def test_select_name_list():
    s = db.execute_query("SELECT name FROM RECORD TestHouse", unique=True)
    column = s.get_property_values("name")
    assert column == ("Buckingham Palace",)

    s = db.execute_query("SELECT location.name FROM RECORD TestHouse",
                         unique=True)
    column = s.get_property_values("name")
    assert column == ("Buckingham Palace",)

    s = db.execute_query("SELECT name FROM PROPERTY TestHeight", unique=True)
    column = s.get_property_values("name")
    assert column == ("TestHeight",)

    s = db.execute_query("SELECT TestWindow.TestHeight.name FROM "
                         "RECORD TestHouse", unique=True)
    # current limitation of get_property_values - no lists
    # column = s.get_property_values(("TestWindow", "TestHeight", "name"))
    # assert column == ("TestHeight",)
    assert s.get_property_values(("TestWindow"))[0][0].get_property(
        "TestHeight").name == "TestHeight"


def test_select_with_subtyping_semantics():
    s = db.execute_query(
        "SELECT name FROM RECORD TestHouse WITH TestHousePart",
        unique=True)
    column = s.get_property_values("name")
    assert column == ("Buckingham Palace",)

    s = db.execute_query(
        "SELECT TestDoor.TestHeight FROM RECORD TestHouse WITH TestHousePart",
        unique=True)
    column = s.get_property_values(("TestDoor", "TestHeight"))
    assert column == (21.5,)

    s = db.execute_query(
        "SELECT TestHousePart.TestHeight FROM RECORD TestHouse WITH TestHousePart",
        unique=True)
    column = s.get_property_values(("TestHousePart", "TestHeight"))
    # this is a current limitation of get_property_values which will only
    # return the value of the first matching property
    # assert column == ([[20.5], 21.5,])

    assert len(s.properties) == 2
    # both the door and the window have been returned
    print(s.properties[0].value)
    assert (s.properties[0].name,
            s.properties[1].name) == ("TestHousePart", "TestHousePart")
    assert (s.properties[0].value[0].properties[0].value,
            s.properties[1].value.properties[0].value) == (20.5, 21.5)


def test_select_with_subtyping_semantics_second_level():
    s = db.execute_query(
        "SELECT TestHousePart.TestHouseProperty FROM RECORD TestHouse WITH TestHousePart",
        unique=True)

    assert len(s.properties) == 2
    # both the door and the window have been returned
    assert (s.properties[0].name,
            s.properties[1].name) == ("TestHousePart", "TestHousePart")
    assert (s.properties[0].value[0].properties[0].value,
            s.properties[1].value.properties[0].value) == (20.5, 21.5)


def test_select_with_subtyping_semantics_and_name_duplicates():
    db.Property(name="TestHousePart", description="This is a duplicate",
                datatype=db.TEXT).insert(unique=False)
    test_select_with_subtyping_semantics()


def test_select_update():
    """Ensure that the result of a SELECT query can't be used for an
    update (which might result in data loss).

    """
    select_result = db.execute_query("SELECT name FROM RECORD TestHouse")[0]

    with raises(db.CaosDBException):
        select_result.update()


@mark.xfail(reason="see pylib issue #48 and server issue #155")
def test_select_update_with_parent():
    """Ensure that even with a valid parent, the result of a SELECT query
    can't be updated.

    """

    test_house_rt = db.RecordType(name="TestHouse").retrieve()
    select_result = db.execute_query("SELECT name FROM RECORD TestHouse")[0]
    select_result.add_parent(test_house_rt)

    with raises(db.CaosDBException):
        select_result.update()


@mark.xfail(reason="see pylib issue #48 and server issue #155")
def test_select_update_with_force():
    """The update of the result of a SELECT query  may be forced."""
    test_house_rt = db.RecordType(name="TestHouse").retrieve()
    select_result = db.execute_query("SELECT name FROM RECORD TestHouse")[0]
    select_result.add_parent(test_house_rt)

    # TODO: The syntax may change here depending on the implementation
    # of caosdb-pylib#48
    select_result.update(force_incomplete=True)

    # only name has been selected and updated, so no properties remain:
    rec = db.Record(name=select_result.name).retrieve()
    assert len(rec.properties) == 0