Select Git revision
Install_develop.rst
-
Joscha Schmiedt authoredJoscha Schmiedt authored
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