Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • caosdb/src/caosdb-pyinttest
1 result
Show changes
Commits on Source (4)
......@@ -24,6 +24,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
evaluated correctly.
* Test for [caosdb-server#99](https://gitlab.com/caosdb/caosdb-server/-/issues/99).
* Tests for retrieving older versions.
* Tests for `BEFORE`, `AFTER`, `UNTIL`, `SINCE` keywords for query
transaction filters
[caosdb-server#132](https://gitlab.indiscale.com/caosdb/src/caosdb-server/-/issues/132)
### Changed (for changes in existing functionality)
......
......@@ -29,27 +29,27 @@
import os
import random
import caosdb as h
import caosdb as db
from pytest import mark
from nose.tools import assert_true, assert_equal, assert_is_not_none
from pytest import mark, raises
from caosdb.connection.connection import get_connection
from caosdb.exceptions import EmptyUniqueQueryError
from lxml import etree
def setup_module():
h.administration.set_server_property("AUTH_OPTIONAL", "TRUE")
db.administration.set_server_property("AUTH_OPTIONAL", "TRUE")
def setup():
try:
h.execute_query("FIND Test*").delete()
db.execute_query("FIND Test*").delete()
except Exception as e:
print(e)
def teardown():
h.configure_connection()
db.configure_connection()
setup()
try:
os.remove("test.dat")
......@@ -58,120 +58,85 @@ def teardown():
def test_query_with_reference_by_parent():
h.RecordType("TestExperiment").insert()
h.RecordType("TestProtocolLog").insert()
h.RecordType("TestSpecialProtocolLog").add_parent(
db.RecordType("TestExperiment").insert()
db.RecordType("TestProtocolLog").insert()
db.RecordType("TestSpecialProtocolLog").add_parent(
name="TestProtocolLog").insert()
sp_rec = h.Record("TestSpecialProtocolLog1").add_parent(
sp_rec = db.Record("TestSpecialProtocolLog1").add_parent(
name="TestSpecialProtocolLog").insert()
exp_rec = h.Record("TestExperiment1").add_parent(
exp_rec = db.Record("TestExperiment1").add_parent(
name="TestExperiment").add_property(
name="TestProtocolLog",
datatype=h.LIST("TestProtocolLog"),
datatype=db.LIST("TestProtocolLog"),
value=[
sp_rec.id]).insert()
assert_equal(2, len(h.execute_query("FIND TestExperiment")))
assert_equal(
exp_rec.id,
h.execute_query(
"FIND TestExperiment WHICH HAS A TestProtocolLog",
unique=True).id)
assert_equal(
exp_rec.id,
h.execute_query(
"FIND TestExperiment.TestProtocolLog",
unique=True).id)
assert_equal(
exp_rec.id, h.execute_query(
"FIND TestExperiment WHICH REFERENCES " + str(sp_rec.id),
unique=True).id)
assert_equal(
exp_rec.id,
h.execute_query(
"FIND TestExperiment WHICH REFERENCES TestSpecialProtocolLog1",
unique=True).id)
assert_equal(
exp_rec.id,
h.execute_query(
"FIND TestExperiment WHICH REFERENCES TestSpecialProtocolLog",
unique=True).id)
assert_equal(exp_rec.id, h.execute_query(
"FIND TestExperiment .-> " + str(sp_rec.id), unique=True).id)
assert_equal(
exp_rec.id,
h.execute_query(
"FIND TestExperiment .-> TestSpecialProtocolLog1",
unique=True).id)
assert_equal(
exp_rec.id,
h.execute_query(
"FIND TestExperiment .-> TestSpecialProtocolLog",
unique=True).id)
assert_equal(0, len(h.execute_query(
"FIND TestExperiment WHICH HAS A TestProtocolLog1")))
assert_equal(0, len(h.execute_query(
"FIND TestExperiment.TestProtocolLog1")))
assert 2 == len(db.execute_query("FIND TestExperiment"))
assert exp_rec.id == db.execute_query(
"FIND TestExperiment WHICH HAS A TestProtocolLog", unique=True).id
assert exp_rec.id == db.execute_query(
"FIND TestExperiment.TestProtocolLog", unique=True).id
assert exp_rec.id == db.execute_query(
"FIND TestExperiment WHICH REFERENCES " + str(sp_rec.id), unique=True).id
assert exp_rec.id == db.execute_query(
"FIND TestExperiment WHICH REFERENCES TestSpecialProtocolLog1",
unique=True).id
assert exp_rec.id == db.execute_query(
"FIND TestExperiment WHICH REFERENCES TestSpecialProtocolLog",
unique=True).id
assert exp_rec.id == db.execute_query(
"FIND TestExperiment .-> " + str(sp_rec.id), unique=True).id
assert exp_rec.id == db.execute_query(
"FIND TestExperiment .-> TestSpecialProtocolLog1", unique=True).id
assert exp_rec.id == db.execute_query(
"FIND TestExperiment .-> TestSpecialProtocolLog", unique=True).id
assert 0 == len(db.execute_query(
"FIND TestExperiment WHICH HAS A TestProtocolLog1"))
assert 0 == len(db.execute_query("FIND TestExperiment.TestProtocolLog1"))
def test_query_with_domains():
person = h.RecordType("TestPerson").insert()
h.Property("TestFirstName", datatype=h.TEXT).insert()
h.Property("TestConductor", datatype=h.REFERENCE).insert()
person = db.RecordType("TestPerson").insert()
db.Property("TestFirstName", datatype=db.TEXT).insert()
db.Property("TestConductor", datatype=db.REFERENCE).insert()
# TODO: new test for TestFirstName with overridden datatype=person
dan = h.Record(
dan = db.Record(
name="TestDaniel").add_property(
name="TestFirstName",
value="Daniel").add_parent(person).insert()
exp = h.RecordType(
exp = db.RecordType(
name="TestExperiment").add_property(
name="TestConductor",
datatype=person,
value=dan.id).insert()
assert_equal(
h.execute_query(
"FIND TestExperiment WHICH HAS A TestConductor->TestPerson",
unique=True).id,
exp.id)
assert_equal(
h.execute_query(
"FIND TestExperiment WHICH HAS A TestConductor=TestPerson",
unique=True).id,
exp.id)
assert_equal(
h.execute_query(
"FIND TestExperiment WHICH HAS A TestConductor=" + str(dan.id),
unique=True).id, exp.id)
assert_equal(
h.execute_query(
"FIND TestExperiment",
unique=True).id,
exp.id)
assert_equal(
h.execute_query(
"FIND TestExperiment WHICH HAS A TestConductor WHICH has a TestFirstName=Daniel",
unique=True).id,
exp.id)
assert db.execute_query(
"FIND TestExperiment WHICH HAS A TestConductor->TestPerson",
unique=True).id == exp.id
assert db.execute_query(
"FIND TestExperiment WHICH HAS A TestConductor=TestPerson",
unique=True).id == exp.id
assert db.execute_query(
"FIND TestExperiment WHICH HAS A TestConductor=" + str(dan.id), unique=True).id == exp.id
assert db.execute_query("FIND TestExperiment", unique=True).id == exp.id
assert db.execute_query(
"FIND TestExperiment WHICH HAS A TestConductor WHICH has a TestFirstName=Daniel",
unique=True).id == exp.id
def test_query1():
p = (
h.Property(
db.Property(
name="TestTextProperty",
description="Test text property (from test_tenpoints.py)",
datatype='text'))
p.insert()
assert_true(int(p.id) >= 100)
assert_equal(p.id, h.execute_query("FIND TestTextProperty")[0].id)
assert_equal(
p.id,
h.execute_query(
"FIND TestTextProperty",
unique=True).id)
assert int(p.id) >= 100
assert p.id == db.execute_query("FIND TestTextProperty")[0].id
assert p.id == db.execute_query("FIND TestTextProperty", unique=True).id
def test_query2():
......@@ -181,20 +146,18 @@ def test_query2():
f.close()
''' prepare file record '''
f = h.File(name="TestFile", path='testfiles/testfile', file="test.dat")
f = db.File(name="TestFile", path='testfiles/testfile', file="test.dat")
f.insert()
f2 = h.File(name="TestFile2", path='testfiles/testfile2', file="test.dat")
f2 = db.File(name="TestFile2", path='testfiles/testfile2', file="test.dat")
f2.insert()
assert_true(int(f.id) >= 100)
assert_equal(2, len(h.execute_query("FIND FILE")))
assert_equal(
f.id,
h.execute_query(
"FIND FILE WHICH IS STORED AT testfiles/testfile",
unique=True).id)
assert_equal(f.id, h.File(path='testfiles/testfile').retrieve().id)
assert int(f.id) >= 100
assert 2 == len(db.execute_query("FIND FILE"))
assert f.id == db.execute_query(
"FIND FILE WHICH IS STORED AT testfiles/testfile",
unique=True).id
assert f.id == db.File(path='testfiles/testfile').retrieve().id
def test_query3():
......@@ -206,347 +169,345 @@ def test_query3():
assert xml.xpath("/Response/Query")
assert xml.xpath("/Response/UserInfo")
assert xml.xpath("/Response/noscript")
assert_equal(4, len(xml))
assert 4 == len(xml)
@mark.slow
def test_conjunction():
rt = h.RecordType(name="TestConjunctionTest").insert()
assert_true(rt.is_valid())
rt = db.RecordType(name="TestConjunctionTest").insert()
assert rt.is_valid()
pa = h.Property(
pa = db.Property(
name="TestConjunctionTestPropertyA",
datatype="INTEGER").insert()
assert_true(pa.is_valid())
assert pa.is_valid()
pb = h.Property(
pb = db.Property(
name="TestConjunctionTestPropertyB",
datatype="INTEGER").insert()
assert_true(pb.is_valid())
assert pb.is_valid()
ra = h.Record(
ra = db.Record(
name="TestA").add_parent(rt).add_property(
property=pa,
value="1").add_property(
property=pb,
value="0")
rb = h.Record(
rb = db.Record(
name="TestB").add_parent(rt).add_property(
property=pa,
value="0").add_property(
property=pb,
value="1")
rab = h.Record(
rab = db.Record(
name="TestAB").add_parent(rt).add_property(
property=pa,
value="1").add_property(
property=pb,
value="1")
rn = h.Record(
rn = db.Record(
name="TestN").add_parent(rt).add_property(
property=pa,
value="0").add_property(
property=pb,
value="0")
c = h.Container().extend([ra, rb, rab, rn]).insert()
assert_true(c.is_valid())
assert_equal(5, len(h.execute_query("FIND TestConjunctionTest")))
assert_equal(4, len(h.execute_query("FIND Record TestConjunctionTest")))
assert_equal(4, len(h.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA")))
assert_equal(4, len(h.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyB")))
assert_equal(2, len(h.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyB=1")))
assert_equal(2, len(h.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyB=0")))
assert_equal(2, len(h.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=1")))
assert_equal(2, len(h.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=0")))
assert_equal(1, len(h.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=1 AND TestConjunctionTestPropertyB=1")))
assert_equal(1, len(h.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=0 AND TestConjunctionTestPropertyB=0")))
assert_equal(1, len(h.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=0 AND TestConjunctionTestPropertyB=1")))
assert_equal(1, len(h.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=1 AND TestConjunctionTestPropertyB=0")))
assert_equal(4, len(h.execute_query(
"FIND Record . TestConjunctionTestPropertyA")))
assert_equal(4, len(h.execute_query(
"FIND Record . TestConjunctionTestPropertyB")))
assert_equal(2, len(h.execute_query(
"FIND Record . TestConjunctionTestPropertyB=1")))
assert_equal(2, len(h.execute_query(
"FIND Record . TestConjunctionTestPropertyB=0")))
assert_equal(2, len(h.execute_query(
"FIND Record . TestConjunctionTestPropertyA=1")))
assert_equal(2, len(h.execute_query(
"FIND Record . TestConjunctionTestPropertyA=0")))
assert_equal(1, len(h.execute_query(
"FIND RECORD . TestConjunctionTestPropertyA=1 AND TestConjunctionTestPropertyB=1")))
assert_equal(1, len(h.execute_query(
"FIND RECORD . TestConjunctionTestPropertyA=0 AND TestConjunctionTestPropertyB=0")))
assert_equal(1, len(h.execute_query(
"FIND RECORD . TestConjunctionTestPropertyA=0 AND TestConjunctionTestPropertyB=1")))
assert_equal(1, len(h.execute_query(
"FIND RECORD . TestConjunctionTestPropertyA=1 AND TestConjunctionTestPropertyB=0")))
c = db.Container().extend([ra, rb, rab, rn]).insert()
assert c.is_valid()
assert 5 == len(db.execute_query("FIND TestConjunctionTest"))
assert 4 == len(db.execute_query("FIND Record TestConjunctionTest"))
assert 4 == len(db.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA"))
assert 4 == len(db.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyB"))
assert 2 == len(db.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyB=1"))
assert 2 == len(db.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyB=0"))
assert 2 == len(db.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=1"))
assert 2 == len(db.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=0"))
assert 1 == len(db.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=1 AND TestConjunctionTestPropertyB=1"))
assert 1 == len(db.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=0 AND TestConjunctionTestPropertyB=0"))
assert 1 == len(db.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=0 AND TestConjunctionTestPropertyB=1"))
assert 1 == len(db.execute_query(
"FIND TestConjunctionTest . TestConjunctionTestPropertyA=1 AND TestConjunctionTestPropertyB=0"))
assert 4 == len(db.execute_query(
"FIND Record . TestConjunctionTestPropertyA"))
assert 4 == len(db.execute_query(
"FIND Record . TestConjunctionTestPropertyB"))
assert 2 == len(db.execute_query(
"FIND Record . TestConjunctionTestPropertyB=1"))
assert 2 == len(db.execute_query(
"FIND Record . TestConjunctionTestPropertyB=0"))
assert 2 == len(db.execute_query(
"FIND Record . TestConjunctionTestPropertyA=1"))
assert 2 == len(db.execute_query(
"FIND Record . TestConjunctionTestPropertyA=0"))
assert 1 == len(db.execute_query(
"FIND RECORD . TestConjunctionTestPropertyA=1 AND TestConjunctionTestPropertyB=1"))
assert 1 == len(db.execute_query(
"FIND RECORD . TestConjunctionTestPropertyA=0 AND TestConjunctionTestPropertyB=0"))
assert 1 == len(db.execute_query(
"FIND RECORD . TestConjunctionTestPropertyA=0 AND TestConjunctionTestPropertyB=1"))
assert 1 == len(db.execute_query(
"FIND RECORD . TestConjunctionTestPropertyA=1 AND TestConjunctionTestPropertyB=0"))
@mark.slow
def test_disjunction():
rt = h.RecordType(name="TestDisjunctionTest").insert()
assert_true(rt.is_valid())
rt = db.RecordType(name="TestDisjunctionTest").insert()
assert rt.is_valid()
pa = h.Property(
pa = db.Property(
name="TestDisjunctionTestPropertyA",
datatype="INTEGER").insert()
assert_true(pa.is_valid())
assert pa.is_valid()
pb = h.Property(
pb = db.Property(
name="TestDisjunctionTestPropertyB",
datatype="DOUBLE").insert()
assert_true(pb.is_valid())
assert pb.is_valid()
ra = h.Record(
ra = db.Record(
name="TestA").add_parent(rt).add_property(
property=pa,
value="1").add_property(
property=pb,
value="0")
rb = h.Record(
rb = db.Record(
name="TestB").add_parent(rt).add_property(
property=pa,
value="0").add_property(
property=pb,
value="1")
rab = h.Record(
rab = db.Record(
name="TestAB").add_parent(rt).add_property(
property=pa,
value="1").add_property(
property=pb,
value="1")
rn = h.Record(
rn = db.Record(
name="TestN").add_parent(rt).add_property(
property=pa,
value="0").add_property(
property=pb,
value="0")
c = h.Container().extend([ra, rb, rab, rn]).insert()
assert_true(c.is_valid())
assert_equal(5, len(h.execute_query("FIND TestDisjunctionTest")))
assert_equal(4, len(h.execute_query("FIND Record TestDisjunctionTest")))
assert_equal(4, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA")))
assert_equal(4, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyB")))
assert_equal(2, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyB=1")))
assert_equal(2, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyB=0")))
assert_equal(2, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=1")))
assert_equal(2, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=0")))
assert_equal(3, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=1 OR TestDisjunctionTestPropertyB=1")))
assert_equal(3, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=0 OR TestDisjunctionTestPropertyB=0")))
assert_equal(3, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=0 OR TestDisjunctionTestPropertyB=1")))
assert_equal(3, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=1 OR TestDisjunctionTestPropertyB=0")))
assert_equal(2, len(h.execute_query(
"FIND TestDisjunctionTest . THE GREATEST TestDisjunctionTestPropertyB")))
assert_equal(3, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=1 OR THE GREATEST TestDisjunctionTestPropertyB")))
assert_equal(
rn.id,
h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA='-1' OR ( TestDisjunctionTestPropertyB=0 AND TestDisjunctionTestPropertyA=0)",
unique=True).id)
assert_equal(3, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=0 OR ( TestDisjunctionTestPropertyB=0 AND TestDisjunctionTestPropertyA=1)")))
assert_equal(2, len(h.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=1 OR ( TestDisjunctionTestPropertyB=0 AND TestDisjunctionTestPropertyA=1)")))
c = db.Container().extend([ra, rb, rab, rn]).insert()
assert c.is_valid()
assert 5 == len(db.execute_query("FIND TestDisjunctionTest"))
assert 4 == len(db.execute_query("FIND Record TestDisjunctionTest"))
assert 4 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA"))
assert 4 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyB"))
assert 2 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyB=1"))
assert 2 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyB=0"))
assert 2 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=1"))
assert 2 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=0"))
assert 3 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=1 OR TestDisjunctionTestPropertyB=1"))
assert 3 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=0 OR TestDisjunctionTestPropertyB=0"))
assert 3 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=0 OR TestDisjunctionTestPropertyB=1"))
assert 3 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=1 OR TestDisjunctionTestPropertyB=0"))
assert 2 == len(db.execute_query(
"FIND TestDisjunctionTest . THE GREATEST TestDisjunctionTestPropertyB"))
assert 3 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=1 OR THE GREATEST TestDisjunctionTestPropertyB"))
assert rn.id == db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA='-1' OR ( TestDisjunctionTestPropertyB=0 AND TestDisjunctionTestPropertyA=0)",
unique=True).id
assert 3 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=0 OR ( TestDisjunctionTestPropertyB=0 AND TestDisjunctionTestPropertyA=1)"))
assert 2 == len(db.execute_query(
"FIND TestDisjunctionTest . TestDisjunctionTestPropertyA=1 OR ( TestDisjunctionTestPropertyB=0 AND TestDisjunctionTestPropertyA=1)"))
@mark.slow
def test_greatest():
pAB = h.Property(name="TestPropertyAB", datatype=h.DOUBLE).insert()
assert_true(pAB.is_valid())
pA = h.Property(
pAB = db.Property(name="TestPropertyAB", datatype=db.DOUBLE).insert()
assert pAB.is_valid()
pA = db.Property(
name="TestPropertyA",
datatype=h.DOUBLE).add_parent(pAB).insert()
assert_true(pA.is_valid())
pB = h.Property(name="TestPropertyB", datatype=h.DOUBLE).add_parent(pAB)
assert_equal(len(pB.get_parents()), 1)
datatype=db.DOUBLE).add_parent(pAB).insert()
assert pA.is_valid()
pB = db.Property(name="TestPropertyB", datatype=db.DOUBLE).add_parent(pAB)
assert len(pB.get_parents()) == 1
print(pB)
print(pB)
pB.insert()
assert_true(pB.is_valid())
assert pB.is_valid()
rt = h.RecordType(name="TestRecordType").insert()
assert_true(rt.is_valid())
rt = db.RecordType(name="TestRecordType").insert()
assert rt.is_valid()
rec1 = h.Record(name="TestRecord1").add_parent(
rec1 = db.Record(name="TestRecord1").add_parent(
rt).add_property(pA, value=1.0).insert()
assert_true(rec1.is_valid())
assert rec1.is_valid()
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord1 WHICH HAS THE GREATEST TestPropertyA",
unique=True)
assert_equal(c.id, rec1.id)
assert c.id == rec1.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyA",
unique=True)
assert_equal(c.id, rec1.id)
assert c.id == rec1.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord1 WHICH HAS THE GREATEST TestPropertyAB",
unique=True)
assert_equal(c.id, rec1.id)
assert c.id == rec1.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyAB",
unique=True)
assert_equal(c.id, rec1.id)
assert c.id == rec1.id
rec2 = h.Record(name="TestRecord2").add_parent(
rec2 = db.Record(name="TestRecord2").add_parent(
rt).add_property(pA, value=2.0).insert()
assert_true(rec2.is_valid())
assert rec2.is_valid()
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord2 WHICH HAS THE GREATEST TestPropertyA",
unique=True)
assert_equal(c.id, rec2.id)
assert c.id == rec2.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord1 WHICH HAS THE GREATEST TestPropertyA",
unique=True)
assert_equal(c.id, rec1.id)
assert c.id == rec1.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyA",
unique=True)
assert_equal(c.id, rec2.id)
assert c.id == rec2.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord2 WHICH HAS THE GREATEST TestPropertyAB",
unique=True)
assert_equal(c.id, rec2.id)
assert c.id == rec2.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyAB",
unique=True)
assert_equal(c.id, rec2.id)
assert c.id == rec2.id
rec3 = h.Record(name="TestRecord3").add_parent(
rec3 = db.Record(name="TestRecord3").add_parent(
rt).add_property(pA, value=1.5).insert()
assert_true(rec3.is_valid())
assert rec3.is_valid()
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord3 WHICH HAS THE GREATEST TestPropertyA",
unique=True)
assert_equal(c.id, rec3.id)
assert c.id == rec3.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord2 WHICH HAS THE GREATEST TestPropertyA",
unique=True)
assert_equal(c.id, rec2.id)
assert c.id == rec2.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord1 WHICH HAS THE GREATEST TestPropertyA",
unique=True)
assert_equal(c.id, rec1.id)
assert c.id == rec1.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyA",
unique=True)
assert_equal(c.id, rec2.id)
assert c.id == rec2.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord3 WHICH HAS THE GREATEST TestPropertyAB",
unique=True)
assert_equal(c.id, rec3.id)
assert c.id == rec3.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord2 WHICH HAS THE GREATEST TestPropertyAB",
unique=True)
assert_equal(c.id, rec2.id)
assert c.id == rec2.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord1 WHICH HAS THE GREATEST TestPropertyAB",
unique=True)
assert_equal(c.id, rec1.id)
assert c.id == rec1.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyAB",
unique=True)
assert_equal(c.id, rec2.id)
assert c.id == rec2.id
rec4 = h.Record(name="TestRecord4").add_parent(
rec4 = db.Record(name="TestRecord4").add_parent(
rt).add_property(pB, value=0.1).insert()
assert_true(rec4.is_valid())
assert rec4.is_valid()
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord4 WHICH HAS THE GREATEST TestPropertyA")
assert_equal(len(c), 0)
assert len(c) == 0
c = h.execute_query(
c = db.execute_query(
"FIND TestRecord4 WHICH HAS THE GREATEST TestPropertyB",
unique=True)
assert_equal(c.id, rec4.id)
assert c.id == rec4.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyB",
unique=True)
assert_equal(c.id, rec4.id)
assert c.id == rec4.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyAB",
unique=True)
assert_equal(c.id, rec2.id)
assert c.id == rec2.id
rec5 = h.Record(
rec5 = db.Record(
name="TestRecord5").add_parent(rt).add_property(
pB,
value=200).add_property(
pA,
value=1.5).insert()
assert_true(rec5.is_valid())
assert rec5.is_valid()
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyAB",
unique=True)
assert_equal(c.id, rec5.id)
assert c.id == rec5.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyB",
unique=True)
assert_equal(c.id, rec5.id)
assert c.id == rec5.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyA",
unique=True)
assert_equal(c.id, rec2.id)
assert c.id == rec2.id
rec6 = h.Record(
rec6 = db.Record(
name="TestRecord6").add_parent(rt).add_property(
pAB,
value=400).add_property(
......@@ -554,96 +515,96 @@ def test_greatest():
value=100).add_property(
pA,
value=2).insert()
assert_true(rec6.is_valid())
assert rec6.is_valid()
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyAB",
unique=True)
assert_equal(c.id, rec6.id)
assert c.id == rec6.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyB",
unique=True)
assert_equal(c.id, rec5.id)
assert c.id == rec5.id
c = h.execute_query(
c = db.execute_query(
"FIND TestRecordType WHICH HAS THE GREATEST TestPropertyA")
assert_equal(c[0].id, rec2.id)
assert_equal(c[1].id, rec6.id)
assert c[0].id == rec2.id
assert c[1].id == rec6.id
def test_wildcard_values():
ptext = h.Property(name="TestTextProperty", datatype=h.TEXT).insert()
rt = h.RecordType(
ptext = db.Property(name="TestTextProperty", datatype=db.TEXT).insert()
rt = db.RecordType(
name="TestRecordType").add_property(
ptext, value="abcdefg").insert()
assert_true(rt.is_valid())
rt2 = h.execute_query(
assert rt.is_valid()
rt2 = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE "abc*"',
unique=True)
assert_equal(rt2.id, rt.id)
assert rt2.id == rt.id
rt2 = h.execute_query(
rt2 = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE abc*',
unique=True)
assert_equal(rt2.id, rt.id)
assert rt2.id == rt.id
rt2 = h.execute_query(
rt2 = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE abc*efg',
unique=True)
assert_equal(rt2.id, rt.id)
assert rt2.id == rt.id
rt2 = h.execute_query(
rt2 = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE abc*g',
unique=True)
assert_equal(rt2.id, rt.id)
assert rt2.id == rt.id
rt2 = h.execute_query(
rt2 = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE *',
unique=True)
assert_equal(rt2.id, rt.id)
assert rt2.id == rt.id
rt2 = h.execute_query(
rt2 = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE a*',
unique=True)
assert_equal(rt2.id, rt.id)
assert rt2.id == rt.id
rt2 = h.execute_query(
rt2 = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE *abc*',
unique=True)
assert_equal(rt2.id, rt.id)
assert rt2.id == rt.id
rt2 = h.execute_query(
rt2 = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE *fg',
unique=True)
assert_equal(rt2.id, rt.id)
assert rt2.id == rt.id
rt2 = h.execute_query(
rt2 = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE a*f*g',
unique=True)
assert_equal(rt2.id, rt.id)
assert rt2.id == rt.id
rt2 = h.execute_query(
rt2 = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE a*d*g',
unique=True)
assert_equal(rt2.id, rt.id)
assert rt2.id == rt.id
c = h.execute_query(
c = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE abc')
assert_equal(0, len(c))
assert 0 == len(c)
c = h.execute_query(
c = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty = abc*')
assert_equal(0, len(c))
assert 0 == len(c)
c = h.execute_query(
c = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty = *abc*')
assert_equal(0, len(c))
assert 0 == len(c)
c = h.execute_query(
c = db.execute_query(
'FIND TestRecordType WHICH HAS A TestTextProperty LIKE *f')
assert_equal(0, len(c))
assert 0 == len(c)
def store_file(path, name=None, f=None):
......@@ -657,10 +618,10 @@ def store_file(path, name=None, f=None):
if name is None:
name = "TestFile%030x" % random.randrange(16**30)
file_ = h.File(name=name,
description="Testfile Desc",
path=path,
file="test.dat")
file_ = db.File(name=name,
description="Testfile Desc",
path=path,
file="test.dat")
file_.insert()
if tmp_file is not None:
......@@ -690,148 +651,152 @@ def test_stored_at_wildcards():
file8 = store_file("rootdir/subdir1/test%8.dat", f=upload_file)
c = h.execute_query("FIND FILE WHICH IS STORED AT /*.dat")
assert_equal(len(c), 2)
assert_is_not_none(c.get_entity_by_id(file1.id))
assert_is_not_none(c.get_entity_by_id(file5.id))
c = h.execute_query("FIND FILE WHICH IS STORED AT /**/subdir*/*.dat")
assert_equal(len(c), 4)
assert_is_not_none(c.get_entity_by_id(file3.id))
assert_is_not_none(c.get_entity_by_id(file4.id))
assert_is_not_none(c.get_entity_by_id(file7.id))
assert_is_not_none(c.get_entity_by_id(file8.id))
c = h.execute_query("FIND FILE WHICH IS STORED AT *subdir**.dat")
assert_equal(len(c), 4)
assert_is_not_none(c.get_entity_by_id(file3.id))
assert_is_not_none(c.get_entity_by_id(file4.id))
assert_is_not_none(c.get_entity_by_id(file7.id))
assert_is_not_none(c.get_entity_by_id(file8.id))
c = h.execute_query("FIND FILE WHICH IS STORED AT *subdir1**.dat")
assert_equal(len(c), 4)
assert_is_not_none(c.get_entity_by_id(file3.id))
assert_is_not_none(c.get_entity_by_id(file4.id))
assert_is_not_none(c.get_entity_by_id(file7.id))
assert_is_not_none(c.get_entity_by_id(file8.id))
c = h.execute_query("FIND FILE WHICH IS STORED AT *subdir2**.dat")
assert_equal(len(c), 1)
assert_is_not_none(c.get_entity_by_id(file4.id))
c = h.execute_query("FIND FILE WHICH IS STORED AT /rootdir/*.dat")
assert_equal(len(c), 2)
assert_is_not_none(c.get_entity_by_id(file2.id))
assert_is_not_none(c.get_entity_by_id(file6.id))
f = h.execute_query("FIND FILE WHICH IS STORED AT test1.dat", unique=True)
assert_equal(f.id, file1.id)
f = h.execute_query("FIND FILE WHICH IS STORED AT /test1.dat", unique=True)
assert_equal(f.id, file1.id)
f = h.execute_query("FIND FILE WHICH IS STORED AT *test1.dat", unique=True)
assert_equal(f.id, file1.id)
f = h.execute_query(
c = db.execute_query("FIND FILE WHICH IS STORED AT /*.dat")
assert len(c) == 2
assert c.get_entity_by_id(file1.id) is not None
assert c.get_entity_by_id(file5.id) is not None
c = db.execute_query("FIND FILE WHICH IS STORED AT /**/subdir*/*.dat")
assert len(c) == 4
assert c.get_entity_by_id(file3.id) is not None
assert c.get_entity_by_id(file4.id) is not None
assert c.get_entity_by_id(file7.id) is not None
assert c.get_entity_by_id(file8.id) is not None
c = db.execute_query("FIND FILE WHICH IS STORED AT *subdir**.dat")
assert len(c) == 4
assert c.get_entity_by_id(file3.id) is not None
assert c.get_entity_by_id(file4.id) is not None
assert c.get_entity_by_id(file7.id) is not None
assert c.get_entity_by_id(file8.id) is not None
c = db.execute_query("FIND FILE WHICH IS STORED AT *subdir1**.dat")
assert len(c) == 4
assert c.get_entity_by_id(file3.id) is not None
assert c.get_entity_by_id(file4.id) is not None
assert c.get_entity_by_id(file7.id) is not None
assert c.get_entity_by_id(file8.id) is not None
c = db.execute_query("FIND FILE WHICH IS STORED AT *subdir2**.dat")
assert len(c) == 1
assert c.get_entity_by_id(file4.id) is not None
c = db.execute_query("FIND FILE WHICH IS STORED AT /rootdir/*.dat")
assert len(c) == 2
assert c.get_entity_by_id(file2.id) is not None
assert c.get_entity_by_id(file6.id) is not None
f = db.execute_query("FIND FILE WHICH IS STORED AT test1.dat", unique=True)
assert f.id == file1.id
f = db.execute_query(
"FIND FILE WHICH IS STORED AT /test1.dat",
unique=True)
assert f.id == file1.id
f = db.execute_query(
"FIND FILE WHICH IS STORED AT *test1.dat",
unique=True)
assert f.id == file1.id
f = db.execute_query(
"FIND FILE WHICH IS STORED AT /*test1.dat",
unique=True)
assert_equal(f.id, file1.id)
assert f.id == file1.id
f = h.execute_query("FIND FILE WHICH IS STORED AT *1.dat", unique=True)
assert_equal(f.id, file1.id)
f = db.execute_query("FIND FILE WHICH IS STORED AT *1.dat", unique=True)
assert f.id == file1.id
f = h.execute_query("FIND FILE WHICH IS STORED AT /*1.dat", unique=True)
assert_equal(f.id, file1.id)
f = db.execute_query("FIND FILE WHICH IS STORED AT /*1.dat", unique=True)
assert f.id == file1.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT **test1.dat",
unique=True)
assert_equal(f.id, file1.id)
assert f.id == file1.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT /**test1.dat",
unique=True)
assert_equal(f.id, file1.id)
assert f.id == file1.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT rootdir/test2.dat",
unique=True)
assert_equal(f.id, file2.id)
assert f.id == file2.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT /rootdir/test2.dat",
unique=True)
assert_equal(f.id, file2.id)
assert f.id == file2.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT *rootdir/test2.dat",
unique=True)
assert_equal(f.id, file2.id)
assert f.id == file2.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT /*rootdir/test2.dat",
unique=True)
assert_equal(f.id, file2.id)
assert f.id == file2.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT **rootdir/test2.dat",
unique=True)
assert_equal(f.id, file2.id)
assert f.id == file2.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT /**rootdir/test2.dat",
unique=True)
assert_equal(f.id, file2.id)
assert f.id == file2.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT *dir/test2.dat",
unique=True)
assert_equal(f.id, file2.id)
assert f.id == file2.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT /*dir/test2.dat",
unique=True)
assert_equal(f.id, file2.id)
assert f.id == file2.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT **dir/test2.dat",
unique=True)
assert_equal(f.id, file2.id)
assert f.id == file2.id
f = h.execute_query(
f = db.execute_query(
"FIND FILE WHICH IS STORED AT /**dir/test2.dat",
unique=True)
assert_equal(f.id, file2.id)
assert f.id == file2.id
c = h.execute_query("FIND FILE WHICH IS STORED AT *.dat")
assert_equal(len(c), 8)
c = db.execute_query("FIND FILE WHICH IS STORED AT *.dat")
assert len(c) == 8
c = h.execute_query("FIND FILE WHICH IS STORED AT *test*.dat")
assert_equal(len(c), 8)
c = db.execute_query("FIND FILE WHICH IS STORED AT *test*.dat")
assert len(c) == 8
c = h.execute_query("FIND FILE WHICH IS STORED AT *test%*.dat")
assert_equal(len(c), 2)
assert_is_not_none(c.get_entity_by_id(file7.id))
assert_is_not_none(c.get_entity_by_id(file8.id))
c = db.execute_query("FIND FILE WHICH IS STORED AT *test%*.dat")
assert len(c) == 2
assert c.get_entity_by_id(file7.id) is not None
assert c.get_entity_by_id(file8.id) is not None
f = h.execute_query(
f = db.execute_query(
r"FIND FILE WHICH IS STORED AT *test\**.dat",
unique=True)
assert_equal(f.id, file6.id)
assert f.id == file6.id
f = h.execute_query(
f = db.execute_query(
r"FIND FILE WHICH IS STORED AT /rootdir/subdir1/subdir2/**",
unique=True)
assert_equal(f.id, file4.id)
assert f.id == file4.id
f = h.execute_query(
f = db.execute_query(
r"FIND FILE WHICH IS STORED AT '/rootdir/subdir1/subdir2/**'",
unique=True)
assert_equal(f.id, file4.id)
assert f.id == file4.id
c = h.execute_query(
c = db.execute_query(
r"FIND FILE WHICH IS STORED AT /rootdir/subdir1/**")
assert len(c) == 4
assert c.get_entity_by_id(file3.id) is not None
......@@ -839,7 +804,7 @@ def test_stored_at_wildcards():
assert c.get_entity_by_id(file7.id) is not None
assert c.get_entity_by_id(file8.id) is not None
c = h.execute_query(
c = db.execute_query(
r"FIND FILE WHICH IS STORED AT '/rootdir/subdir1/**'")
assert len(c) == 4
assert c.get_entity_by_id(file3.id) is not None
......@@ -847,7 +812,7 @@ def test_stored_at_wildcards():
assert c.get_entity_by_id(file7.id) is not None
assert c.get_entity_by_id(file8.id) is not None
c = h.execute_query(
c = db.execute_query(
r"FIND FILE WHICH IS STORED AT /rootdir/**")
assert len(c) == 6
assert c.get_entity_by_id(file2.id) is not None
......@@ -857,7 +822,7 @@ def test_stored_at_wildcards():
assert c.get_entity_by_id(file7.id) is not None
assert c.get_entity_by_id(file8.id) is not None
c = h.execute_query(
c = db.execute_query(
r"FIND FILE WHICH IS STORED AT '/rootdir/**'")
assert len(c) == 6
assert c.get_entity_by_id(file2.id) is not None
......@@ -870,85 +835,91 @@ def test_stored_at_wildcards():
@mark.slow
def test_int():
pint = h.Property(name="TestIntegerProperty", datatype=h.INTEGER).insert()
pdouble = h.Property(name="TestDoubleProperty", datatype=h.DOUBLE).insert()
h.RecordType(
pint = db.Property(
name="TestIntegerProperty",
datatype=db.INTEGER).insert()
pdouble = db.Property(
name="TestDoubleProperty",
datatype=db.DOUBLE).insert()
db.RecordType(
name="TestRecordType").add_property(
pint,
value="100").add_property(
pdouble,
value="100.5").insert()
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty>50",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty>=50",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty>=100",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty<=100",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty<=200",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty<200",
unique=True)
assert_equal(0, len(h.execute_query(
"FIND TestRecordType WITH TestIntegerProperty>200")))
assert_equal(0, len(h.execute_query(
"FIND TestRecordType WITH TestIntegerProperty>=200")))
assert_equal(0, len(h.execute_query(
"FIND TestRecordType WITH TestIntegerProperty=200")))
assert_equal(0, len(h.execute_query(
"FIND TestRecordType WITH TestIntegerProperty<50")))
assert_equal(0, len(h.execute_query(
"FIND TestRecordType WITH TestIntegerProperty<=50")))
assert 0 == len(db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty>200"))
assert 0 == len(db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty>=200"))
assert 0 == len(db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty=200"))
assert 0 == len(db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty<50"))
assert 0 == len(db.execute_query(
"FIND TestRecordType WITH TestIntegerProperty<=50"))
h.execute_query("FIND TestRecordType WITH TestDoubleProperty", unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty",
unique=True)
db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty>50",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty>=50",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty>=100.5",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty>=100.5",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty<=100.5",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty<=200",
unique=True)
h.execute_query(
db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty<200",
unique=True)
assert_equal(0, len(h.execute_query(
"FIND TestRecordType WITH TestDoubleProperty>200")))
assert_equal(0, len(h.execute_query(
"FIND TestRecordType WITH TestDoubleProperty>=200")))
assert_equal(0, len(h.execute_query(
"FIND TestRecordType WITH TestDoubleProperty=200")))
assert_equal(0, len(h.execute_query(
"FIND TestRecordType WITH TestDoubleProperty<50")))
assert_equal(0, len(h.execute_query(
"FIND TestRecordType WITH TestDoubleProperty<=50")))
assert 0 == len(db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty>200"))
assert 0 == len(db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty>=200"))
assert 0 == len(db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty=200"))
assert 0 == len(db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty<50"))
assert 0 == len(db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty<=50"))
def test_query_benchmark():
h.Property("TestProperty", datatype=h.TEXT).insert()
db.Property("TestProperty", datatype=db.TEXT).insert()
body = get_connection().retrieve(
entity_uri_segments=["Entity"],
......@@ -958,70 +929,70 @@ def test_query_benchmark():
print(body)
xml = etree.fromstring(body)
# see #4
#assert_equal(3, len(xml))
#assert_equal("query", xml[0].tag.lower())
#assert_equal("transactionbenchmark", xml[0][3].tag.lower())
#assert 3 == len(xml)
#assert "query" == xml[0].tag.lower()
#assert "transactionbenchmark" == xml[0][3].tag.lower()
#benchmark = xml[0][3]
#assert_true(len(benchmark) > 0)
#assert len(benchmark) > 0
def test_like_query():
""" See https://git.goltzsche.net/caosdb/customers/glaz_awi/ext-awi/issues/1
"""
h.execute_query("FIND box with number like '**'")
h.execute_query("FIND box with number like 'a'")
h.execute_query("FIND box with number like '*7*'")
h.execute_query("FIND box with number like '7*'")
h.execute_query("FIND box with number like '*7'")
h.execute_query("FIND box with number like '7'")
h.execute_query("FIND box with number = '7'")
db.execute_query("FIND box with number like '**'")
db.execute_query("FIND box with number like 'a'")
db.execute_query("FIND box with number like '*7*'")
db.execute_query("FIND box with number like '7*'")
db.execute_query("FIND box with number like '*7'")
db.execute_query("FIND box with number like '7'")
db.execute_query("FIND box with number = '7'")
def test_backref_like():
h.RecordType("TestData").insert()
other = h.Record("other").add_parent("TestData").insert()
sme = h.Record("s_m_e").add_property(
db.RecordType("TestData").insert()
other = db.Record("other").add_parent("TestData").insert()
sme = db.Record("s_m_e").add_property(
"TestData", other.id).add_parent("TestData").insert()
print(sme)
h.execute_query("FIND ENTITY WHICH IS REFERENCED BY s_m_e", unique=True)
h.execute_query("FIND ENTITY WHICH IS REFERENCED BY *m*", unique=True)
h.execute_query("FIND ENTITY WHICH IS REFERENCED BY s*", unique=True)
h.execute_query("FIND ENTITY WHICH IS REFERENCED BY *e", unique=True)
db.execute_query("FIND ENTITY WHICH IS REFERENCED BY s_m_e", unique=True)
db.execute_query("FIND ENTITY WHICH IS REFERENCED BY *m*", unique=True)
db.execute_query("FIND ENTITY WHICH IS REFERENCED BY s*", unique=True)
db.execute_query("FIND ENTITY WHICH IS REFERENCED BY *e", unique=True)
def test_query_by_name():
rt = h.RecordType("TestRT").insert()
rec1 = h.Record("TestRec1").add_parent("TestRT").insert()
rec2 = h.Record("TestRec2").add_parent("TestRT").insert()
rt = db.RecordType("TestRT").insert()
rec1 = db.Record("TestRec1").add_parent("TestRT").insert()
rec2 = db.Record("TestRec2").add_parent("TestRT").insert()
assert len(h.execute_query("FIND TestRT")) == 3
assert len(h.execute_query("FIND RECORD TestRT")) == 2
assert len(h.execute_query("FIND RECORDTYPE TestRT")) == 1
assert len(db.execute_query("FIND TestRT")) == 3
assert len(db.execute_query("FIND RECORD TestRT")) == 2
assert len(db.execute_query("FIND RECORDTYPE TestRT")) == 1
# find rt via name
assert h.execute_query("FIND RECORDTYPE WITH name = 'TestRT'",
unique=True).id == rt.id
assert h.execute_query("FIND RECORDTYPE WITH NAME = 'TestRT'",
unique=True).id == rt.id
assert h.execute_query("FIND RECORDTYPE WITH NAme = 'TestRT'",
unique=True).id == rt.id
assert h.execute_query("FIND TestRT WITH NAme = 'TestRT'",
unique=True).id == rt.id
assert h.execute_query("FIND RECORDTYPE WITH name LIKE Test*",
unique=True).id == rt.id
assert db.execute_query("FIND RECORDTYPE WITH name = 'TestRT'",
unique=True).id == rt.id
assert db.execute_query("FIND RECORDTYPE WITH NAME = 'TestRT'",
unique=True).id == rt.id
assert db.execute_query("FIND RECORDTYPE WITH NAme = 'TestRT'",
unique=True).id == rt.id
assert db.execute_query("FIND TestRT WITH NAme = 'TestRT'",
unique=True).id == rt.id
assert db.execute_query("FIND RECORDTYPE WITH name LIKE Test*",
unique=True).id == rt.id
# find rec1 via name
assert h.execute_query("FIND TestRT WITH name = 'TestRec1'",
unique=True).id == rec1.id
assert h.execute_query("FIND RECORD TestRT WITH name = 'TestRec1'",
unique=True).id == rec1.id
assert h.execute_query("FIND RECORD TestRT WITH name LIKE '*Rec1'",
unique=True).id == rec1.id
assert db.execute_query("FIND TestRT WITH name = 'TestRec1'",
unique=True).id == rec1.id
assert db.execute_query("FIND RECORD TestRT WITH name = 'TestRec1'",
unique=True).id == rec1.id
assert db.execute_query("FIND RECORD TestRT WITH name LIKE '*Rec1'",
unique=True).id == rec1.id
# find both records via name
assert len(h.execute_query("FIND TestRT WITH name LIKE 'TestRec*'")) == 2
assert len(h.execute_query("FIND TestRT WITH name LIKE '*Rec*'")) == 2
assert len(h.execute_query("FIND ENTITY WITH name LIKE 'TestRec*'")) == 2
assert len(db.execute_query("FIND TestRT WITH name LIKE 'TestRec*'")) == 2
assert len(db.execute_query("FIND TestRT WITH name LIKE '*Rec*'")) == 2
assert len(db.execute_query("FIND ENTITY WITH name LIKE 'TestRec*'")) == 2
@mark.xfail(reason="Issue: https://gitlab.com/caosdb/caosdb-server/-/issues/96")
......@@ -1047,10 +1018,10 @@ def test_referenced_as():
def test_query_cache():
entity = h.RecordType("TestRT").insert()
entity = db.RecordType("TestRT").insert()
# fill cache
query = h.Query("COUNT TestRT")
query = db.Query("COUNT TestRT")
assert query.cached is None
response = query.execute()
assert response == 1
......@@ -1059,7 +1030,7 @@ def test_query_cache():
etag = query.etag
# cached == True
query = h.Query("COUNT TestRT")
query = db.Query("COUNT TestRT")
assert query.cached is None
response = query.execute()
assert response == 1
......@@ -1067,7 +1038,7 @@ def test_query_cache():
assert query.etag == etag
# cached == True
query = h.Query("FIND TestRT")
query = db.Query("FIND TestRT")
assert query.cached is None
response = query.execute(unique=True)
assert response.id == entity.id
......@@ -1075,7 +1046,7 @@ def test_query_cache():
assert query.etag == etag
# cached == True
query = h.Query("SELECT bla FROM TestRT")
query = db.Query("SELECT bla FROM TestRT")
assert query.cached is None
response = query.execute(unique=True)
assert response.id == entity.id
......@@ -1083,7 +1054,7 @@ def test_query_cache():
assert query.etag == etag
# no cache flag
query = h.Query("SELECT bla FROM TestRT")
query = db.Query("SELECT bla FROM TestRT")
assert query.cached is None
response = query.execute(unique=True, cache=False)
assert response.id == entity.id
......@@ -1091,7 +1062,7 @@ def test_query_cache():
assert query.etag == etag
# cached == True
query = h.Query("SELECT bla FROM TestRT")
query = db.Query("SELECT bla FROM TestRT")
assert query.cached is None
response = query.execute(unique=True)
assert response.id == entity.id
......@@ -1099,10 +1070,10 @@ def test_query_cache():
assert query.etag == etag
# write resets cache
another_entity = h.Record().add_parent("TestRT").insert()
another_entity = db.Record().add_parent("TestRT").insert()
# cached == False
query = h.Query("COUNT TestRT")
query = db.Query("COUNT TestRT")
assert query.cached is None
response = query.execute()
assert response == 2
......@@ -1111,7 +1082,7 @@ def test_query_cache():
new_etag = query.etag
# cached == True
query = h.Query("COUNT TestRT")
query = db.Query("COUNT TestRT")
assert query.cached is None
response = query.execute()
assert response == 2
......@@ -1120,63 +1091,63 @@ def test_query_cache():
def test_query_cache_with_permissions():
h.RecordType("TestRT").insert()
h.RecordType("TestRT2").insert()
public_record = h.Record().add_parent("TestRT").insert()
referenced = h.Record().add_parent("TestRT2").insert()
private_record = h.Record().add_parent(
db.RecordType("TestRT").insert()
db.RecordType("TestRT2").insert()
public_record = db.Record().add_parent("TestRT").insert()
referenced = db.Record().add_parent("TestRT2").insert()
private_record = db.Record().add_parent(
"TestRT").add_property("TestRT2",
referenced).insert(flags={"ACL": None})
# proof: query as anonymous works in principle
h.configure_connection(password_method="unauthenticated")
query = h.Query("COUNT Record TestRT")
db.configure_connection(password_method="unauthenticated")
query = db.Query("COUNT Record TestRT")
response = query.execute()
assert response == 2
assert query.cached == False
query = h.Query("COUNT Record WHICH IS REFERENCED BY TestRT")
query = db.Query("COUNT Record WHICH IS REFERENCED BY TestRT")
response = query.execute()
assert response == 1
assert query.cached == False
# remove permissions for anonymous
h.configure_connection()
db.configure_connection()
private_record.deny(role="anonymous", permission="RETRIEVE:*")
private_record.update_acl() # note: this clears the cache
# as authenticated user to fill the cache
query = h.Query("COUNT Record TestRT")
query = db.Query("COUNT Record TestRT")
response = query.execute()
assert response == 2
assert query.cached == False
query = h.Query("COUNT Record WHICH IS REFERENCED BY TestRT")
query = db.Query("COUNT Record WHICH IS REFERENCED BY TestRT")
response = query.execute()
assert response == 1
assert query.cached == False
# as anonymous
h.configure_connection(password_method="unauthenticated")
query = h.Query("COUNT Record TestRT")
db.configure_connection(password_method="unauthenticated")
query = db.Query("COUNT Record TestRT")
response = query.execute()
assert response == 1
assert query.cached == True
query = h.Query("COUNT Record WHICH IS REFERENCED BY TestRT")
query = db.Query("COUNT Record WHICH IS REFERENCED BY TestRT")
response = query.execute()
assert response == 0
# Caching was not possible due to complex permissions
assert query.cached == False
# try again as authenticated user
h.configure_connection()
query = h.Query("COUNT Record TestRT")
db.configure_connection()
query = db.Query("COUNT Record TestRT")
response = query.execute()
assert query.cached == True
assert response == 2
query = h.Query("COUNT Record WHICH IS REFERENCED BY TestRT")
query = db.Query("COUNT Record WHICH IS REFERENCED BY TestRT")
response = query.execute()
assert response == 1
# Caching was not possible due to complex permissions
......@@ -1185,33 +1156,33 @@ def test_query_cache_with_permissions():
def test_find_star():
"""related: caosdb-server#116"""
rt = h.RecordType("TestRT").insert()
response = h.execute_query("FIND *")[0]
rt = db.RecordType("TestRT").insert()
response = db.execute_query("FIND *")[0]
assert response.id == rt.id
def test_find_entity():
rt = h.RecordType("TestRT").insert()
response = h.execute_query("FIND ENTITY")[0]
rt = db.RecordType("TestRT").insert()
response = db.execute_query("FIND ENTITY")[0]
assert response.id == rt.id
def test_count_any_version_of_entity():
vcount1 = h.execute_query("COUNT ANY VERSION OF ENTITY")
ecount1 = h.execute_query("COUNT ENTITY")
vcount1 = db.execute_query("COUNT ANY VERSION OF ENTITY")
ecount1 = db.execute_query("COUNT ENTITY")
assert ecount1 == 0
assert vcount1 == 0
rt = h.RecordType("TestRT").insert()
vcount2 = h.execute_query("COUNT ANY VERSION OF ENTITY")
ecount2 = h.execute_query("COUNT ENTITY")
rt = db.RecordType("TestRT").insert()
vcount2 = db.execute_query("COUNT ANY VERSION OF ENTITY")
ecount2 = db.execute_query("COUNT ENTITY")
assert vcount2 == vcount1 + 1
assert ecount2 == ecount1 + 1
rt.description = "update"
rt.update()
vcount3 = h.execute_query("COUNT ANY VERSION OF ENTITY")
ecount3 = h.execute_query("COUNT ENTITY")
vcount3 = db.execute_query("COUNT ANY VERSION OF ENTITY")
ecount3 = db.execute_query("COUNT ENTITY")
assert vcount3 == vcount2 + 1
assert ecount3 == ecount2
......@@ -1219,26 +1190,115 @@ def test_count_any_version_of_entity():
def test_find_nasty_path():
"""related: caosdb-server#31"""
file_1 = store_file("0in.txt")
assert h.execute_query(
assert db.execute_query(
"FIND FILE WHICH IS STORED AT 0in.txt",
unique=True).id == file_1.id
file_2 = store_file(".asdf.txt")
assert h.execute_query(
assert db.execute_query(
"FIND FILE WHICH IS STORED AT .asdf.txt",
unique=True).id == file_2.id
file_3 = store_file(".WITH")
assert h.execute_query(
assert db.execute_query(
"FIND FILE WHICH IS STORED AT .WITH",
unique=True).id == file_3.id
file_4 = store_file("STORED")
assert h.execute_query(
assert db.execute_query(
"FIND FILE WHICH IS STORED AT STORED",
unique=True).id == file_4.id
file_5 = store_file("STORED AT") # hehe
assert h.execute_query(
assert db.execute_query(
"FIND FILE WHICH IS STORED AT STORED AT",
unique=True).id == file_5.id
def test_transaction_time():
rt = db.RecordType("TestRT").insert(flags={"H": None})
date = rt.version.date
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED AT " + date,
unique=True).id
with raises(EmptyUniqueQueryError):
db.execute_query("FIND TestRT WHICH WAS INSERTED AFTER " + date,
unique=True)
with raises(EmptyUniqueQueryError):
db.execute_query("FIND TestRT WHICH WAS INSERTED BEFORE " + date,
unique=True)
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED SINCE " + date,
unique=True).id
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED UNTIL " + date,
unique=True).id
with raises(EmptyUniqueQueryError):
db.execute_query("FIND TestRT WHICH WAS INSERTED AFTER TODAY",
unique=True)
with raises(EmptyUniqueQueryError):
db.execute_query("FIND TestRT WHICH WAS INSERTED BEFORE TODAY",
unique=True)
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED SINCE TODAY",
unique=True).id
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED UNTIL TODAY",
unique=True).id
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED TODAY",
unique=True).id
year = date[:4]
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED IN " + year,
unique=True).id
with raises(EmptyUniqueQueryError):
db.execute_query("FIND TestRT WHICH WAS INSERTED AFTER " + year,
unique=True)
with raises(EmptyUniqueQueryError):
db.execute_query("FIND TestRT WHICH WAS INSERTED BEFORE " + year,
unique=True)
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED SINCE " + year,
unique=True).id
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED UNTIL " + year,
unique=True).id
old_year = "2020"
with raises(EmptyUniqueQueryError):
db.execute_query(
"FIND TestRT WHICH WAS INSERTED IN " + old_year,
unique=True).id
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED AFTER " + old_year,
unique=True).id
with raises(EmptyUniqueQueryError):
db.execute_query("FIND TestRT WHICH WAS INSERTED BEFORE " + old_year,
unique=True)
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED SINCE " + old_year,
unique=True).id
with raises(EmptyUniqueQueryError):
db.execute_query("FIND TestRT WHICH WAS INSERTED UNTIL " + old_year,
unique=True)
next_year = str(int(year) + 1)
with raises(EmptyUniqueQueryError):
db.execute_query(
"FIND TestRT WHICH WAS INSERTED IN " + next_year,
unique=True).id
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED BEFORE " + next_year,
unique=True).id
with raises(EmptyUniqueQueryError):
db.execute_query("FIND TestRT WHICH WAS INSERTED AFTER " + next_year,
unique=True)
assert rt.id == db.execute_query(
"FIND TestRT WHICH WAS INSERTED UNTIL " + next_year,
unique=True).id
with raises(EmptyUniqueQueryError):
db.execute_query("FIND TestRT WHICH WAS INSERTED SINCE " + next_year,
unique=True)