test_cached.py

# -*- 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