Skip to content
Snippets Groups Projects
Commit b5f011c0 authored by Timm Fitschen's avatar Timm Fitschen Committed by Florian Spreckelsen
Browse files

TST: Add tests for after/until/before/since keywords

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