Select Git revision
test_tickets.py
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_tickets.py 53.89 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
"""
from __future__ import absolute_import, print_function, unicode_literals
import caosdb as db
from tests import test_misc
from caosdb.exceptions import CaosDBException, TransactionError, EntityError, UniqueNamesError, AmbiguityException, EntityDoesNotExistError
from nose.tools import (nottest, assert_true, assert_raises, assert_not_equals,
assert_equals, assert_false, assert_is_not_none,
assert_is_none)
from nose import with_setup
def setup_module():
try:
db.execute_query("FIND ENTITY WITH ID > 100").delete()
except Exception as e:
print(e)
def setup():
pass
def teardown():
setup_module()
def test_ticket_103a():
# strict flag...
rt = db.RecordType(name="TestRecordType")
try:
# insert w/o strict flag
haswarnings = False
rt.insert(strict=False)
assert_true(rt.is_valid())
for w in rt.get_warnings():
haswarnings = True
break
assert_true(haswarnings)
# update w/o strict flag
haswarnings = False
rt.name = "TestRecordTypeUpdate"
rt.update(strict=False)
assert_true(rt.is_valid())
for w in rt.get_warnings():
haswarnings = True
break
assert_true(haswarnings)
# update w/ strict flag
rt.name = "TestRecordTypeUpdate2"
try:
rt.update(strict=True)
assert_true(False, "This should have raised an 128-Error (strict)")
except TransactionError as exc:
print(exc)
assert_equals(128, int(exc.get_code()))
rt = exc.get_entities()[0]
assert_false(rt.is_valid())
for w in rt.get_warnings():
haswarnings = True
break
assert_true(haswarnings)
for w in rt.get_errors():
if w.get_code() == 128:
hasstricterror = True
break
assert_true(hasstricterror)
finally:
try:
rt.delete()
except BaseException:
pass
# insert w/ strict flag
haswarnings = False
hasstricterror = False
rt = db.RecordType(name="TestRecordType")
try:
rt.insert(strict=True, raise_exception_on_error=False)
assert_false(rt.is_valid())
for w in rt.get_warnings():
haswarnings = True
break
assert_true(haswarnings)
for w in rt.get_errors():
if w.get_code() == 128:
hasstricterror = True
break
assert_true(hasstricterror)
finally:
try:
rt.delete()
except BaseException:
pass
def test_ticket_103c():
# strict flag
try:
rt1 = db.RecordType(name="TestStrict2")
rt1.insert(unique=True)
assert_true(rt1.is_valid())
finally:
try:
rt1.delete()
except BaseException:
pass
def test_ticket_103b():
# unique flag
try:
rt1 = db.RecordType(name="NameDuplicatesEntity")
rt2 = db.RecordType(name="NameDuplicatesEntity")
rt1.insert(
unique=True, strict=False)
assert_true(rt1.is_valid())
try:
rt2.insert(unique=True)
assert_true(False)
except UniqueNamesError:
pass
assert_false(rt2.is_valid())
finally:
try:
rt1.delete()
except BaseException:
pass
try:
rt2.delete()
except BaseException:
pass
def test_ticket_102():
"""EntityDoesNotExistException."""
try:
p = db.Property(name="Non-ExistentProperty")
p.retrieve(
unique=True, raise_exception_on_error=True)
assert_true(False)
except CaosDBException as e:
assert_true(isinstance(e, EntityDoesNotExistError))
assert_is_not_none(e.get_entity())
assert_equals(e.get_entity().name, p.name)
assert_true(e.get_entity().has_errors())
assert_false(p.has_errors())
assert_false(p.is_valid())
assert_false(e.get_entity().is_valid())
def test_ticket_101():
"""RecordType and Property constructor work differently?"""
rec = db.Record("some_name")
assert_is_not_none(rec.name)
assert_equals("some_name", rec.name)
assert_is_none(rec.id)
rt = db.RecordType("some_name")
assert_is_not_none(rt.name)
assert_equals("some_name", rt.name)
assert_is_none(rt.id)
p = db.Property("some_name")
assert_is_not_none(p.name)
assert_equals("some_name", p.name)
assert_is_none(p.id)
f = db.File("some_name")
assert_is_not_none(f.name)
assert_equals("some_name", f.name)
assert_is_none(f.id)
def test_ticket_100():
try:
rt = db.RecordType(name="a_b").insert()
db.RecordType(name="a_b").retrieve()
finally:
try:
rt.delete()
except BaseException:
pass
def test_ticket_96():
try:
rt1 = db.RecordType(name="a_b")
rt2 = db.RecordType(name="a_b")
rt1.insert()
rt2.insert(unique=False)
c = db.Container().append(
db.RecordType(name="a_b")).retrieve(unique=False)
assert_equals(len(c), 2)
try:
c = db.Container().append(
db.RecordType(name="a_b")).retrieve(unique=True)
assert_true(False)
except AmbiguityException as e:
print(repr(e))
finally:
try:
rt2.delete()
except BaseException:
pass
try:
rt1.delete()
except BaseException:
pass
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():
try:
p = db.Property(
name="SimpleDoubleProperty", datatype="DOUBLE").insert()
rt1 = db.RecordType(name="SimpleRT1").insert()
rt2 = db.RecordType(name="SimpleRT2").add_parent(rt1).insert()
rt3 = db.RecordType(name="SimpleRT3").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 SimpleDoubleProperty").execute(
unique=True)
assert_equals(p.id, cp.id)
crt123 = db.Query("FIND RECORDTYPE SimpleRT1").execute(unique=False)
assert_true(crt123.get_entity_by_id(rt1.id).is_valid())
assert_true(crt123.get_entity_by_id(rt2.id).is_valid())
assert_true(crt123.get_entity_by_id(rt3.id).is_valid())
cr1 = db.Query("FIND RECORD . SimpleDoubleProperty='3.14'").execute(
unique=True)
assert_equals(r1.id, cr1.id)
cr23 = db.Query("FIND RECORD . SimpleRT1").execute(unique=False)
assert_true(cr23.get_entity_by_id(r2.id).is_valid())
assert_true(cr23.get_entity_by_id(r3.id).is_valid())
cr3 = db.Query("FIND RECORD . SimpleRT2").execute(unique=True)
assert_equals(r3.id, cr3.id)
cr2 = db.Query("FIND RECORD . SimpleRT1->" + str(r1.id)).execute(
unique=True)
assert_equals(r2.id, cr2.id)
cr3 = db.Query("FIND RECORD . SimpleRT1->" + str(r2.id)).execute(
unique=True)
assert_equals(r3.id, cr3.id)
cr3 = db.Query(
"FIND RECORD WHICH HAS A PROPERTY blabla=4 OR SimpleRT1->SimpleRT2 WHICH HAS A PROPERTY SimpleRT1->" +
str(
r1.id) +
"").execute(
unique=True)
assert_equals(r3.id, cr3.id)
cr3 = db.Query(
"FIND SimpleRT1 . SimpleRT1.SimpleRT1.SimpleDoubleProperty='3.14'"
).execute(unique=True)
assert_equals(r3.id, cr3.id)
cr3 = db.Query(
"FIND RECORD SimpleRT1 . SimpleRT1.SimpleRT1.SimpleDoubleProperty='3.14'"
).execute(unique=True)
assert_equals(r3.id, cr3.id)
cr3 = db.Query(
"FIND RECORD . SimpleRT1.SimpleRT1.SimpleDoubleProperty='3.14'"
).execute(unique=True)
assert_equals(r3.id, cr3.id)
finally:
r3.delete()
r2.delete()
r1.delete()
rt3.delete()
rt2.delete()
rt1.delete()
p.delete()
@with_setup(setup, teardown)
def test_ticket_120a():
p = db.Property(name="TestSimpleDoubleProperty", 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)
@with_setup(setup, teardown)
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_true(rt1.is_valid())
id_rt1 = rt1.id
# retrieve it
r = db.RecordType(id=id_rt1).retrieve()
assert_true(r.is_valid())
assert_equals(id_rt1, r.id)
# create property
p.insert()
assert_true(p.is_valid())
# add property to record type
r.add_property(id=p.id, name='TestwaveVelocity', importance="obligatory")
r.name = 'TestRT1update'
r.update()
assert_is_not_none(r.name)
assert_equals('TestRT1update', r.name)
assert_true(r.is_valid())
assert_is_not_none(r.get_properties())
assert_is_not_none(r.get_properties().get_by_name('TestwaveVelocity'))
assert_is_not_none(
r.get_properties().get_by_name('TestwaveVelocity').datatype)
assert_equals(
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_false(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_true(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_true(f.is_valid())
f2 = db.File(name=path)
f2.retrieve()
assert_true(f2.is_valid())
finally:
os.remove(path)
f.delete()
@with_setup(setup, teardown)
def test_ticket_83():
rt1 = db.RecordType(name="TestRT1").insert()
rt2 = db.RecordType(name="TestRT2").insert()
assert_true(rt1.is_valid())
assert_true(rt2.is_valid())
crt1 = db.RecordType(id=rt1.id).retrieve()
crt2 = db.RecordType(id=rt2.id).retrieve()
assert_true(crt1.is_valid())
assert_true(crt2.is_valid())
c = db.Container().extend([crt1, crt2]).retrieve()
assert_equals(2, len(c))
def test_ticket_139():
rt1 = db.RecordType(name="TestRT1").insert()
assert_true(rt1.is_valid())
assert_equals("TestRT1", rt1.name)
assert_equals(1, len(db.execute_query("FIND TestRT1")))
rt1.delete()
assert_false(rt1.is_valid())
assert_equals(0, len(db.execute_query("FIND TestRT1")))
rt1.insert()
assert_true(rt1.is_valid())
assert_equals("TestRT1", rt1.name)
assert_equals(1, len(db.execute_query("FIND TestRT1")))
rt1.delete()
assert_false(rt1.is_valid())
assert_equals(0, len(db.execute_query("FIND TestRT1")))
assert_equals("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_equals(1, len(rt_person_x.get_properties()))
rt_person_x.delete()
assert_equals(1, len(rt_person_x.get_properties()))
rt_person_x.insert()
assert_equals(1, len(rt_person_x.get_properties()))
rt_person_x.delete()
assert_equals(1, len(rt_person_x.get_properties()))
rt_person_x.insert()
assert_equals(1, len(rt_person_x.get_properties()))
finally:
if len(db.execute_query("FIND Person_x")) > 0:
rt_person_x.delete()
rt_person.delete()
@with_setup(setup, teardown)
def test_ticket_137():
# insert RecordType
rt1 = db.RecordType("TestRT1").insert()
assert_true(rt1.is_valid())
crt1 = db.execute_query("FIND TestRT1", unique=True)
assert_true(crt1.is_valid())
assert_equals(rt1.id, crt1.id)
crt1.delete()
assert_false(crt1.is_valid())
assert_equals(0, len(db.execute_query("FIND TestRT1")))
try:
rt1.insert()
raise AssertionError("This should have raised an TransactionError")
except TransactionError as e:
print(e)
@with_setup(setup, teardown)
def test_ticket_132():
# insert RecordType
rt1 = db.RecordType("TestRT1").insert()
assert_true(rt1.is_valid())
# get as Property by id
crt1 = db.execute_query("FIND TestRT1", unique=True)
p1 = db.Property(id=crt1.id).retrieve()
assert_true(p1.is_valid())
assert_true(isinstance(p1, db.Property))
# get as Property by name
p2 = db.Property("TestRT1").retrieve()
assert_true(p2.is_valid())
assert_true(isinstance(p2, db.Property))
@with_setup(setup, teardown)
@nottest
def test_ticket_39():
import os
scratch = os.path.realpath(db.get_config().get(
"EndToEndTests", "test_tickets.test_ticket_39.scratch"))
assert_true(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", from_location=testfile).insert()
assert_true(f.is_valid())
f.delete()
# make unreadable
os.chmod(testfile, 0o000)
try:
f = db.File(
path="testfiles/file1", from_location=testfile).insert()
except EntityError as e:
assert_equals(
"Insufficient read permission for this file. Please make it readable.",
e.msg)
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_equals(13, int(e.get_error().code))
def test_ticket_123a():
try:
p = db.Property(
name="SimpleDoubleProperty", datatype="Double").insert()
assert_true(p.is_valid())
assert_equals(p.id,
db.execute_query(
"FIND SimpleDoubleProperty", unique=True).id)
assert_equals(p.id,
db.execute_query("FIND SimpleDouble*", unique=True).id)
assert_equals(p.id,
db.execute_query("FIND SimpleD*Property",
unique=True).id)
assert_equals(p.id,
db.execute_query("FIND *leDoubleProperty",
unique=True).id)
assert_equals(p.id,
db.execute_query(
"FIND SimpleDoubleProperty*", unique=True).id)
assert_equals(p.id,
db.execute_query(
"FIND SimpleDouble*Property", unique=True).id)
assert_equals(p.id,
db.execute_query(
"FIND *Simpl*eDoublePr*operty", unique=True).id)
assert_equals(p.id,
db.execute_query("FIND *Simp*oubl*oper*",
unique=True).id)
finally:
p.delete()
def test_ticket_123():
try:
p = db.Property(
name="SimpleDoubleProperty", datatype="Double").insert()
assert_true(p.is_valid())
assert_equals(p.id,
db.execute_query(
"FIND <<SimpleDoubleProperty>>", unique=True).id)
assert_equals(p.id,
db.execute_query("FIND <<SimpleDouble>>",
unique=True).id)
assert_equals(p.id,
db.execute_query(
"FIND <<SimpleD.*Property>>", unique=True).id)
assert_equals(p.id,
db.execute_query(
"FIND <<leDoubleProperty>>", unique=True).id)
assert_equals(p.id,
db.execute_query(
"FIND <<SimpleDoubleProperty>>", unique=True).id)
assert_equals(p.id,
db.execute_query(
"FIND <<SimpleDoubleProperty>>", unique=True).id)
assert_equals(p.id,
db.execute_query(
"FIND <<SimpleDoubleProperty>>", unique=True).id)
assert_equals(p.id,
db.execute_query(
"FIND <<Simp[asdfgjkle]eDoubl.*oper>>",
unique=True).id)
finally:
p.delete()
def test_ticket_143():
try:
p = db.Property(
name="SimpleDoubleProperty", datatype="DOUBLE").insert()
assert_true(p.is_valid())
rt = db.RecordType(name="SimpleRecordType").add_property(p).insert()
assert_true(rt.is_valid())
r1 = db.Record().add_parent(rt).add_property(p, value="1.0").insert()
assert_true(r1.is_valid())
r2 = db.Record().add_parent(rt).add_property(p, value="2.0").insert()
assert_true(r2.is_valid())
s = db.execute_query("FIND SimpleRecordType")
assert_equals(3, len(s))
assert_true(s.is_valid())
s = db.execute_query("FIND RECORD SimpleRecordType")
assert_equals(2, len(s))
assert_true(s.is_valid())
s = db.execute_query("FIND SimpleRecordType WITH SimpleDoubleProperty")
assert_equals(3, len(s))
assert_true(s.is_valid())
s = db.execute_query(
"FIND RECORD SimpleRecordType WITH SimpleDoubleProperty")
assert_equals(2, len(s))
assert_true(s.is_valid())
s = db.execute_query(
"FIND SimpleRecordType WITH SimpleDoubleProperty>0")
assert_equals(2, len(s))
assert_true(s.is_valid())
s = db.execute_query(
"FIND SimpleRecordType WITH THE GREATEST SimpleDoubleProperty>0")
assert_equals(1, len(s))
assert_true(s.is_valid())
assert_equals(
r2.id,
db.execute_query(
"FIND SimpleRecordType WITH THE GREATEST SimpleDoubleProperty>0",
unique=True).id)
assert_equals(
r2.id,
db.execute_query(
"FIND SimpleRecordType WITH THE GREATEST SimpleDoubleProperty",
unique=True).id)
assert_equals(
r1.id,
db.execute_query(
"FIND SimpleRecordType WITH THE SMALLEST SimpleDoubleProperty",
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
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_true(obs.is_valid())
tstep = db.Property(
name="ticket147_TimeStep", datatype="DOUBLE").insert()
assert_true(tstep.is_valid())
bmsim = db.RecordType(name="ticket147_BarkleyModelSimulation"
).add_property(tstep).insert()
assert_true(bmsim.is_valid())
frm = db.RecordType(name="ticket147_FrequencyMeasurement"
).add_property(obs).add_property(bmsim).insert()
assert_true(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_true(c.is_valid())
for e in c:
assert_true(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_true(c2.is_valid())
'''
s = db.execute_query("FIND ticket147*")
assert_true(s.is_valid())
assert_equals(14, len(s))
s = db.execute_query("FIND ticket147_Bar*")
assert_true(s.is_valid())
assert_equals(6, len(s))
s = db.execute_query("FIND RECORD ticket147_Bar*")
assert_true(s.is_valid())
assert_equals(5, len(s))
s = db.execute_query("FIND ticket147_Fre*")
assert_true(s.is_valid())
assert_equals(6, len(s))
s = db.execute_query("FIND RECORD ticket147_Fre*")
assert_true(s.is_valid())
assert_equals(5, len(s))
s = db.execute_query("FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius=2.0")
assert_true(s.is_valid())
assert_equals(2, len(s))
s = db.execute_query("FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius='2.0'")
assert_true(s.is_valid())
assert_equals(2, len(s))
s = db.execute_query("FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_BarkleyModelSimulation")
assert_true(s.is_valid())
assert_equals(6, len(s))
s = db.execute_query("FIND Record ticket147_Fre* WHICH HAS A PROPERTY ticket147_BarkleyModelSimulation")
assert_true(s.is_valid())
assert_equals(5, len(s))'''
s = db.execute_query(
"FIND Record ticket147_Fre* WHICH HAS A PROPERTY ticket147_BarkleyModelSimulation WHICH HAS A ticket147_TimeStep"
)
assert_true(s.is_valid())
assert_equals(5, len(s))
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS A PROPERTY ticket147_ObstacleRadius='2.0' AND A PROPERTY ( ticket147_BarkleyModelSimulation )"
)
assert_true(s.is_valid())
assert_equals(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_true(s.is_valid())
assert_equals(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_true(s.is_valid())
assert_equals(2, len(s))
assert_equals(r22.id, s[0].id)
assert_equals(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_true(s.is_valid())
assert_equals(1, len(s))
assert_equals(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_true(s.is_valid())
assert_equals(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_true(s.is_valid())
assert_equals(1, len(s))
assert_equals(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_true(s.is_valid())
assert_equals(1, len(s))
assert_equals(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_true(s.is_valid())
assert_equals(1, len(s))
assert_equals(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_true(s.is_valid())
assert_equals(1, len(s))
assert_equals(r21.id, s[0].id)
# typo: SMALLES
try:
s = db.execute_query(
"FIND ticket147_Fre* WHICH HAS THE SMALLEST ticket147_ObstacleRadius AND A PROPERTY ( ticket147_BarkleyModelSimulation WHICH HAS THE SMALLES ticket147_TimeStep)"
)
raise AssertionError("This should raise a TransactionError")
except TransactionError:
pass
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_true(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_true(nt_data.is_valid())
assert_equals(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_true(nt_data.is_valid())
assert_equals(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_true(nt_data.is_valid())
assert_equals(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_true(nt_data.is_valid())
assert_equals(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_equals(0, db.execute_query("COUNT SimpleTextProperty"))
p = db.Property(
name="SimpleTextProperty",
description="simple text property (from test_tickets.py:test_ticket173)",
datatype='text')
p.insert()
assert_true(p.is_valid())
assert_equals(1, db.execute_query("COUNT SimpleTextProperty"))
finally:
try:
p.delete()
except BaseException:
pass
def test_ticket_166():
try:
# structure
obs = db.Property(
name="ticket166_ObstacleRadius", datatype="DOUBLE").insert()
assert_true(obs.is_valid())
tstep = db.Property(
name="ticket166_TimeStep", datatype="DOUBLE").insert()
assert_true(tstep.is_valid())
bmsim = db.RecordType(name="ticket166_BarkleyModelSimulation"
).add_property(tstep).insert()
assert_true(bmsim.is_valid())
frm = db.RecordType(name="ticket166_FrequencyMeasurement"
).add_property(obs).add_property(bmsim).insert()
assert_true(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_true(c.is_valid())
for e in c:
assert_true(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_true(c2.is_valid())
# retrieve all 10
q = db.Query("FIND RECORD ticket166*")
assert_equals(10, len(q.execute()))
# retrieve first 5
q.putFlag("P", "0L5")
ret = q.execute()
assert_equals(5, len(ret))
for e in ret:
assert_equals(bmsim.id, e.get_parents()[0].id)
# retrieve from 5 to 9 (4 entities)
q.putFlag("P", "5L4")
ret = q.execute()
assert_equals(4, len(ret))
for e in ret:
assert_equals(frm.id, e.get_parents()[0].id)
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_161():
try:
rt = db.RecordType(name="RecordTypeTest")
rt.insert()
assert_true(rt.is_valid())
rt2 = db.RecordType(name="RecordTypeTest")
assert_raises(UniqueNamesError, rt2.insert)
finally:
try:
rt2.delete()
except BaseException:
pass
try:
rt.delete()
except BaseException:
pass
def test_ticket_178():
from lxml import etree
# lxml umlaut
xml = etree.Element("Björn")
assert_equals(xml.tag, "Björn")
try:
rt = db.RecordType(name="Person")
rt.insert()
assert_true(rt.is_valid())
p = db.Property(name="First name", datatype="TEXT")
p.insert()
assert_true(p.is_valid())
r = db.Record().add_parent(parent=rt).add_property(
name="First name", value="Björn")
r.insert()
assert_true(r.is_valid())
assert_true(
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_true(rt.is_valid())
p = db.Property(name="First name", datatype="TEXT")
p.insert()
assert_true(p.is_valid())
r = db.Record().add_parent(parent=rt).add_property(
name="First name", value="Björn")
r.insert()
assert_true(r.is_valid())
r2 = db.execute_query("Find Person WITH 'First name'=Björn")[0]
assert_true(r2.is_valid())
assert_equals(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_true(p.is_valid())
rt = db.RecordType(name="RecordTypeTest").add_property(
name="SomePropertyTest", importance="obligatory", value="blabla")
rt.insert()
assert_true(rt.is_valid())
rt2 = db.execute_query(
"FIND Entity . SomePropertyTest=blabla", unique=True)
assert_true(rt2.is_valid())
assert_equals(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_true(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_true(c.is_valid())
for e in c:
assert_true(e.is_valid())
assert_equals(
rt_b.id,
db.execute_query(
"FIND TestEntity WHICH IS REFERENCED BY TestAnnotation",
unique=True).id)
assert_equals(rt_b.id,
db.execute_query(
"FIND Entity WHICH IS REFERENCED BY TestAnnotation",
unique=True).id)
assert_equals(
rt_b.id,
db.execute_query(
"FIND RecordType WHICH IS REFERENCED BY TestAnnotation",
unique=True).id)
assert_equals(
0,
len(
db.execute_query(
"FIND Record WHICH IS REFERENCED BY TestAnnotation")))
assert_equals(
rt_b.id,
db.execute_query(
"FIND TestEntity WHICH IS REFERENCED WITH TestComment",
unique=True).id)
assert_equals(
rt_b.id,
db.execute_query(
"FIND Entity WHICH IS REFERENCED BY TestAnnotation WITH TestComment",
unique=True).id)
assert_equals(
rt_b.id,
db.execute_query(
"FIND RecordType WHICH IS REFERENCED BY TestAnnotation WITH TestComment",
unique=True).id)
assert_equals(0, len(db.execute_query(
"FIND Record WHICH IS REFERENCED BY TestAnnotation WITH TestComment")))
assert_equals(
rt_b.id,
db.execute_query(
"FIND TestEntity WHICH IS REFERENCED BY TestAnnotation WITH TestComment",
unique=True).id)
assert_equals(
rt_b.id,
db.execute_query(
"FIND Entity WHICH IS REFERENCED BY TestAnnotation WITH TestComment",
unique=True).id)
assert_equals(
rt_b.id,
db.execute_query(
"FIND RecordType WHICH IS REFERENCED BY TestAnnotation WITH TestComment",
unique=True).id)
assert_equals(0, len(db.execute_query(
"FIND Record WHICH IS REFERENCED BY TestAnnotation WITH TestComment")))
assert_equals(
rt_b.id,
db.execute_query(
"FIND TestEntity WHICH IS REFERENCED BY TestAnnotation WITH TestComment='blablabla'",
unique=True).id)
assert_equals(
rt_b.id,
db.execute_query(
"FIND Entity WHICH IS REFERENCED BY TestAnnotation WITH TestComment='blablabla'",
unique=True).id)
assert_equals(
rt_b.id,
db.execute_query(
"FIND RecordType WHICH IS REFERENCED BY TestAnnotation WITH TestComment='blablabla'",
unique=True).id)
assert_equals(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_true(c.is_valid())
assert_true(rt1.is_valid())
for i in range(len(rt1.get_properties())):
assert_equals(i + 1.0, float(rt1.get_properties()[i].value))
assert_true(rt2.is_valid())
for i in range(len(rt2.get_properties())):
assert_equals(i + 1.0, float(rt2.get_properties()[i].value))
assert_true(rt3.is_valid())
for i in range(len(rt3.get_properties())):
assert_equals(i + 1.0, float(rt3.get_properties()[i].value))
assert_true(rt4.is_valid())
for i in range(len(rt4.get_properties())):
assert_equals(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_true(rt4.is_valid())
assert_equals(1.0, float(rt4.get_properties()[0].value))
assert_equals(3.0, float(rt4.get_properties()[1].value))
assert_equals(2.0, float(rt4.get_properties()[2].value))
assert_equals(4.0, float(rt4.get_properties()[3].value))
# retrieve again:
rt42 = db.execute_query("FIND Ticket174_RecordType4", unique=True)
assert_equals(1.0, float(rt42.get_properties()[0].value))
assert_equals(3.0, float(rt42.get_properties()[1].value))
assert_equals(2.0, float(rt42.get_properties()[2].value))
assert_equals(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_equals(
rec.id,
db.execute_query(
"FIND Record SimulationTestRecordType WHICH WAS CREATED BY ME",
unique=True).id)
assert_equals(
rec.id,
db.execute_query(
"FIND Record SimulationTestRecordType WHICH WAS CREATED TODAY",
unique=True).id)
assert_equals(
rec.id,
db.execute_query(
"FIND Record SimulationTestRecordType WHICH WAS CREATED TODAY BY ME",
unique=True).id)
assert_equals(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():
try:
# Properties:
Email = db.Property(name="EmailAddressTest", datatype="TEXT").insert()
Fname = db.Property(name="FirstNameTest", datatype="TEXT").insert()
Lname = db.Property(name="LastNameTest", datatype="TEXT").insert()
# Record Types:
Person = db.RecordType(name="PersonTestRecordType").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="PersonWithEmailTestRecordType") .add_parent(
parent=db.RecordType("PersonTestRecordType").retrieve(),
inheritance="all") .add_property(
property=Email).insert()
assert_true(Person2.is_valid())
assert_equals(Email.id, Person2.get_property("EmailAddressTest").id)
assert_equals(Fname.id, Person2.get_property("FirstNameTest").id)
assert_equals(Lname.id, Person2.get_property("LastNameTest").id)
p2 = db.execute_query(
"FIND RecordType PersonWithEmailTestRecordType", unique=True)
assert_equals(p2.id, Person2.id)
assert_equals(Email.id, p2.get_property("EmailAddressTest").id)
assert_equals(Fname.id, p2.get_property("FirstNameTest").id)
assert_equals(Lname.id, p2.get_property("LastNameTest").id)
finally:
try:
Person2.delete()
except BaseException:
pass
try:
Person.delete()
except BaseException:
pass
try:
Email.delete()
except BaseException:
pass
try:
Fname.delete()
except BaseException:
pass
try:
Lname.delete()
except BaseException:
pass
def test_ticket_155():
try:
# Properties:
p = db.Property(
name="SimpleTextProperty",
description="standard desc",
datatype="TEXT").insert()
assert_true(p.is_valid())
assert_equals(p.description, "standard desc")
p2 = db.execute_query("FIND SimpleTextProperty", True)
assert_true(p2.is_valid())
assert_equals(p2.description, "standard desc")
rt = db.RecordType(name="SimpleRecordType").add_property(
p, description="overridden standard desc").insert()
assert_true(rt.is_valid())
assert_equals(
rt.get_property("SimpleTextProperty").description,
"overridden standard desc")
rt2 = db.execute_query("FIND SimpleRecordType", True)
assert_true(rt2.is_valid())
assert_equals(
rt2.get_property("SimpleTextProperty").description,
"overridden standard desc")
rt3 = db.RecordType(name="SimpleRecordType2").add_parent(
rt, inheritance="all").insert()
assert_true(rt3.is_valid())
assert_equals(
rt3.get_property("SimpleTextProperty").description,
"overridden standard desc")
rt4 = db.execute_query("FIND SimpleRecordType2", True)
assert_true(rt4.is_valid())
assert_equals(
rt4.get_property("SimpleTextProperty").description,
"overridden standard desc")
finally:
try:
rt3.delete()
except BaseException:
pass
try:
rt.delete()
except BaseException:
pass
try:
p.delete()
except BaseException:
pass
def test_ticket_191():
try:
rt = db.RecordType(name="RecordTypeTest")
rt.insert()
c = db.Container().retrieve(
[rt.id, rt.id], raise_exception_on_error=False, sync=False)
print(c)
assert_equals(len(c), 1)
finally:
try:
rt.delete()
except BaseException:
pass
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/"
print(wrongbasepath)
assert_raises(
CaosDBException, db.configure_connection(url=wrongbasepath)._login)
finally:
db.configure_connection()
def test_ticket_115():
try:
p1 = db.Property(name="SimpleDoubleProperty1", datatype=db.DOUBLE)
p2 = db.Property(name="SimpleDoubleProperty2", datatype=db.DOUBLE)
rt = db.RecordType(
name="SimpleRecordType").add_property(p1).add_property(p2)
c = db.Container().extend([p1, p2, rt])
c.insert()
assert_true(c.is_valid())
rtc = db.execute_query("FIND SimpleRecordType", unique=True)
assert_equals(len(rtc.get_properties()), 2)
rtc.get_properties().remove("SimpleDoubleProperty1")
assert_equals(len(rtc.get_properties()), 1)
rtc.update()
assert_true(c.is_valid())
assert_equals(len(rtc.get_properties()), 1)
finally:
try:
db.execute_query("FIND Simple*").delete()
except BaseException:
pass