Tests for caosdb-server!77

    Depends on caosdb-pylib!51 (merged)

tests/test_query.py

@@ -35,6 +35,10 @@ from caosdb.connection.connection import get_connection
 from lxml import etree
 
 
+def setup_module():
+    h.administration.set_server_property("AUTH_OPTIONAL", "TRUE")
+
+
 def setup():
     try:
         h.execute_query("FIND Test*").delete()
@@ -43,6 +47,7 @@ def setup():
 
 
 def teardown():
+    h.configure_connection()
     setup()
     try:
         import os
@@ -1024,3 +1029,106 @@ def test_referenced_as():
                               "TestParty AS Guests")
     assert len(guests) == 3
     assert set([e.id for e in guests]) == guest_ids
+
+
+def test_query_cache():
+    entity = h.RecordType("TestRT").insert()
+
+    # fill cache
+    query = h.Query("COUNT TestRT")
+    assert query.cached is None
+    response = query.execute()
+    assert response == 1
+    assert query.cached == False
+
+    # cached == True
+    query = h.Query("COUNT TestRT")
+    assert query.cached is None
+    response = query.execute()
+    assert response == 1
+    assert query.cached == True
+
+    # cached == True
+    query = h.Query("FIND TestRT")
+    assert query.cached is None
+    response = query.execute(unique=True)
+    assert response.id == entity.id
+    assert query.cached == True
+
+    # cached == True
+    query = h.Query("SELECT bla FROM TestRT")
+    assert query.cached is None
+    response = query.execute(unique=True)
+    assert response.id == entity.id
+    assert query.cached == True
+
+    # no cache flag
+    query = h.Query("SELECT bla FROM TestRT")
+    assert query.cached is None
+    response = query.execute(unique=True, cache=False)
+    assert response.id == entity.id
+    assert query.cached == False
+
+    # cached == True
+    query = h.Query("SELECT bla FROM TestRT")
+    assert query.cached is None
+    response = query.execute(unique=True)
+    assert response.id == entity.id
+    assert query.cached == True
+
+    # write resets cache
+    another_entity = h.Record().add_parent("TestRT").insert()
+
+    # cached == False
+    query = h.Query("COUNT TestRT")
+    assert query.cached is None
+    response = query.execute()
+    assert response == 2
+    assert query.cached == False
+
+    # cached == True
+    query = h.Query("COUNT TestRT")
+    assert query.cached is None
+    response = query.execute()
+    assert response == 2
+    assert query.cached == True
+
+
+def test_query_cache_with_permissions():
+    h.RecordType("TestRT").insert()
+    h.RecordType("TestRT2").insert()
+    public_record = h.Record().add_parent("TestRT").insert()
+    private_record = h.Record().add_parent(
+        "TestRT").insert(flags={"ACL": None})
+
+    # proof: query as anonymous works in principle
+    h.configure_connection(password_method="unauthenticated")
+    query = h.Query("COUNT Record TestRT")
+    response = query.execute()
+    assert response == 2
+    assert query.cached == False
+
+    # remove permissions for anonymous
+    h.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")
+    response = query.execute()
+    assert response == 2
+    assert query.cached == False
+
+    # as anonymous
+    h.configure_connection(password_method="unauthenticated")
+    query = h.Query("COUNT Record TestRT")
+    response = query.execute()
+    assert response == 1
+    assert query.cached == True
+
+    # try again as authenticated user
+    h.configure_connection()
+    query = h.Query("COUNT Record TestRT")
+    response = query.execute()
+    assert query.cached == True
+    assert response == 2