Select Git revision
test_tickets.py
-
Timm Fitschen authored
AGPLv3 Veröffentlichung gemäß Dienstanweisung vom 15. August 2018.
Timm Fitschen authoredAGPLv3 Veröffentlichung gemäß Dienstanweisung vom 15. August 2018.
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