diff --git a/cista/lrucache.py b/cista/lrucache.py index 18ebe66..524d757 100644 --- a/cista/lrucache.py +++ b/cista/lrucache.py @@ -2,16 +2,43 @@ from time import monotonic class LRUCache: + """ + LRUCache is a least-recently-used (LRU) cache with expiry time. + + Attributes: + open (callable): Function to open a new handle. + capacity (int): Max number of items in the cache. + maxage (float): Max age for items in cache in seconds. + cache (list): Internal list storing the cache items. + """ def __init__(self, open: callable, *, capacity: int, maxage: float): + """ + Initialize LRUCache. + + Args: + open (callable): Function to open a new handle. + capacity (int): Maximum capacity of the cache. + maxage (float): Max age for items in cache in seconds. + """ self.open = open self.capacity = capacity self.maxage = maxage self.cache = [] # Each item is a tuple: (key, handle, timestamp), recent items first def __contains__(self, key): + """Check if key is in cache.""" return any(rec[0] == key for rec in self.cache) def __getitem__(self, key): + """ + Retrieve an item by its key. + + Args: + key: The key to retrieve. + + Returns: + The corresponding item's handle. + """ # Take from cache or open a new one for i, (k, f, ts) in enumerate(self.cache): if k == key: @@ -20,15 +47,22 @@ class LRUCache: else: f = self.open(key) # Add/restore to end of cache - self.cache.append((key, f, monotonic())) + self.cache.insert(0, (key, f, monotonic())) self.expire_items() + print(self.cache) return f def expire_items(self): + """ + Expire items that are either too old or exceed cache capacity. + """ ts = monotonic() - self.maxage while len(self.cache) > self.capacity or self.cache and self.cache[-1][2] < ts: self.cache.pop()[1].close() def close(self): + """ + Close the cache and remove all items. + """ self.capacity = 0 self.expire_items() diff --git a/tests/test_lrucache.py b/tests/test_lrucache.py new file mode 100644 index 0000000..302e3a5 --- /dev/null +++ b/tests/test_lrucache.py @@ -0,0 +1,55 @@ +import pytest +from unittest.mock import Mock +from time import sleep +from cista.lrucache import LRUCache # Replace with actual import + +def mock_open(key): + mock = Mock() + mock.close = Mock() + mock.content = f"content-{key}" + return mock + +def test_contains(): + cache = LRUCache(open=mock_open, capacity=2, maxage=10) + assert "key1" not in cache + cache["key1"] + assert "key1" in cache + +def test_getitem(): + cache = LRUCache(open=mock_open, capacity=2, maxage=10) + assert cache["key1"].content == "content-key1" + +def test_capacity(): + cache = LRUCache(open=mock_open, capacity=2, maxage=10) + item1 = cache["key1"] + item2 = cache["key2"] + cache["key3"] + assert "key1" not in cache + item1.close.assert_called() + +def test_expiry(): + cache = LRUCache(open=mock_open, capacity=2, maxage=0.1) + item = cache["key1"] + sleep(0.2) # Wait for expiration + cache.expire_items() + assert "key1" not in cache + item.close.assert_called() + +def test_close(): + cache = LRUCache(open=mock_open, capacity=2, maxage=10) + item = cache["key1"] + cache.close() + assert "key1" not in cache + item.close.assert_called() + +def test_lru_mechanism(): + cache = LRUCache(open=mock_open, capacity=2, maxage=10) + item1 = cache["key1"] + item2 = cache["key2"] + cache["key1"] # Make key1 recently used + cache["key3"] # This should remove key2 + assert "key1" in cache + assert "key2" not in cache + assert "key3" in cache + item2.close.assert_called() + item1.close.assert_not_called()