Cleanup LRUCache, fix bug, add unit tests (pytest).
This commit is contained in:
parent
4e3c0e2b98
commit
46cfc831ff
|
@ -2,16 +2,43 @@ from time import monotonic
|
||||||
|
|
||||||
|
|
||||||
class LRUCache:
|
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):
|
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.open = open
|
||||||
self.capacity = capacity
|
self.capacity = capacity
|
||||||
self.maxage = maxage
|
self.maxage = maxage
|
||||||
self.cache = [] # Each item is a tuple: (key, handle, timestamp), recent items first
|
self.cache = [] # Each item is a tuple: (key, handle, timestamp), recent items first
|
||||||
|
|
||||||
def __contains__(self, key):
|
def __contains__(self, key):
|
||||||
|
"""Check if key is in cache."""
|
||||||
return any(rec[0] == key for rec in self.cache)
|
return any(rec[0] == key for rec in self.cache)
|
||||||
|
|
||||||
def __getitem__(self, key):
|
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
|
# Take from cache or open a new one
|
||||||
for i, (k, f, ts) in enumerate(self.cache):
|
for i, (k, f, ts) in enumerate(self.cache):
|
||||||
if k == key:
|
if k == key:
|
||||||
|
@ -20,15 +47,22 @@ class LRUCache:
|
||||||
else:
|
else:
|
||||||
f = self.open(key)
|
f = self.open(key)
|
||||||
# Add/restore to end of cache
|
# Add/restore to end of cache
|
||||||
self.cache.append((key, f, monotonic()))
|
self.cache.insert(0, (key, f, monotonic()))
|
||||||
self.expire_items()
|
self.expire_items()
|
||||||
|
print(self.cache)
|
||||||
return f
|
return f
|
||||||
|
|
||||||
def expire_items(self):
|
def expire_items(self):
|
||||||
|
"""
|
||||||
|
Expire items that are either too old or exceed cache capacity.
|
||||||
|
"""
|
||||||
ts = monotonic() - self.maxage
|
ts = monotonic() - self.maxage
|
||||||
while len(self.cache) > self.capacity or self.cache and self.cache[-1][2] < ts:
|
while len(self.cache) > self.capacity or self.cache and self.cache[-1][2] < ts:
|
||||||
self.cache.pop()[1].close()
|
self.cache.pop()[1].close()
|
||||||
|
|
||||||
def close(self):
|
def close(self):
|
||||||
|
"""
|
||||||
|
Close the cache and remove all items.
|
||||||
|
"""
|
||||||
self.capacity = 0
|
self.capacity = 0
|
||||||
self.expire_items()
|
self.expire_items()
|
||||||
|
|
55
tests/test_lrucache.py
Normal file
55
tests/test_lrucache.py
Normal file
|
@ -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()
|
Loading…
Reference in New Issue
Block a user