Skip to content
Snippets Groups Projects
Commit 9f9a1ae4 authored by Timm Fitschen's avatar Timm Fitschen
Browse files

TST: SELECT QUERY for properties of referenced entities.

parent fcc6489c
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
CHANGELOG.md
+ 2
− 0
View file @ 9f9a1ae4
......@@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added (for new features)
* Tests for deeply nested SELECT queries
- Tests for [#62](https://gitlab.com/caosdb/caosdb-server/-/issues/62)
* Tests for One-time Authentication Tokens
* Test for [caosdb-pylib#31](https://gitlab.com/caosdb/caosdb-pylib/-/issues/31)
* Tests for [caosdb-server#62](https://gitlab.com/caosdb/caosdb-server/-/issues/62)
......
tests/test_datatype.py
+ 49
− 44
View file @ 9f9a1ae4
......@@ -26,8 +26,7 @@
@author: tf
"""
import caosdb as db
# @UnresolvedImport
from nose.tools import nottest, assert_raises, with_setup, assert_true, assert_equal
from pytest import raises
def setup():
......@@ -44,24 +43,20 @@ def teardown():
pass
@with_setup(setup, teardown)
def test_override_with_non_existing_ref():
rt1 = db.RecordType("TestRecordType1").insert()
rt2 = db.RecordType("TestRecordType2").insert()
db.Property("TestProperty", datatype=db.TEXT).insert()
with assert_raises(db.TransactionError) as cm:
with raises(db.TransactionError) as cm:
db.Record(
name="TestRecord").add_property(
name="TestProperty",
datatype=rt2,
value="Non-Existing").add_parent(rt1).insert()
assert_equal(
cm.exception.get_errors()[0].msg,
"Referenced entity does not exist.")
assert cm.value.get_errors()[0].msg == "Referenced entity does not exist."
@with_setup(setup, teardown)
def test_override_with_existing_ref():
rt1 = db.RecordType("TestRecordType1").insert()
rt2 = db.RecordType("TestRecordType2").insert()
......@@ -74,10 +69,9 @@ def test_override_with_existing_ref():
datatype=rt2,
value="TestRecord").add_parent(rt1).insert()
assert_true(rec2.is_valid())
assert rec2.is_valid() is True
@with_setup(setup, teardown)
def test_reference_datatype_sequencial():
rt = db.RecordType(name="TestRT", description="TestRTDesc").insert()
p = db.Property(
......@@ -85,22 +79,21 @@ def test_reference_datatype_sequencial():
description="RefOnTestRT",
datatype="TestRT").insert()
assert_true(p.is_valid())
assert p.is_valid() is True
dt = db.execute_query("FIND TestProp", unique=True).datatype
assert_equal(dt, "TestRT")
assert dt == "TestRT"
rt2 = db.RecordType(
"TestRT2",
description="TestRT2Desc").add_property(
name="TestProp",
value="TestRT").insert()
assert_true(rt2.is_valid())
assert_equal(rt2.get_property("TestProp").value, rt.id)
assert_equal(rt2.get_property("TestProp").datatype, "TestRT")
assert rt2.is_valid() is True
assert rt2.get_property("TestProp").value == rt.id
assert rt2.get_property("TestProp").datatype == "TestRT"
@with_setup(setup, teardown)
def test_reference_datatype_at_once():
rt = db.RecordType(name="TestRT")
rt2 = db.RecordType(name="TestRT2").add_property(name="TestProp")
......@@ -112,10 +105,9 @@ def test_reference_datatype_at_once():
p = db.Property(name="TestProp", datatype="TestRT")
c = db.Container().extend([rt, rt2, rec, rec2, p]).insert()
assert_true(c.is_valid())
assert c.is_valid() is True
@with_setup(setup, teardown)
def test_generic_reference_success():
rt1 = db.RecordType(name="TestRT1").insert()
rt2 = db.RecordType(name="TestRT2").insert()
......@@ -127,14 +119,13 @@ def test_generic_reference_success():
name="TestP1",
value=rec1.id).insert()
assert_true(rt1.is_valid())
assert_true(rt2.is_valid())
assert_true(p.is_valid())
assert_true(rec1.is_valid())
assert_true(rec2.is_valid())
assert rt1.is_valid() is True
assert rt2.is_valid() is True
assert p.is_valid() is True
assert rec1.is_valid() is True
assert rec2.is_valid() is True
@with_setup(setup, teardown)
def test_generic_reference_failure():
db.RecordType(name="TestRT2").insert()
db.Property(name="TestP1", datatype=db.REFERENCE).insert()
......@@ -143,49 +134,45 @@ def test_generic_reference_failure():
name="TestRT2").add_property(
name="TestP1",
value="asdf")
assert_raises(db.TransactionError, rec2.insert)
raises(db.TransactionError, rec2.insert)
@with_setup(setup, teardown)
def test_unknown_datatype1():
p = db.Property(name="TestP", datatype="Non-Existing")
try:
p.insert()
assert_true(False)
assert False is True
except db.TransactionError as e:
assert_equal(e.msg, "Unknown datatype.")
assert e.msg == "Unknown datatype."
@with_setup(setup, teardown)
def test_unknown_datatype2():
p = db.Property(name="TestP", datatype="12345687654334567")
try:
p.insert()
assert_true(False)
assert False is True
except db.TransactionError as e:
assert_equal(e.msg, "Unknown datatype.")
assert e.msg == "Unknown datatype."
@with_setup(setup, teardown)
def test_unknown_datatype3():
p = db.Property(name="TestP", datatype="-134")
try:
p.insert()
assert_true(False)
assert False is True
except db.TransactionError as e:
assert_equal(e.msg, "Unknown datatype.")
assert e.msg == "Unknown datatype."
@with_setup(setup, teardown)
def test_wrong_refid():
rt1 = db.RecordType(name="TestRT1").insert()
rt2 = db.RecordType(name="TestRT2").insert()
rt3 = db.RecordType(name="TestRT3").insert()
p = db.Property(name="TestP1", datatype=rt1.id).insert()
assert_true(p.is_valid())
assert_true(rt1.is_valid())
assert_true(rt2.is_valid())
assert_true(rt3.is_valid())
assert p.is_valid() is True
assert rt1.is_valid() is True
assert rt2.is_valid() is True
assert rt3.is_valid() is True
rec1 = db.Record().add_parent(name="TestRT1").insert()
rec2 = db.Record().add_parent(name="TestRT2").insert()
......@@ -195,14 +182,32 @@ def test_wrong_refid():
value=rec2.id)
try:
rec3.insert()
assert_true(False)
assert False is True
except db.TransactionError:
assert_equal(
rec3.get_property("TestP1").get_errors()[0].description,
'Reference not qualified. The value of this Reference Property is to be a child of its data type.')
desc = ('Reference not qualified. The value of this Reference '
'Property is to be a child of its data type.')
err = rec3.get_property("TestP1").get_errors()[0]
assert err.description == desc
rec4 = db.Record().add_parent(
name="TestRT3").add_property(
name="TestP1",
value=rec1.id).insert()
assert_true(rec4.is_valid())
assert rec4.is_valid() is True
def test_datatype_mismatch_in_response():
p = db.Property("TestDoubleProperty", datatype=db.DOUBLE).insert()
with raises(ValueError) as exc:
rt = db.RecordType().add_property(p, "not a double")
assert exc.value.args[0] == ("could not convert string to "
"float: 'not a double'")
# add property by name,
rt = db.RecordType("TestEntity").add_property(p.name, "not a double")
with raises(db.TransactionError) as exc:
# should not raise ValueError but transaction error.
rt.insert()
assert exc.value.get_errors()[0].msg == "Cannot parse value to double."
tests/test_query.py
+ 22
− 0
View file @ 9f9a1ae4
......@@ -1002,3 +1002,25 @@ def test_query_by_name():
assert len(h.execute_query("FIND TestRT WITH name LIKE 'TestRec*'")) == 2
assert len(h.execute_query("FIND TestRT WITH name LIKE '*Rec*'")) == 2
assert len(h.execute_query("FIND ENTITY WITH name LIKE 'TestRec*'")) == 2
@mark.xfail(reason="Issue: https://gitlab.com/caosdb/caosdb-server/-/issues/96")
def test_referenced_as():
rt_person = db.RecordType("TestPerson").insert()
db.RecordType("TestParty").insert()
# three persons
g1 = db.Record().add_parent("TestPerson").insert()
g2 = db.Record().add_parent("TestPerson").insert()
g3 = db.Record().add_parent("TestPerson").insert()
guest_ids = set(g1.id, g2.id, g3.id)
party = db.Record(
"Diamond Jubilee of Elizabeth II").add_parent("TestParty")
party.add_property(rt_person, datatype=db.LIST(rt_person), name="Guests",
value=[g1, g2, g3])
party.insert()
guests = db.execute_query("FIND RECORD TestPerson WHICH IS REFERENCED BY "
"TestParty AS Guests")
assert len(guests) == 3
assert set([e.id for e in guests]) == guest_ids
tests/test_select.py
+ 251
− 70
View file @ 9f9a1ae4
......@@ -25,123 +25,304 @@
@author: tf
"""
import caosdb as h
# @UnresolvedImport
from nose.tools import assert_true, assert_equal, assert_is_not_none, assert_is_none
import caosdb as db
def setup_module():
print("SETUP")
teardown_module()
h.Property(name="TestPropertyOne", datatype=h.TEXT).insert()
h.Property(
db.Property(name="TestPropertyOne", datatype=db.TEXT).insert()
db.Property(
name="TestPropertyTwo",
description="Desc2",
datatype=h.TEXT).insert()
h.RecordType(
datatype=db.TEXT).insert()
db.RecordType(
name="TestRecordType",
description="DescRecTy").add_property(
name="TestPropertyOne").add_property(
name="TestPropertyTwo").insert()
name="TestPropertyOne", value="v1").add_property(
name="TestPropertyTwo", value="v2").insert()
rt_house = db.RecordType("TestHouse", description="TestHouseDesc").insert()
db.RecordType("TestWindow").insert()
rt_person = db.RecordType("TestPerson",
description="TestPersonDesc").insert()
db.RecordType("TestParty", description="TestPartyDesc").insert()
db.Property("TestHeight", description="TestHeightDesc", datatype=db.DOUBLE,
unit="ft").insert()
db.Property("TestDate", description="TestDateDesc",
datatype=db.DATETIME).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).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():
print("TEARDOWN")
try:
h.execute_query("FIND Test*").delete()
except BaseException:
pass
d = db.execute_query("FIND Test*")
if len(d) > 0:
d.delete()
def test_id1():
p1 = h.execute_query("FIND TestPropertyOne", unique=True)
assert_true(p1.is_valid())
assert_is_not_none(p1.name)
assert_is_not_none(p1.datatype)
assert_is_none(p1.description)
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 = h.execute_query("SELECT id FROM TestPropertyOne", unique=True)
assert_true(p1_c.is_valid())
assert_equal(p1_c.id, p1.id)
assert_is_none(p1_c.name)
assert_is_none(p1_c.datatype)
assert_is_none(p1_c.description)
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 = h.execute_query("FIND TestPropertyTwo", unique=True)
assert_true(p2.is_valid())
assert_is_not_none(p2.name)
assert_is_not_none(p2.datatype)
assert_is_not_none(p2.description)
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 = h.execute_query("SELECT id FROM TestPropertyTwo", unique=True)
assert_true(p2_c.is_valid())
assert_equal(p2_c.id, p2.id)
assert_is_none(p2_c.name)
assert_is_none(p2_c.datatype)
assert_is_none(p2_c.description)
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 = h.execute_query("SELECT description FROM TestProperty*")
assert_equal(len(p3s), 2)
p3s = db.execute_query("SELECT description FROM TestProperty*")
assert len(p3s) == 2
for e in p3s:
assert_is_not_none(e.id)
assert e.id is not None
def test_name1():
p3s = h.execute_query("SELECT description FROM TestProperty*")
assert_equal(len(p3s), 2)
p3s = db.execute_query("SELECT description FROM TestProperty*")
assert len(p3s) == 2
for e in p3s:
assert_is_not_none(e.name)
assert e.name is not None
def test_name2():
p3s = h.execute_query("SELECT name FROM TestProperty*")
assert_equal(len(p3s), 2)
p3s = db.execute_query("SELECT name FROM TestProperty*")
assert len(p3s) == 2
for e in p3s:
assert_is_not_none(e.name)
assert_is_none(e.description)
assert e.name is not None
assert e.description is None
def test_multi1():
p1 = h.execute_query(
p1 = db.execute_query(
"SELECT id, name, description FROM TestPropertyOne",
unique=True)
assert_is_not_none(p1.id)
assert_equal(p1.name, "TestPropertyOne")
assert_is_none(p1.description)
assert p1.id is not None
assert p1.name == "TestPropertyOne"
assert p1.description is None
p2 = h.execute_query(
p2 = db.execute_query(
"SELECT id, name, description FROM TestPropertyTwo",
unique=True)
assert_is_not_none(p2.id)
assert_equal(p2.name, "TestPropertyTwo")
assert_equal(p2.description, "Desc2")
assert p2.id is not None
assert p2.name == "TestPropertyTwo"
assert p2.description == "Desc2"
def test_sub1():
rt = h.execute_query("FIND TestRecordType", unique=True)
assert_is_not_none(rt.id)
assert_is_not_none(rt.name)
assert_is_not_none(rt.get_property("TestPropertyOne"))
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 = h.execute_query(
rt = db.execute_query(
"SELECT TestPropertyOne FROM TestRecordType",
unique=True)
assert_is_not_none(rt.id)
assert_is_not_none(rt.name)
assert_is_not_none(rt.get_property("TestPropertyOne"))
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 = h.execute_query(
rt = db.execute_query(
"SELECT TestPropertyTwo.description FROM TestRecordType",
unique=True)
assert_is_not_none(rt.id)
assert_is_not_none(rt.name)
assert_is_not_none(rt.get_property("TestPropertyTwo"))
assert_equal(rt.get_property("TestPropertyTwo").description, "Desc2")
assert_is_none(rt.get_property("TestPropertyTwo").datatype)
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():
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] == window.id
row = s.get_property_values("name", ("TestWindow", "TestHeight"))[0]
assert row[0] == "Buckingham Palace"
assert row[1] == 20.5
row = s.get_property_values(
"name", ("TestWindow", "TestHeight", "unit"))[0]
assert row[0] == "Buckingham Palace"
assert row[1] == "ft"
def test_subref_deep():
p = db.execute_query(
"SELECT name, Testdate, location, location.TestWindow.Testheight FROM "
"RECORD TestParty", unique=True)
row = p.get_property_values("name", "Testdate",
("location", "Testwindow", "Testheight"))
assert row == ("Diamond Jubilee of Elizabeth II", "2012-02-06", 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]
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 TestWindow.TestHeight.unit FROM "
"RECORD TestHouse", unique=True)
column = s.get_property_values(("TestWindow", "TestHeight", "unit"))
assert column == ("ft",)
s = db.execute_query("SELECT TestHeight.unit.TestWindow FROM "
"RECORD TestWindow", unique=True)
column = s.get_property_values(("TestHeight", "unit", "TestWindow"))
assert column == (None,)
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 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)
column = s.get_property_values(("TestWindow", "TestHeight", "description"))
assert column == ('TestHeightDesc',)
s = db.execute_query("SELECT TestHeight.description.TestWindow FROM "
"RECORD TestWindow", unique=True)
column = s.get_property_values(("TestHeight", "description", "TestWindow"))
assert column == (None,)
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 TestWindow.TestHeight.id FROM "
"RECORD TestHouse", unique=True)
column = s.get_property_values(("TestWindow", "TestHeight", "id"))
assert column == (height_id,)
s = db.execute_query("SELECT TestHeight.id.TestWindow FROM "
"RECORD TestWindow", unique=True)
column = s.get_property_values(("TestHeight", "id", "TestWindow"))
assert column == (None,)
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 TestWindow.TestHeight.name FROM "
"RECORD TestHouse", unique=True)
column = s.get_property_values(("TestWindow", "TestHeight", "name"))
assert column == ("TestHeight",)
s = db.execute_query("SELECT TestHeight.name.TestWindow FROM "
"RECORD TestWindow", unique=True)
column = s.get_property_values(("TestHeight", "name", "TestWindow"))
assert column == (None,)
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment