Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import unittest
import caosdb as db
from caosadvancedtools.models.data_model import DataModel
class DataModelTest(unittest.TestCase):
def test_creation(self):
# create RT and one property
dm = DataModel()
dm.append(db.RecordType(name="TestRecord"))
dm.append(db.Property(name="testproperty", datatype=db.INTEGER))
dm.sync_data_model(noquestion=True)
db.execute_query("FIND RECORDTYPE TestRecord", unique=True)
db.execute_query("FIND PROPERTY testproperty", unique=True)
# add the property to the RT
dm = DataModel()
dm.extend([
db.RecordType(name="TestRecord").add_property(name="testproperty"),
db.Property(name="testproperty", datatype=db.INTEGER)])
dm.sync_data_model(noquestion=True)
rt = db.execute_query("FIND RECORDTYPE TestRecord", unique=True)
assert rt.get_property("testproperty") is not None
# replace the one property
dm = DataModel([
db.RecordType(name="TestRecord").add_property(name="test"),
db.Property(name="test", datatype=db.INTEGER)])
dm.sync_data_model(noquestion=True)
db.execute_query("FIND RECORDTYPE TestRecord", unique=True)
rt = db.execute_query("FIND RECORDTYPE TestRecord", unique=True)
assert rt.get_property("test") is not None
def test_missing(self):
# Test sync with missing prop
# insert propt
dm = DataModel([db.Property(name="testproperty", datatype=db.INTEGER)])
dm.sync_data_model(noquestion=True)
# insert RT using the prop separatly
maintained = {"one": db.RecordType(name="TestRecord").add_property(
name="testproperty")}
dm = DataModel(maintained.values())
dm.sync_data_model(noquestion=True)
rt = db.execute_query("FIND RECORDTYPE TestRecord", unique=True)
assert rt.get_property("testproperty") is not None
def test_get_existing_entities(self):
db.RecordType(name="TestRecord").insert()
c = db.Container().extend([
db.Property(name="test"),
db.RecordType(name="TestRecord")])
exist = DataModel.get_existing_entities(c)
assert len(exist) == 1
assert exist[0].name == "TestRecord"
def tearDown(self):
try:
tests = db.execute_query("FIND test*")
tests.delete()
except Exception:
pass