ENH: Integration tests for client-side caching.

tests/test_pylib_caching.py

+# This file is a part of the CaosDB Project.
+#
+# Copyright (C) 2023 IndiScale GmbH <info@indiscale.com>
+# Copyright (C) 2023 Daniel Hornung <d.hornung@indiscale.com>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+"""Test the Python-side caching of server queries."""
+
+import os
+from datetime import datetime
+
+import caosdb as db
+from caosdb.cached import cached_query, cached_get_entity_by, cache_clear
+
+
+def setup_function():
+    d = db.execute_query("FIND Entity WITH ID > 99")
+    if len(d) > 0:
+        d.delete()
+    cache_clear()
+
+    with open("test.dat", "w", encoding="utf-8") as upload_file:
+        upload_file.write("hello world\n")
+
+
+def teardown_function():
+    setup_function()
+
+    try:
+        os.remove("test.dat")
+    except Exception as e:
+        print(e)
+
+
+def test_caching():
+    """Test if cached functions work at all."""
+
+    rect = db.RecordType(name="RT1").insert()
+    rec = db.Record(name="rec1").add_parent(rect).insert()
+
+    for res in (cached_get_entity_by(eid=rec.id),
+                cached_get_entity_by(name=rec.name),
+                cached_get_entity_by(query="FIND RECORD rec1"),
+                cached_query("FIND RECORD rec1")[0],
+                ):
+        assert res.id == rec.id
+
+    file_ = db.File(name="TestFile",
+                    description="Testfile Desc",
+                    path="testfiles/test.dat",
+                    file="test.dat").insert()
+    res = cached_get_entity_by(path="testfiles/test.dat")
+    assert file_.id == res.id
+
+
+def test_caching_speed():
+    """Test if caching is faster that uncached access."""
+    rect = db.RecordType(name="RT1").insert()
+    db.Record(name="rec1").add_parent(rect).insert()
+
+    # Retrieve once to set up server-side caching.
+    db.execute_query("FIND RECORD rec1")
+
+    # uncached
+    before = datetime.now()
+    cached_query("FIND RECORD rec1")
+    after = datetime.now()
+    time_uncached = after - before
+
+    # cached
+    before = datetime.now()
+    cached_query("FIND RECORD rec1")
+    after = datetime.now()
+    time_cached = after - before
+
+    error_str = f"Cached query was not faster than uncached, {time_uncached} vs. {time_cached}"
+    assert time_cached < time_uncached, error_str

tox.ini

@@ -5,6 +5,9 @@ skip_missing_interpreters = true
 [pycodestyle]
 max_line_length = 100
 
+[flake8]
+max-line-length=100
+
 [testenv]
 setenv = PASSWORD_STORE_DIR = {env:HOME}/.password-store
 passenv = PYCAOSDBINI