TST: more tests for query cache and "find entity" queries

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