""" Tests for backend storage implementations. """ from abc import ABC, abstractmethod from collections.abc import Generator, Mapping from contextlib import closing, contextmanager from pathlib import Path import sqlite3 from typing import Any from unittest.mock import MagicMock import pytest from funkycache.storage import ( CacheStorage, MemoryCacheStorage, SQLiteCacheStorage, CannotOverwriteValueException, ) class CacheStorageTestSuite(ABC): """ A common test suite for all implementations of CacheStorage. This should be subclassed as a `test*` class, where you write fixtures and tests that are specific to a single implementation. """ @abstractmethod @contextmanager def create_storage[V]( self, tmp_path: Path, data_type: Any ) -> Generator[CacheStorage[V]]: """ Return an empty cache storage backend. """ @pytest.mark.parametrize( "key", [ pytest.param({"n": "10"}, id="str"), pytest.param({"n": 10}, id="int"), # # A Path isn't serialisable by json.dumps(…) with the # default arguments, but we still want to cache it. pytest.param({"n": Path("10.txt")}, id="path"), # # A dict with Path keys isn't serialisable by json.dumps(…), # even if you pass default=str. pytest.param({Path("n"): "numbers"}, id="path_key"), pytest.param([{Path("n"): "numbers"}], id="path_key_nested_in_list"), pytest.param({"1": {Path("n"): "numbers"}}, id="path_key_nested_in_obj"), ], ) def test_cache(self, key: Mapping[str, Any], tmp_path: Path) -> None: """ Test the basic properties of a cache. """ with self.create_storage(tmp_path, data_type=int) as storage: fib10 = 55 fib20 = 6765 # Test empty storage with pytest.raises(KeyError): storage.get(key) # Store and retrieve a value storage.set(key, value=fib10) assert storage.get(key) == fib10 # Overwriting an existing value is an error. with pytest.raises(CannotOverwriteValueException) as exc: storage.set(key, value=fib20) assert repr(key) in exc.value.args[0] assert storage.get(key) == fib10 # Writing a key with the same value as the already-stored # value is allowed. storage.set(key, value=fib10) def test_can_store_non_ascii_values(self, tmp_path: Path) -> None: """ The storage can hold values which don't encode as ASCII. """ with self.create_storage(tmp_path, data_type=str) as storage: key = {"type": "place"} storage.set(key, value="café") assert storage.get(key) == "café" class TestMemoryCacheStorage(CacheStorageTestSuite): """ Tests for MemoryCacheStorage. """ @contextmanager def create_storage[V]( self, tmp_path: Path, data_type: Any ) -> Generator[CacheStorage[V]]: """ Return an empty cache storage backend. """ yield MemoryCacheStorage[V]() class TestSQLiteCacheStorage(CacheStorageTestSuite): """ Tests for SQLiteCacheStorage. """ @contextmanager def create_storage[V]( self, tmp_path: Path, data_type: Any ) -> Generator[CacheStorage[V]]: """ Return an empty cache storage backend. """ storage = SQLiteCacheStorage(":memory:", data_type=data_type) yield storage storage.close() def test_operations_persist(self, tmp_path: Path) -> None: """ Cache operations persist over opening/closing the cache. """ key = {"n": "one"} db_path = tmp_path / "cache.db" with SQLiteCacheStorage[int](db_path, data_type="int") as cache: cache.set(key, value=1) with SQLiteCacheStorage[int](db_path, data_type="int") as cache: assert cache.get(key) == 1 def test_unexpected_error_during_set(self) -> None: """ If we get an unexpected IntegrityError during set() that isn't about overwriting an existing key, it's re-raised. """ class OtherIntegrityError(sqlite3.IntegrityError): sqlite_errorname = "SQLITE_CONSTRAINT_UNIQUE" with SQLiteCacheStorage[int](":memory:", data_type=int) as cache: # Close the original connection, then replace it with a mock # that simulates a different IntegrityError. cache.con.close() mock_con = MagicMock() mock_con.execute.side_effect = MagicMock( side_effect=OtherIntegrityError("UNIQUE constraint failed") ) cache.con = mock_con with pytest.raises(sqlite3.IntegrityError) as exc_info: cache.set(key={"n": "one"}, value=1) assert exc_info.value.sqlite_errorname == "SQLITE_CONSTRAINT_UNIQUE" def test_reset_if_fingerprint_is_wrong(self, tmp_path: Path) -> None: """ If the database contains a fingerprint from an older version of FunkyCache, old cache entries are dropped. """ key = {"n": "one"} db_path = tmp_path / "cache.db" with SQLiteCacheStorage[int](db_path, data_type="int") as cache: cache.set(key, value=1) with closing(sqlite3.connect(db_path)) as con: cur = con.cursor() cur.execute( "UPDATE funkycache_meta SET value=? WHERE key=?", ("old-fingerprint", "fingerprint"), ) con.commit() with SQLiteCacheStorage[int](db_path, data_type="int") as cache: with pytest.raises(KeyError): cache.get(key) def test_reset_if_fingerprint_is_missing(self, tmp_path: Path) -> None: """ If the database doesn't have a fingerprint from an older version of FunkyCache, old cache entries are dropped. """ key = {"n": "one"} db_path = tmp_path / "cache.db" with SQLiteCacheStorage[int](db_path, data_type="int") as cache: cache.set(key, value=1) with closing(sqlite3.connect(db_path)) as con: cur = con.cursor() cur.execute("DELETE FROM funkycache_meta") con.commit() with SQLiteCacheStorage[int](db_path, data_type="int") as cache: with pytest.raises(KeyError): cache.get(key) def test_error_if_meta_is_wrong_format(self, tmp_path: Path) -> None: """ If looking up the fingerprint funkycache_meta table fails for a reason other than the table missing, the error is re-raised. """ key = {"n": "one"} db_path = tmp_path / "cache.db" with SQLiteCacheStorage[int](db_path, data_type="int") as cache: cache.set(key, value=1) # Delete the original table, then recreate it without a "key" column. # Looking up by key will now fail. with closing(sqlite3.connect(db_path)) as con: cur = con.cursor() cur.execute("DROP TABLE funkycache_meta") cur.execute("CREATE TABLE funkycache_meta(value)") con.commit() with pytest.raises(sqlite3.OperationalError, match="no such column: key"): SQLiteCacheStorage[int](db_path, data_type="int")