Skip to content
Snippets Groups Projects
Select Git revision
  • 90dd774ae455d355f74844e355cd1ea92a11c87b
  • main default protected
  • f-sss4grpc
  • dev
  • 108-implement-rpc-call-for-server-side-scripting
  • f-windows-conan-create
  • f-to-string
  • f-update-requirements
  • f-related-projects
  • f-role
  • f-remote-path
  • f-rel-path
  • f-consol-message
  • v0.3.0
  • v0.2.2
  • v0.2.1
  • v0.2.0
  • v0.1.2
  • v0.1.1
  • v0.1
  • v0.0.19
  • v0.0.18
  • v0.0.16
  • v0.0.15
  • v0.0.10
  • v0.0.9
  • v0.0.8
  • v0.0.7
  • v0.0.6
  • v0.0.5
  • v0.0.4
  • v0.0.3
  • v0.0.2
33 results

Install_develop.rst

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    test_select.py 21.11 KiB
    # 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