Select Git revision
Install_develop.rst
-
Joscha Schmiedt authored
(not functional yet, only how it should work later on)
Joscha Schmiedt authored(not functional yet, only how it should work later on)
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