# -*- 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, fill_cache, 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): 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 fill_cache({'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 fill_cache({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 fill_cache({'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_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