Select Git revision
test_converters.py
-
Henrik tom Wörden authoredHenrik tom Wörden authored
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_tickets.py 45.15 KiB
# -*- coding: 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
#
# 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 18.02.2015.
@author: tf
"""
import caosdb as db
from caosdb.exceptions import (CaosDBException,
TransactionError, UniqueNamesError)
import pytest
from tests import test_misc
def setup():
d = db.execute_query("FIND Test*")
if len(d) > 0:
d.delete()
def teardown():
setup()
def test_ticket_103a():
# strict flag...
rt = db.RecordType(name="TestRecordType")
try:
# insert w/o strict flag
haswarnings = False
rt.insert(strict=False)
assert rt.is_valid()
for w in rt.get_warnings():
haswarnings = True
break
assert haswarnings
# update w/o strict flag
haswarnings = False
rt.name = "TestRecordTypeUpdate"
rt.update(strict=False)
assert rt.is_valid()
for w in rt.get_warnings():
haswarnings = True
break
assert haswarnings
# update w/ strict flag
rt.name = "TestRecordTypeUpdate2"
with pytest.raises(TransactionError) as e:
rt.update(strict=True)
exc = e.value
print(exc)
assert 128 == int(exc.errors[0].code)
rt = exc.entities[0]
assert not rt.is_valid()
for w in rt.get_warnings():
haswarnings = True
break
assert haswarnings
for w in rt.get_errors():
if int(w.get_code()) == 128:
hasstricterror = True
break
assert hasstricterror
finally:
try:
rt.delete()
except BaseException:
pass
# insert w/ strict flag without raising an exception
haswarnings = False
hasstricterror = False
rt = db.RecordType(name="TestRecordType")
try:
rt.insert(strict=True, raise_exception_on_error=False)
assert not rt.is_valid()
for w in rt.get_warnings():
haswarnings = True
break
assert haswarnings
for w in rt.get_errors():
if w.get_code() == 128:
hasstricterror = True
break
assert hasstricterror
finally:
try:
rt.delete()
except BaseException:
pass
def test_ticket_103c():
# strict flag
rt1 = db.RecordType(name="TestStrict2")
rt1.insert(unique=True)
assert rt1.is_valid()
def test_ticket_103b():
# unique flag
rt1 = db.RecordType(name="TestNameDuplicatesEntity")
rt2 = db.RecordType(name="TestNameDuplicatesEntity")
rt1.insert(
unique=True, strict=False)
assert rt1.is_valid()
with pytest.raises(TransactionError) as te:
rt2.insert(unique=True)
assert te.value.has_error(UniqueNamesError)
assert not rt2.is_valid()
def test_ticket_101():
"""RecordType and Property constructor work differently?"""
rec = db.Record("some_name")
assert rec.name is not None
assert "some_name" == rec.name
assert rec.id is None
rt = db.RecordType("some_name")
assert rt.name is not None
assert "some_name" == rt.name
assert rt.id is None
p = db.Property("some_name")
assert p.name is not None
assert "some_name" == p.name
assert p.id is None
f = db.File("some_name")
assert f.name is not None
assert "some_name" == f.name
assert f.id is None
def test_ticket_100():
db.RecordType(name="Testa_b").insert()
db.RecordType(name="Testa_b").retrieve()
def test_ticket_114():
try:
db.execute_query("FIND rt_nice").delete()
except BaseException:
pass
try:
db.execute_query("FIND rtfirst").delete()
except BaseException:
pass
try:
db.execute_query("FIND first").delete()
except BaseException:
pass
p = db.Property(name="first", datatype="double").insert()
rt1 = db.RecordType(name="rtfirst").add_property(
name="first", importance="obligatory").insert()
rt2 = db.RecordType(name='rt_nice').add_property(
name="rtfirst", importance="obligatory")
try:
rt2.insert()
except TransactionError:
rt1.delete()
p.delete()
raise
else:
rt2.delete()
rt1.delete()
p.delete()
def test_ticket_120():
p = db.Property(
name="TestDoubleProperty", datatype="DOUBLE").insert()
rt1 = db.RecordType(name="TestRT1").insert()
rt2 = db.RecordType(name="TestRT2").add_parent(rt1).insert()
rt3 = db.RecordType(name="TestRT3").add_parent(rt2).insert()
r1 = db.Record().add_parent(rt1).add_property(
id=p.id, value=3.14).insert()
r2 = db.Record().add_parent(rt2).add_property(
id=rt1.id, value=r1).insert()
r3 = db.Record().add_parent(rt3).add_property(
id=rt2.id, value=r2).insert()
cp = db.Query("FIND PROPERTY TestDoubleProperty").execute(
unique=True)
assert p.id == cp.id
crt123 = db.Query("FIND RECORDTYPE TestRT1").execute(unique=False)
assert crt123.get_entity_by_id(rt1.id).is_valid()
assert crt123.get_entity_by_id(rt2.id).is_valid()
assert crt123.get_entity_by_id(rt3.id).is_valid()
cr1 = db.Query("FIND RECORD . TestDoubleProperty='3.14'").execute(
unique=True)
assert r1.id == cr1.id
cr23 = db.Query("FIND RECORD . TestRT1").execute(unique=False)
assert cr23.get_entity_by_id(r2.id).is_valid()
assert cr23.get_entity_by_id(r3.id).is_valid()
cr3 = db.Query("FIND RECORD . TestRT2").execute(unique=True)
assert r3.id == cr3.id
cr2 = db.Query("FIND RECORD . TestRT1->" + str(r1.id)).execute(
unique=True)
assert r2.id == cr2.id
cr3 = db.Query("FIND RECORD . TestRT1->" + str(r2.id)).execute(
unique=True)
assert r3.id == cr3.id
cr3 = db.Query(
"FIND RECORD WHICH HAS A PROPERTY blabla=4 OR TestRT1->TestRT2"
" WHICH HAS A PROPERTY TestRT1->" +
str(
r1.id) +
"").execute(
unique=True)
assert r3.id == cr3.id
cr3 = db.Query(
"FIND TestRT1 . TestRT1.TestRT1.TestDoubleProperty='3.14'"
).execute(unique=True)
assert r3.id == cr3.id
cr3 = db.Query(
"FIND RECORD TestRT1 . "
"TestRT1.TestRT1.TestDoubleProperty='3.14'"
).execute(unique=True)
assert r3.id == cr3.id
cr3 = db.Query(
"FIND RECORD . TestRT1.TestRT1.TestDoubleProperty='3.14'"
).execute(unique=True)
assert r3.id == cr3.id
def test_ticket_120a():
p = db.Property(name="TestDoubleProperty", datatype="DOUBLE")
rt1 = db.RecordType(name="TestRT1").add_property(p)
rt2 = db.RecordType(name="TestRT2").add_property(rt1, inheritance='None')
rt3 = db.RecordType(name="TestRT3").add_property(rt2, inheritance='None')
c = db.Container()
c.extend([p, rt1, rt2, rt3]).insert()
c.delete(raise_exception_on_error=True)
def test_ticket_117():
p = db.Property(
name="TestwaveVelocity",
datatype="Double",
unit="a.u.",
description="The measured velocity of the wavefront.")
# insert record type
rt1 = db.RecordType(name="TestRT1").insert()
assert rt1.is_valid()
id_rt1 = rt1.id
# retrieve it
r = db.RecordType(id=id_rt1).retrieve()
assert r.is_valid()
assert id_rt1 == r.id
# create property
p.insert()
assert p.is_valid()
# add property to record type
r.add_property(id=p.id, name='TestwaveVelocity', importance="obligatory")
r.name = 'TestRT1update'
r.update()
assert r.name is not None
assert 'TestRT1update' == r.name
assert r.is_valid()
assert r.get_properties() is not None
assert r.get_properties().get_by_name('TestwaveVelocity') is not None
assert r.get_properties().get_by_name('TestwaveVelocity').datatype is not None
assert str("DOUBLE").lower() == r.get_properties(
).get_by_name('TestwaveVelocity').datatype.lower()
def test_ticket_106():
try:
rt = db.RecordType(name="test RT 2")
rt.insert()
while True:
try:
rt.insert()
raise AssertionError(
"2nd insert should raise an TransactionError")
except db.exceptions.TransactionError as e:
print(e)
break
finally:
rt.delete()
def test_ticket_91():
import os
path = "ticket_91_test_file.py"
try:
assert not os.path.isfile(path)
upload_file = open(path, "w")
upload_file.write('print("hello world!")\n')
upload_file.close()
f = db.File(
name=path,
description="Python File Test",
path="pythontestfiles/python_test_file.py",
file=path)
f.insert()
assert f.is_valid()
finally:
try:
os.remove(path)
except BaseException:
pass
try:
f.delete()
except BaseException:
pass
def test_ticket_86():
import os
path = "ticket_86_test_file.py"
try:
upload_file = open(path, "w")
upload_file.write('print("hello world!")\n')
upload_file.close()
f = db.File(
name=path,
description="Python File Test",
path="pythontestfiles/python_test_file.py",
file=path)
f.insert()
assert f.is_valid()
f2 = db.File(name=path)
f2.retrieve()
assert f2.is_valid()
finally:
os.remove(path)
f.delete()
def test_ticket_83():
rt1 = db.RecordType(name="TestRT1").insert()
rt2 = db.RecordType(name="TestRT2").insert()
assert rt1.is_valid()
assert rt2.is_valid()
crt1 = db.RecordType(id=rt1.id).retrieve()
crt2 = db.RecordType(id=rt2.id).retrieve()
assert crt1.is_valid()
assert crt2.is_valid()
c = db.Container().extend([crt1, crt2]).retrieve()
assert 2 == len(c)
def test_ticket_139():
rt1 = db.RecordType(name="TestRT1").insert()
assert rt1.is_valid()
assert "TestRT1" == rt1.name
assert 1 == len(db.execute_query("FIND TestRT1"))
rt1.delete()
assert not rt1.is_valid()
assert 0 == len(db.execute_query("FIND TestRT1"))
rt1.insert()
assert rt1.is_valid()
assert "TestRT1" == rt1.name
assert 1 == len(db.execute_query("FIND TestRT1"))
rt1.delete()
assert not rt1.is_valid()
assert 0 == len(db.execute_query("FIND TestRT1"))
assert "TestRT1" == rt1.name
def test_ticket_138():
try:
rt_person = db.RecordType("Person").insert()
rt_person_x = db.RecordType("Person_x")
rt_person_x.add_property(rt_person)
rt_person_x.insert()
assert 1 == len(rt_person_x.get_properties())
rt_person_x.delete()
assert 1 == len(rt_person_x.get_properties())
rt_person_x.insert()
assert 1 == len(rt_person_x.get_properties())
rt_person_x.delete()
assert 1 == len(rt_person_x.get_properties())
rt_person_x.insert()
assert 1 == len(rt_person_x.get_properties())
finally:
if len(db.execute_query("FIND Person_x")) > 0:
rt_person_x.delete()
rt_person.delete()
def test_ticket_137():
# insert RecordType
rt1 = db.RecordType("TestRT1").insert()
assert rt1.is_valid()
crt1 = db.execute_query("FIND TestRT1", unique=True)
assert crt1.is_valid()
assert rt1.id == crt1.id
crt1.delete()
assert not crt1.is_valid()
assert 0 == len(db.execute_query("FIND TestRT1"))
try:
rt1.insert()
raise AssertionError("This should have raised an TransactionError")
except TransactionError as e:
print(e)
@pytest.mark.skip
def test_ticket_39():
import os
scratch = os.path.realpath(db.get_config().get(
"EndToEndTests", "test_tickets.test_ticket_39.scratch"))
assert os.path.isdir(scratch)
testfile = os.path.join(scratch, "test.dat")
try:
# insert RecordType
upload_file = open(testfile, "w")
upload_file.write("hello world\n")
upload_file.close()
# check if the server can connect to this filesystem
f = db.File(path="testfiles/file1", pickup=testfile).insert()
assert f.is_valid()
f.delete()
# make unreadable
os.chmod(testfile, 0o000)
with pytest.raises(TransactionError) as te:
f = db.File(
path="testfiles/file1", pickup=testfile).insert()
e = te.value.errors[0]
assert e.msg == "Insufficient read permission for this file. Please make it readable."
finally:
os.chmod(testfile, 0o600)
if len(db.execute_query("FIND RT1")) > 0:
rt1.delete()
try:
os.remove(testfile)
except Exception as e:
print(e)
def test_ticket_128():
try:
db.execute_query(r"FIND 5\#):xw;;-`;BY6~PjsI^*g.$+eY#n.aA9zm")
except TransactionError as e:
assert 13 == int(e.error.code)
def test_ticket_123a():
try:
p = db.Property(
name="TestDoubleProperty", datatype="Double").insert()
assert p.is_valid()
assert p.id == db.execute_query(
"FIND TestDoubleProperty", unique=True).id
assert p.id == db.execute_query("FIND TestDouble*", unique=True).id
assert p.id == db.execute_query(
"FIND TestD*Property", unique=True).id
assert p.id == db.execute_query(
"FIND *stDoubleProperty", unique=True).id
assert p.id == db.execute_query(
"FIND TestDoubleProperty*", unique=True).id
assert p.id == db.execute_query(
"FIND TestDouble*Property", unique=True).id
assert p.id == db.execute_query(
"FIND *Te*tDoublePr*operty", unique=True).id
assert p.id == db.execute_query(
"FIND *T*oubl*oper*", unique=True).id
finally:
p.delete()
def test_ticket_123():
try:
p = db.Property(
name="TestDoubleProperty", datatype="Double").insert()
assert p.is_valid()
assert p.id == db.execute_query(
"FIND <<TestDoubleProperty>>", unique=True).id
assert p.id == db.execute_query(
"FIND <<TestDouble>>", unique=True).id
assert p.id == db.execute_query(
"FIND <<TestD.*Property>>", unique=True).id
assert p.id == db.execute_query(
"FIND <<stDoubleProperty>>", unique=True).id
assert p.id == db.execute_query(
"FIND <<TestDoubleProperty>>", unique=True).id
assert p.id == db.execute_query(
"FIND <<TestDoubleProperty>>", unique=True).id
assert p.id == db.execute_query(
"FIND <<TestDoubleProperty>>", unique=True).id
assert p.id == db.execute_query(
"FIND <<Te[asdfgjkle]tDoubl.*oper>>", unique=True).id
finally:
p.delete()
def test_ticket_143():
try:
p = db.Property(
name="TestDoubleProperty", datatype="DOUBLE").insert()
assert p.is_valid()
rt = db.RecordType(name="TestRecordType").add_property(p).insert()
assert rt.is_valid()
r1 = db.Record().add_parent(rt).add_property(p, value="1.0").insert()
assert r1.is_valid()
r2 = db.Record().add_parent(rt).add_property(p, value="2.0").insert()
assert r2.is_valid()
s = db.execute_query("FIND TestRecordType")
assert 3 == len(s)
assert s.is_valid()
s = db.execute_query("FIND RECORD TestRecordType")
assert 2 == len(s)
assert s.is_valid()
s = db.execute_query("FIND TestRecordType WITH TestDoubleProperty")
assert 3 == len(s)
assert s.is_valid()
s = db.execute_query(
"FIND RECORD TestRecordType WITH TestDoubleProperty")
assert 2 == len(s)
assert s.is_valid()
s = db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty>0")
assert 2 == len(s)
assert s.is_valid()
s = db.execute_query(
"FIND TestRecordType WITH THE GREATEST TestDoubleProperty>0")
assert 1 == len(s)
assert s.is_valid()
assert r2.id == db.execute_query(
"FIND TestRecordType WITH THE GREATEST TestDoubleProperty>0",
unique=True).id
assert r2.id == db.execute_query(
"FIND TestRecordType WITH THE GREATEST TestDoubleProperty",
unique=True).id
assert r1.id == db.execute_query(
"FIND TestRecordType WITH THE SMALLEST TestDoubleProperty",
unique=True).id
finally:
try:
r2.delete()
except BaseException:
pass
try:
r1.delete()
except BaseException:
pass
try:
rt.delete()
except BaseException:
pass
try:
p.delete()
except BaseException:
pass
@pytest.mark.slow
def test_ticket_147():
try:
db.execute_query("FIND ticket147_*").delete()
except BaseException:
pass
try:
# structure
obs = db.Property(
name="ticket147_ObstacleRadius", datatype="DOUBLE").insert()
assert obs.is_valid()
tstep = db.Property(
name="ticket147_TimeStep", datatype="DOUBLE").insert()
assert tstep.is_valid()
bmsim = db.RecordType(name="ticket147_BarkleyModelSimulation"
).add_property(tstep).insert()
assert bmsim.is_valid()
frm = db.RecordType(name="ticket147_FrequencyMeasurement"
).add_property(obs).add_property(bmsim).insert()
assert frm.is_valid()
# data
c = db.Container()
r1 = db.Record().add_parent(bmsim).add_property(tstep, value="0.001")
r2 = db.Record().add_parent(bmsim).add_property(tstep, value="0.001")
r3 = db.Record().add_parent(bmsim).add_property(tstep, value="0.002")
r4 = db.Record().add_parent(bmsim).add_property(tstep, value="0.002")
r5 = db.Record().add_parent(bmsim).add_property(tstep, value="0.003")
c.extend([r1, r2, r3, r4, r5]).insert()
assert c.is_valid()
for e in c:
assert e.is_valid()
c2 = db.Container()
r21 = db.Record().add_parent(frm).add_property(
obs, value="1.0").add_property(
id=bmsim.id, value=r1)
r22 = db.Record().add_parent(frm).add_property(
obs, value="2.0").add_property(
id=bmsim.id, value=r2)
r23 = db.Record().add_parent(frm).add_property(
obs, value="2.0").add_property(
id=bmsim.id, value=r3)
r24 = db.Record().add_parent(frm).add_property(
obs, value="3.0").add_property(
id=bmsim.id, value=r4)
r25 = db.Record().add_parent(frm).add_property(
obs, value="3.0").add_property(
id=bmsim.id, value=r5)
c2.extend([r21, r22, r23, r24, r25]).insert()
assert c2.is_valid()
'''
s = db.execute_query("FIND ticket147*")
assert s.is_valid()
assert 14 == len(s)
s = db.execute_query("FIND ticket147_Bar*")
assert s.is_valid()
assert 6 == len(s)
s = db.execute_query("FIND RECORD ticket147_Bar*")
assert s.is_valid()
assert 5 == len(s)
s = db.execute_query("FIND ticket147_Fre*")
assert s.is_valid()
assert 6 == len(s)
s = db.execute_query("FIND RECORD ticket147_Fre*")
assert s.is_valid()
assert 5 == len(s)
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius=2.0")
assert s.is_valid()
assert 2 == len(s)
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius='2.0'")
assert s.is_valid()
assert 2 == len(s)
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_BarkleyModelSimulation")
assert s.is_valid()
assert 6 == len(s)
s = db.execute_query(
"FIND Record ticket147_Fre* WHICH HAS A PROPERTY ticket147_BarkleyModelSimulation")
assert s.is_valid()
assert 5 == len(s)'''
s = db.execute_query(
"FIND Record ticket147_Fre* WHICH HAS A PROPERTY ticket147_BarkleyModelSimulation WHICH HAS A ticket147_TimeStep"
)
assert s.is_valid()
assert 5 == len(s)
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius='2.0' AND A PROPERTY ( ticket147_BarkleyModelSimulation )"
)
assert s.is_valid()
assert 2 == len(s)
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius='2.0' AND A PROPERTY ( ticket147_BarkleyModelSimulation WHICH HAS A ticket147_TimeStep)"
)
assert s.is_valid()
assert 2 == len(s)
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius='2.0' AND A PROPERTY ( ticket147_BarkleyModelSimulation WHICH HAS A ticket147_TimeStep<='0.0021')"
)
assert s.is_valid()
assert 2 == len(s)
assert r22.id == s[0].id
assert r23.id == s[1].id
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius='2.0' AND A PROPERTY ( ticket147_BarkleyModelSimulation WHICH HAS A ticket147_TimeStep='0.002')"
)
assert s.is_valid()
assert 1 == len(s)
assert r23.id == s[0].id
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius>='2.0' AND A PROPERTY ( ticket147_BarkleyModelSimulation WHICH HAS A ticket147_TimeStep>='0.002')"
)
assert s.is_valid()
assert 3 == len(s)
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius='2.0' AND A PROPERTY ( ticket147_BarkleyModelSimulation WHICH HAS A ticket147_TimeStep='0.001')"
)
assert s.is_valid()
assert 1 == len(s)
assert r22.id == s[0].id
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius='2.0' AND A PROPERTY ( ticket147_BarkleyModelSimulation WHICH HAS ticket147_TimeStep='0.001')"
)
assert s.is_valid()
assert 1 == len(s)
assert r22.id == s[0].id
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius='2.0' AND A PROPERTY ( ticket147_BarkleyModelSimulation WHICH HAS THE SMALLEST ticket147_TimeStep)"
)
assert s.is_valid()
assert 1 == len(s)
assert r22.id == s[0].id
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS THE SMALLEST ticket147_ObstacleRadius AND A PROPERTY ( ticket147_BarkleyModelSimulation WHICH HAS THE SMALLEST ticket147_TimeStep)"
)
assert s.is_valid()
assert 1 == len(s)
assert r21.id == s[0].id
# typo: SMALLES
with pytest.raises(TransactionError):
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS THE SMALLEST ticket147_ObstacleRadius AND A PROPERTY ( ticket147_BarkleyModelSimulation WHICH HAS THE SMALLES ticket147_TimeStep)"
)
finally:
try:
c2.delete()
except BaseException:
pass
try:
c.delete()
except BaseException:
pass
try:
frm.delete()
except BaseException:
pass
try:
bmsim.delete()
except BaseException:
pass
try:
tstep.delete()
except BaseException:
pass
try:
obs.delete()
except BaseException:
pass
def test_ticket_89():
try:
nt_user = db.Property(name="NT_user", datatype="TEXT").insert()
assert nt_user.is_valid()
nt_data = db.RecordType(name="NT_testfile", id=-100)
# nt_data = <File id="-100" name="NT_testfile"
# path="testfiles/NT_test.txt"/>
nt_data.insert()
assert nt_data.is_valid()
assert 0 == len(nt_data.get_properties())
# nt_data =
# <File id="318" cuid="c0c28bad-fb30-4e74-94ad-10934363e755" name="NT_testfile" path="testfiles/NT_test.txt" checksum="cf83e1357eefb8bdf1542850d66d8007d620e4050b5715dc83f4a921d36ce9ce47d0d13c5d85f2b0ff8318d2877eec2f63b931bd47417a81a538327af927da3e" size="0">
# <Warning code="106" description="Entity has no description."/>
# </File>
nt_data.add_property(nt_user, value="user1")
nt_data.update()
assert nt_data.is_valid()
assert 1 == len(nt_data.get_properties())
# nt_data =
# <File id="318" cuid="c0c28bad-fb30-4e74-94ad-10934363e755" name="NT_testfile" path="testfiles/NT_test.txt" checksum="cf83e1357eefb8bdf1542850d66d8007d620e4050b5715dc83f4a921d36ce9ce47d0d13c5d85f2b0ff8318d2877eec2f63b931bd47417a81a538327af927da3e" size="0">
# <Warning code="106" description="Entity has no description."/>
# <Property id="316" name="NT_user" description="Test Property of type 'string' for nosetests" datatype="TEXT" importance="FIX">user1</Property>
# </File>
nt_data.add_property(nt_user, value="user2")
nt_data.update()
assert nt_data.is_valid()
assert 2 == len(nt_data.get_properties())
# nt_data =
# <File id="318" cuid="c0c28bad-fb30-4e74-94ad-10934363e755" name="NT_testfile" path="testfiles/NT_test.txt" checksum="cf83e1357eefb8bdf1542850d66d8007d620e4050b5715dc83f4a921d36ce9ce47d0d13c5d85f2b0ff8318d2877eec2f63b931bd47417a81a538327af927da3e" size="0">
# <Warning code="106" description="Entity has no description."/>
# <Property id="316" name="NT_user" description="Test Property of type 'string' for nosetests" datatype="TEXT" importance="FIX">user2</Property>
# <Property id="316" name="NT_user" description="Test Property of type 'string' for nosetests" datatype="TEXT" importance="FIX">user2</Property>
# <Property id="316" name="NT_user" description="Test Property of type 'string' for nosetests" datatype="TEXT" importance="FIX">user1</Property>
# </File>
nt_data.retrieve()
assert nt_data.is_valid()
assert 2 == len(nt_data.get_properties())
# response:
# <Entities>
# <File id="318" name="NT_testfile" path="testfiles/NT_test.txt" checksum="CF83E1357EEFB8BDF1542850D66D8007D620E4050B5715DC83F4A921D36CE9CE47D0D13C5D85F2B0FF8318D2877EEC2F63B931BD47417A81A538327AF927DA3E" size="0">
# <Property id="316" name="NT_user" description="Test Property of type 'string' for nosetests" datatype="TEXT" importance="FIX">user1</Property>
# <Property id="316" name="NT_user" description="Test Property of type 'string' for nosetests" datatype="TEXT" importance="FIX">user2</Property>
# </File>
# </Entities>
# nt_data=
# <File id="318" cuid="c0c28bad-fb30-4e74-94ad-10934363e755" name="NT_testfile" path="testfiles/NT_test.txt" checksum="CF83E1357EEFB8BDF1542850D66D8007D620E4050B5715DC83F4A921D36CE9CE47D0D13C5D85F2B0FF8318D2877EEC2F63B931BD47417A81A538327AF927DA3E" size="0">
# <Property id="316" name="NT_user" description="Test Property of type 'string' for nosetests" datatype="TEXT" importance="FIX">user2</Property>
# <Property id="316" name="NT_user" description="Test Property of type 'string' for nosetests" datatype="TEXT" importance="FIX">user2</Property>
# <Property id="316" name="NT_user" description="Test Property of type 'string' for nosetests" datatype="TEXT" importance="FIX">user2</Property>
# <Property id="316" name="NT_user" description="Test Property of type 'string' for nosetests" datatype="TEXT" importance="FIX">user1</Property>
# </File>
finally:
try:
db.execute_query("FIND RECORDTYPE NT_testfile").delete(
raise_exception_on_error=False)
except BaseException:
pass
try:
db.execute_query(
"FIND PROPERTY NT_user",
unique=True).delete(raise_exception_on_error=False)
except BaseException:
pass
def test_ticket_173():
try:
assert 0 == db.execute_query("COUNT TestTextProperty")
p = db.Property(
name="TestTextProperty",
description="simple text property (from test_tickets.py:test_ticket173)",
datatype='text')
p.insert()
assert p.is_valid()
assert 1 == db.execute_query("COUNT TestTextProperty")
finally:
try:
p.delete()
except BaseException:
pass
def test_ticket_166():
# structure
obs = db.Property(
name="Testticket166_ObstacleRadius", datatype="DOUBLE").insert()
assert obs.is_valid()
tstep = db.Property(
name="Testticket166_TimeStep", datatype="DOUBLE").insert()
assert tstep.is_valid()
bmsim = db.RecordType(name="Testticket166_BarkleyModelSimulation"
).add_property(tstep).insert()
assert bmsim.is_valid()
frm = db.RecordType(name="Testticket166_FrequencyMeasurement"
).add_property(obs).add_property(bmsim).insert()
assert frm.is_valid()
# data
c = db.Container()
r1 = db.Record().add_parent(bmsim).add_property(tstep, value="0.001")
r2 = db.Record().add_parent(bmsim).add_property(tstep, value="0.001")
r3 = db.Record().add_parent(bmsim).add_property(tstep, value="0.002")
r4 = db.Record().add_parent(bmsim).add_property(tstep, value="0.002")
r5 = db.Record().add_parent(bmsim).add_property(tstep, value="0.003")
c.extend([r1, r2, r3, r4, r5]).insert()
assert c.is_valid()
for e in c:
assert e.is_valid()
c2 = db.Container()
r21 = db.Record().add_parent(frm).add_property(
obs, value="1.0").add_property(
id=bmsim.id, value=r1)
r22 = db.Record().add_parent(frm).add_property(
obs, value="2.0").add_property(
id=bmsim.id, value=r2)
r23 = db.Record().add_parent(frm).add_property(
obs, value="2.0").add_property(
id=bmsim.id, value=r3)
r24 = db.Record().add_parent(frm).add_property(
obs, value="3.0").add_property(
id=bmsim.id, value=r4)
r25 = db.Record().add_parent(frm).add_property(
obs, value="3.0").add_property(
id=bmsim.id, value=r5)
c2.extend([r21, r22, r23, r24, r25]).insert()
assert c2.is_valid()
# retrieve all 10
q = db.Query("FIND RECORD Testticket166*")
assert 10 == len(q.execute())
# retrieve first 5
q.putFlag("P", "0L5")
ret = q.execute()
assert 5 == len(ret)
for e in ret:
assert bmsim.id == e.get_parents()[0].id
# retrieve from 5 to 9 (4 entities)
q.putFlag("P", "5L4")
ret = q.execute()
assert 4 == len(ret)
for e in ret:
assert frm.id == e.get_parents()[0].id
def test_ticket_178():
from lxml import etree
# lxml umlaut
xml = etree.Element("Björn")
assert xml.tag == "Björn"
try:
rt = db.RecordType(name="Person")
rt.insert()
assert rt.is_valid()
p = db.Property(name="First name", datatype="TEXT")
p.insert()
assert p.is_valid()
r = db.Record().add_parent(parent=rt).add_property(
name="First name", value="Björn")
r.insert()
assert r.is_valid()
assert db.Record(id=r.id).retrieve().get_property(
"First name").value == "Björn"
finally:
try:
r.delete()
except BaseException:
pass
try:
p.delete()
except BaseException:
pass
try:
rt.delete()
except BaseException:
pass
def test_ticket_124():
try:
rt = db.RecordType(name="Person")
rt.insert()
assert rt.is_valid()
p = db.Property(name="First name", datatype="TEXT")
p.insert()
assert p.is_valid()
r = db.Record().add_parent(parent=rt).add_property(
name="First name", value="Björn")
r.insert()
assert r.is_valid()
r2 = db.execute_query("Find Person WITH 'First name'=Björn")[0]
assert r2.is_valid()
assert r2.id == r.id
finally:
try:
r.delete()
except BaseException:
pass
try:
p.delete()
except BaseException:
pass
try:
rt.delete()
except BaseException:
pass
def test_ticket_163():
try:
p = db.Property(name="SomePropertyTest", datatype="TEXT")
p.insert()
assert p.is_valid()
rt = db.RecordType(name="RecordTypeTest").add_property(
name="SomePropertyTest", importance="obligatory", value="blabla")
rt.insert()
assert rt.is_valid()
rt2 = db.execute_query(
"FIND Entity . SomePropertyTest=blabla", unique=True)
assert rt2.is_valid()
assert rt2.id == rt.id
finally:
try:
rt.delete()
except BaseException:
pass
try:
p.delete()
except BaseException:
pass
def test_ticket_165():
try:
db.execute_query("FIND Test*").delete()
except BaseException:
pass
try:
p = db.Property(name="TestComment", datatype="TEXT")
rt_a = db.RecordType(name="TestAnnotation").add_property(
name="TestComment")
rt_a.datatype = "REFERENCE"
rt_b = db.RecordType(name="TestEntity")
assert isinstance(rt_a, db.Entity)
r_a = db.Record().add_parent(rt_a).add_property(
name="TestComment", value="blablabla")
r_a.value = rt_b
c = db.Container().extend([p, rt_a, rt_b, r_a]).insert()
assert c.is_valid()
for e in c:
assert e.is_valid()
assert rt_b.id == db.execute_query(
"FIND TestEntity WHICH IS REFERENCED BY TestAnnotation", unique=True).id
assert rt_b.id == db.execute_query(
"FIND Entity WHICH IS REFERENCED BY TestAnnotation", unique=True).id
assert rt_b.id == db.execute_query(
"FIND RecordType WHICH IS REFERENCED BY TestAnnotation", unique=True).id
assert 0 == len(db.execute_query(
"FIND Record WHICH IS REFERENCED BY TestAnnotation"))
assert rt_b.id == db.execute_query(
"FIND TestEntity WHICH IS REFERENCED WITH TestComment", unique=True).id
assert rt_b.id == db.execute_query(
"FIND Entity WHICH IS REFERENCED BY TestAnnotation WITH TestComment",
unique=True).id
assert rt_b.id == db.execute_query(
"FIND RecordType WHICH IS REFERENCED BY TestAnnotation WITH TestComment",
unique=True).id
assert 0 == len(db.execute_query(
"FIND Record WHICH IS REFERENCED BY TestAnnotation WITH TestComment"))
assert rt_b.id == db.execute_query(
"FIND TestEntity WHICH IS REFERENCED BY TestAnnotation WITH TestComment",
unique=True).id
assert rt_b.id == db.execute_query(
"FIND Entity WHICH IS REFERENCED BY TestAnnotation WITH TestComment",
unique=True).id
assert rt_b.id == db.execute_query(
"FIND RecordType WHICH IS REFERENCED BY TestAnnotation WITH TestComment",
unique=True).id
assert 0 == len(db.execute_query(
"FIND Record WHICH IS REFERENCED BY TestAnnotation WITH TestComment"))
assert rt_b.id == db.execute_query(
"FIND TestEntity WHICH IS REFERENCED BY TestAnnotation WITH TestComment='blablabla'",
unique=True).id
assert rt_b.id == db.execute_query(
"FIND Entity WHICH IS REFERENCED BY TestAnnotation WITH TestComment='blablabla'",
unique=True).id
assert rt_b.id == db.execute_query(
"FIND RecordType WHICH IS REFERENCED BY TestAnnotation WITH TestComment='blablabla'",
unique=True).id
assert 0 == len(db.execute_query(
"FIND Record WHICH IS REFERENCED BY TestAnnotation WITH TestComment='blablabla'"))
c.delete()
finally:
try:
db.execute_query("FIND Test*").delete()
except BaseException:
pass
try:
r_a.delete()
except BaseException:
pass
try:
rt_b.delete()
except BaseException:
pass
try:
rt_b.delete()
except BaseException:
pass
try:
rt_a.delete()
except BaseException:
pass
try:
p.delete()
except BaseException:
pass
def test_ticket_174():
try:
p1 = db.Property(name="Ticket174_P1", datatype="INTEGER")
p2 = db.Property(name="Ticket174_P2", datatype="DOUBLE")
rt1 = db.RecordType(name="Ticket174_RecordType1").add_property(
p1, value=1).add_property(
p2, value=2)
rt2 = db.RecordType(name="Ticket174_RecordType2").add_property(
p2, value=1).add_property(
p1, value=2)
rt3 = db.RecordType(name="Ticket174_RecordType3").add_property(
p1, value=1).add_property(
p1, value=2).add_property(
p2, value=3)
rt4 = db.RecordType(name="Ticket174_RecordType4").add_property(
p2, value=1).add_property(
p1, value=2).add_property(
p1, value=3).add_property(
p2, value=4)
c = db.Container().extend([p1, p2, rt1, rt2, rt3, rt4]).insert()
assert c.is_valid()
assert rt1.is_valid()
for i in range(len(rt1.get_properties())):
assert i + 1.0 == float(rt1.get_properties()[i].value)
assert rt2.is_valid()
for i in range(len(rt2.get_properties())):
assert i + 1.0 == float(rt2.get_properties()[i].value)
assert rt3.is_valid()
for i in range(len(rt3.get_properties())):
assert i + 1.0 == float(rt3.get_properties()[i].value)
assert rt4.is_valid()
for i in range(len(rt4.get_properties())):
assert i + 1.0 == float(rt4.get_properties()[i].value)
# change order:
rt4_i1 = rt4.get_properties()[1]
rt4_i2 = rt4.get_properties()[2]
rt4.get_properties()[2] = rt4_i1
rt4.get_properties()[1] = rt4_i2
rt4.update()
assert rt4.is_valid()
assert 1.0 == float(rt4.get_properties()[0].value)
assert 3.0 == float(rt4.get_properties()[1].value)
assert 2.0 == float(rt4.get_properties()[2].value)
assert 4.0 == float(rt4.get_properties()[3].value)
# retrieve again:
rt42 = db.execute_query("FIND Ticket174_RecordType4", unique=True)
assert 1.0 == float(rt42.get_properties()[0].value)
assert 3.0 == float(rt42.get_properties()[1].value)
assert 2.0 == float(rt42.get_properties()[2].value)
assert 4.0 == float(rt42.get_properties()[3].value)
c.delete()
finally:
try:
rt1.delete()
except BaseException:
pass
try:
rt2.delete()
except BaseException:
pass
try:
rt3.delete()
except BaseException:
pass
try:
rt4.delete()
except BaseException:
pass
try:
p2.delete()
except BaseException:
pass
try:
p1.delete()
except BaseException:
pass
def test_ticket_192():
import time
if time.localtime().tm_hour > time.gmtime().tm_hour:
try:
rt = db.RecordType(name="SimulationTestRecordType").insert()
rec = db.Record().add_parent(rt).insert()
assert rec.id == db.execute_query(
"FIND Record SimulationTestRecordType WHICH WAS CREATED BY ME", unique=True).id
assert rec.id == db.execute_query(
"FIND Record SimulationTestRecordType WHICH WAS CREATED TODAY", unique=True).id
assert rec.id == db.execute_query(
"FIND Record SimulationTestRecordType WHICH WAS CREATED TODAY BY ME",
unique=True).id
assert rec.id == db.execute_query(
"FIND Record WHICH WAS CREATED TODAY BY ME", unique=True).id
finally:
try:
rec.delete()
except BaseException:
pass
try:
rt.delete()
except BaseException:
pass
def test_ticket_200():
# Properties:
Email = db.Property(name="TestEmailAddress", datatype="TEXT").insert()
Fname = db.Property(name="TestFirstName", datatype="TEXT").insert()
Lname = db.Property(name="TestLastName", datatype="TEXT").insert()
# Record Types:
Person = db.RecordType(name="TestPersonRecordType").add_property(
Fname).add_property(Lname).insert()
# These properties are to be inherited:
# FirstName = Property(name="FirstName", datatype="TEXT")
# LastName = Property(name="LastName", datatype="TEXT")
Person2 = db.RecordType(
name="TestPersonWithEmailRecordType") .add_parent(
parent=db.RecordType("TestPersonRecordType").retrieve(),
inheritance="all") .add_property(
property=Email).insert()
assert Person2.is_valid()
assert Email.id == Person2.get_property("TestEmailAddress").id
assert Fname.id == Person2.get_property("TestFirstName").id
assert Lname.id == Person2.get_property("TestLastName").id
p2 = db.execute_query(
"FIND RecordType TestPersonWithEmailRecordType", unique=True)
assert p2.id == Person2.id
assert Email.id == p2.get_property("TestEmailAddress").id
assert Fname.id == p2.get_property("TestFirstName").id
assert Lname.id == p2.get_property("TestLastName").id
def test_ticket_155():
# Properties:
p = db.Property(
name="TestTextProperty",
description="standard desc",
datatype="TEXT").insert()
assert p.is_valid()
assert p.description == "standard desc"
p2 = db.execute_query("FIND TestTextProperty", True)
assert p2.is_valid()
assert p2.description == "standard desc"
rt = db.RecordType(name="TestRecordType").add_property(
p, description="overridden standard desc").insert()
assert rt.is_valid()
assert rt.get_property(
"TestTextProperty").description == "overridden standard desc"
rt2 = db.execute_query("FIND TestRecordType", True)
assert rt2.is_valid()
assert rt2.get_property(
"TestTextProperty").description == "overridden standard desc"
rt3 = db.RecordType(name="TestRecordType2").add_parent(
rt, inheritance="all").insert()
assert rt3.is_valid()
assert rt3.get_property(
"TestTextProperty").description == "overridden standard desc"
rt4 = db.execute_query("FIND TestRecordType2", True)
assert rt4.is_valid()
assert rt4.get_property(
"TestTextProperty").description == "overridden standard desc"
def test_ticket_191():
rt = db.RecordType(name="TestRecordType")
rt.insert()
c = db.Container().retrieve(
[rt.id, rt.id], raise_exception_on_error=False, sync=False)
assert len(c) == 1
def run_test_files_tmp():
test_misc.test_info()
def test_ticket_104():
try:
url = db.get_config().get("Connection", "url")
try:
from urllib.parse import urlparse
except BaseException:
# python2
from urlparse import urlparse
_fullurl = urlparse(url)
_netloc = _fullurl.netloc
_basepath = _fullurl.path
wrongbasepath = _netloc + "/wrongcontextroot/"
with pytest.raises(CaosDBException):
db.configure_connection(url=wrongbasepath)._login()
finally:
db.configure_connection()
def test_ticket_115():
p1 = db.Property(name="TestDoubleProperty1", datatype=db.DOUBLE)
p2 = db.Property(name="TestDoubleProperty2", datatype=db.DOUBLE)
rt = db.RecordType(
name="TestRecordType").add_property(p1).add_property(p2)
c = db.Container().extend([p1, p2, rt])
c.insert()
assert c.is_valid()
rtc = db.execute_query("FIND TestRecordType", unique=True)
assert len(rtc.get_properties()) == 2
rtc.get_properties().remove("TestDoubleProperty1")
assert len(rtc.get_properties()) == 1
rtc.update()
assert c.is_valid()
assert len(rtc.get_properties()) == 1