diff --git a/tests/test_query.py b/tests/test_query.py
index a2974434eda6f3673777dcda900be43e18fc6235..4e98803206daa4b9f1ca28c9fc43a10091ee7732 100644
--- a/tests/test_query.py
+++ b/tests/test_query.py
@@ -27,6 +27,8 @@
"""
+import os
+import random
import caosdb as h
from pytest import mark
@@ -50,7 +52,6 @@ def teardown():
h.configure_connection()
setup()
try:
- import os
os.remove("test.dat")
except Exception as e:
print(e)
@@ -641,40 +642,49 @@ def test_wildcard_values():
assert_equal(0, len(c))
+def store_file(path, name=None, f=None):
+ """insert a file for testing purposes."""
+ tmp_file = None
+ if f is None:
+ tmp_file = open("test.dat", "w")
+ tmp_file.write("hello world\n")
+ tmp_file.close()
+ f = tmp_file
+ 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_.insert()
+
+ if tmp_file is not None:
+ os.remove("test.dat")
+ return file_
+
+
def test_stored_at_wildcards():
upload_file = open("test.dat", "w")
upload_file.write("hello world\n")
upload_file.close()
- global i
- i = 1
-
- def store_file(path):
- global i
- file_ = h.File(name="TestTestFile" + str(i),
- description="Testfile Desc",
- path=path,
- file="test.dat")
- i += 1
- file_.insert()
- return file_
+ file1 = store_file("test1.dat", f=upload_file)
- file1 = store_file("test1.dat")
+ file2 = store_file("/rootdir/test2.dat", f=upload_file)
- file2 = store_file("/rootdir/test2.dat")
+ file3 = store_file("/rootdir/subdir1/test3.dat", f=upload_file)
- file3 = store_file("/rootdir/subdir1/test3.dat")
+ file4 = store_file("/rootdir/subdir1/subdir2/test4.dat", f=upload_file)
- file4 = store_file("/rootdir/subdir1/subdir2/test4.dat")
+ file5 = store_file("test5.dat", f=upload_file)
- file5 = store_file("test5.dat")
+ file6 = store_file("rootdir/test*6.dat", f=upload_file)
- file6 = store_file("rootdir/test*6.dat")
+ file7 = store_file("rootdir/subdir1/test%7.dat", f=upload_file)
- file7 = store_file("rootdir/subdir1/test%7.dat")
-
- file8 = store_file("rootdir/subdir1/test%8.dat")
+ 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)
@@ -1098,8 +1108,10 @@ 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(
- "TestRT").insert(flags={"ACL": None})
+ "TestRT").add_property("TestRT2",
+ referenced).insert(flags={"ACL": None})
# proof: query as anonymous works in principle
h.configure_connection(password_method="unauthenticated")
@@ -1108,6 +1120,11 @@ def test_query_cache_with_permissions():
assert response == 2
assert query.cached == False
+ query = h.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()
private_record.deny(role="anonymous", permission="RETRIEVE:*")
@@ -1119,6 +1136,11 @@ def test_query_cache_with_permissions():
assert response == 2
assert query.cached == False
+ query = h.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")
@@ -1126,9 +1148,82 @@ def test_query_cache_with_permissions():
assert response == 1
assert query.cached == True
+ query = h.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")
response = query.execute()
assert query.cached == True
assert response == 2
+
+ query = h.Query("COUNT Record WHICH IS REFERENCED BY TestRT")
+ response = query.execute()
+ assert response == 1
+ # Caching was not possible due to complex permissions
+ assert query.cached == False
+
+
+def test_find_star():
+ """related: caosdb-server#116"""
+ rt = h.RecordType("TestRT").insert()
+ response = h.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]
+ 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")
+ 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")
+ 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")
+ assert vcount3 == vcount2 + 1
+ assert ecount3 == ecount2
+
+
+def test_find_nasty_path():
+ """related: caosdb-server#31"""
+ file_1 = store_file("0in.txt")
+ assert h.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(
+ "FIND FILE WHICH IS STORED AT .asdf.txt",
+ unique=True).id == file_2.id
+
+ file_3 = store_file(".WITH")
+ assert h.execute_query(
+ "FIND FILE WHICH IS STORED AT .WITH",
+ unique=True).id == file_3.id
+
+ file_4 = store_file("STORED")
+ assert h.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(
+ "FIND FILE WHICH IS STORED AT STORED AT",
+ unique=True).id == file_5.id