Select Git revision
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_cached.py 10.63 KiB
# -*- coding: utf-8 -*-
#
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2023 Henrik tom Wörden <h.tomwoerden@indiscale.com>
# Copyright (C) 2023 IndiScale GmbH <info@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 caosdb.cached module """
from caosdb.cached import (cached_get_entity_by, cache_clear, cache_info, cache_fill,
AccessType, cache_initialize, cached_query)
from unittest.mock import patch
import caosdb as db
from copy import deepcopy
import pytest
DUMMY_SERVER_CONTENT = [
db.Record(name='a', id=101),
db.Record(name='b', id=102),
db.Record(name='c', id=103),
db.File(path='p', id=104),
db.File(path='pp', id=105),
]
@pytest.fixture(autouse=True)
def cache_clean_up():
cache_clear()
yield
cache_clear()
def mocked_name_query(name):
# copy the object, because Entities would normally be created from XML response
return deepcopy([el for el in DUMMY_SERVER_CONTENT if el.name == name][0])
def mocked_id_query(eid):
# copy the object, because Entities would normally be created from XML response
return deepcopy([el for el in DUMMY_SERVER_CONTENT if el.id == eid][0])
def mocked_path_query(path):
# copy the object, because Entities would normally be created from XML response
return deepcopy([el for el in DUMMY_SERVER_CONTENT if el.path == path][0])
def mocked_gen_query(q, unique):
if unique:
if q == 'a':
return DUMMY_SERVER_CONTENT[0]
else:
return None
else:
if q == 'a':
return db.Container().extend([DUMMY_SERVER_CONTENT[0]])
else:
return db.Container().extend(DUMMY_SERVER_CONTENT)
@patch("caosdb.utils.get_entity.get_entity_by_name")
def test_get_by_name(mocked_get_by_name):
mocked_get_by_name.side_effect = mocked_name_query
# first call; not in cache -> mocked_execute is touched
a = cached_get_entity_by(name='a')
assert a.id == 101
assert mocked_get_by_name.call_count == 1
# second call; in cache -> mocked_execute is NOT touched (count is still 1)
b = cached_get_entity_by(name='a')
assert mocked_get_by_name.call_count == 1
# the cache returned the same object
assert a is b
# check the info
assert cache_info().hits == 1
assert cache_info().currsize == 1
# after clearing the test, the mock is used again
cache_clear()
cached_get_entity_by(name='a')
assert mocked_get_by_name.call_count == 2
# we fill the cache manually and make sure the element is used
cache_fill({'lol': db.Entity(id=10001, name='lol')}, AccessType.NAME, unique=True)
# there are now two elements in the cache: a and lol
assert cache_info().currsize == 2
# we can retrieve the inserted element
lol = cached_get_entity_by(name='lol')
assert lol.id == 10001
# this did not touch the mocked function
assert mocked_get_by_name.call_count == 2
# make sure normal retrieval still works (count +1)
c = cached_get_entity_by(name='c')
assert mocked_get_by_name.call_count == 3
assert c.id == 103
@patch("caosdb.utils.get_entity.get_entity_by_id")
def test_get_by_id(mocked_get_by_id):
mocked_get_by_id.side_effect = mocked_id_query
# first call; not in cache -> mocked_execute is touched
b = cached_get_entity_by(eid=102)
assert b.id == 102
assert b.name == 'b'
assert mocked_get_by_id.call_count == 1
# second call; in cache -> mocked_execute is NOT touched (count is still 1)
a = cached_get_entity_by(eid=102)
assert mocked_get_by_id.call_count == 1
# the cache returned the same object
assert a is b
# check the info
assert cache_info().hits == 1
assert cache_info().currsize == 1
# after clearing the test, the mock is used again
cache_clear()
cached_get_entity_by(eid=102)
assert mocked_get_by_id.call_count == 2
# we fill the cache manually and make sure the element is used
cache_fill({10001: db.Entity(id=10001, name='lol')}, AccessType.EID, unique=True)
# there are now two elements in the cache: a and lol
assert cache_info().currsize == 2
# we can retrieve the inserted element
lol = cached_get_entity_by(eid=10001)
assert lol.name == 'lol'
# this did not touch the mocked function
assert mocked_get_by_id.call_count == 2
# make sure normal retrieval still works (count +1)
c = cached_get_entity_by(eid=103)
assert mocked_get_by_id.call_count == 3
assert c.name == 'c'
@patch("caosdb.cached.get_entity.get_entity_by_path")
def test_get_by_path(mocked_get_by_path):
mocked_get_by_path.side_effect = mocked_path_query
# first call; not in cache -> mocked_execute is touched
b = cached_get_entity_by(path='p')
assert b.id == 104
assert mocked_get_by_path.call_count == 1
# second call; in cache -> mocked_execute is NOT touched (count is still 1)
a = cached_get_entity_by(path='p')
assert mocked_get_by_path.call_count == 1
# the cache returned the same object
assert a is b
# check the info
assert cache_info().hits == 1
assert cache_info().currsize == 1
# after clearing the test, the mock is used again
cache_clear()
cached_get_entity_by(path='p')
assert mocked_get_by_path.call_count == 2
# we fill the cache manually and make sure the element is used
cache_fill({'lol': db.File(id=10001, path='lol')}, AccessType.PATH, unique=True)
# there are now two elements in the cache: a and lol
assert cache_info().currsize == 2
# we can retrieve the inserted element
lol = cached_get_entity_by(path='lol')
assert lol.id == 10001
# this did not touch the mocked function
assert mocked_get_by_path.call_count == 2
# make sure normal retrieval still works (count +1)
c = cached_get_entity_by(path='pp')
assert mocked_get_by_path.call_count == 3
assert c.id == 105
@patch("caosdb.cached.execute_query")
def test_get_by_query(mocked_query):
mocked_query.side_effect = mocked_gen_query
# test cache initialization
cache_initialize(maxsize=10)
assert cache_info().currsize == 0
# Non-existent entity
res = cached_get_entity_by(query='stuff')
assert res is None
assert cache_info().currsize == 1
assert cache_info().hits == 0
assert cache_info().misses == 1
res = cached_get_entity_by(query='stuff')
assert res is None
assert cache_info().currsize == 1
assert cache_info().hits == 1
assert cache_info().misses == 1
# Existent entity
a = cached_get_entity_by(query='a')
assert a is not None
assert a.id == 101
assert cache_info().currsize == 2
assert cache_info().hits == 1
assert cache_info().misses == 2
@patch("caosdb.cached.execute_query")
def test_cached_query(mocked_query):
mocked_query.side_effect = mocked_gen_query
# test cache initialization
cache_initialize(maxsize=10)
assert cache_info().maxsize == 10
# first call; not in cache -> mocked_execute is touched
res = cached_query('stuff')
assert len(res) == len(DUMMY_SERVER_CONTENT)
assert mocked_query.call_count == 1
# second call; in cache -> mocked_execute is NOT touched (count is still 1)
a = cached_query('stuff')
assert mocked_query.call_count == 1
# the cache returned the same object
assert a is res
# check the info
assert cache_info().hits == 1
assert cache_info().currsize == 1
# after clearing the test, the mock is used again
cache_clear()
cached_query('stuff')
assert mocked_query.call_count == 2
# we fill the cache manually and make sure the element is used
cache_fill({'lol': db.Container().extend([db.Entity(id=10001, name='lol')])},
AccessType.QUERY, unique=False)
# there are now two elements in the cache: a and lol
assert cache_info().currsize == 2
# we can retrieve the inserted element
lol = cached_query('lol')
assert lol[0].id == 10001
# this did not touch the mocked function
assert mocked_query.call_count == 2
# make sure normal retrieval still works (count +1)
c = cached_query('a')
assert mocked_query.call_count == 3
assert c[0].id == 101
@patch("caosdb.utils.get_entity.get_entity_by_name")
def test_cache_size(mocked_get_by_name):
mocked_get_by_name.side_effect = lambda x: x
# first call; not in cache -> mocked_execute is touched
maxsize = 5
cache_initialize(maxsize=maxsize)
assert cache_info().currsize == 0
names_first = ("a", "b", "c", "d", "e")
names_later = ("A", "B", "C", "D", "E")
names_fill = {"X": None, "Y": None, "Z": None}
# Use the first batch of names
for ii, name in enumerate(names_first, start=1):
cached_get_entity_by(name=name)
assert cache_info().currsize == ii
assert cache_info().hits == 0
assert cache_info().misses == ii
for ii, name in enumerate(names_first, start=1):
cached_get_entity_by(name=name)
assert cache_info().currsize == maxsize
assert cache_info().hits == ii
assert cache_info().misses == maxsize
# use the second batch of names
for ii, name in enumerate(names_later, start=1):
cached_get_entity_by(name=name)
assert cache_info().currsize == maxsize
assert cache_info().hits == len(names_first)
assert cache_info().misses == len(names_first) + ii
for ii, name in enumerate(names_later, start=1):
cached_get_entity_by(name=name)
assert cache_info().currsize == maxsize
assert cache_info().hits == len(names_first) + ii
assert cache_info().misses == len(names_first) + len(names_later)
# The cache is now filled with A,B,C,D,E (oldest to least recently used).
# Let's fill it with X,Y,Z.
cache_fill(names_fill, kind=AccessType.NAME)
# Now, the cache should be: D,E,X,Y,Z
current_misses = cache_info().misses
for name in ("Z", "Y", "X", "E", "D"):
cached_get_entity_by(name=name)
assert cache_info().misses == current_misses
for ii, name in enumerate(("A", "B", "C"), start=1):
cached_get_entity_by(name=name)
assert cache_info().misses == current_misses + ii