mytec
defa3ad440
Major refactoring of RFCP backend: - Modular propagation models (8 models) - SharedMemoryManager for terrain data - ProcessPoolExecutor parallel processing - WebSocket progress streaming - Building filtering pipeline (351k → 15k) - 82 unit tests Performance: Standard preset 38s → 5s (7.6x speedup) Known issue: Detailed preset timeout (fix in 3.1.0)
127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
"""
|
|
Unit tests for the unified cache service.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import time
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
|
|
|
|
from app.services.cache import MemoryCache, CacheManager
|
|
|
|
|
|
class TestMemoryCache:
|
|
def test_get_miss(self):
|
|
cache = MemoryCache("test", max_entries=10)
|
|
assert cache.get("nonexistent") is None
|
|
|
|
def test_put_get(self):
|
|
cache = MemoryCache("test", max_entries=10)
|
|
cache.put("key1", "value1", size_bytes=100)
|
|
assert cache.get("key1") == "value1"
|
|
|
|
def test_overwrite(self):
|
|
cache = MemoryCache("test", max_entries=10)
|
|
cache.put("key1", "v1", size_bytes=100)
|
|
cache.put("key1", "v2", size_bytes=200)
|
|
assert cache.get("key1") == "v2"
|
|
assert cache.size == 1
|
|
assert cache.size_bytes == 200
|
|
|
|
def test_eviction_by_entries(self):
|
|
cache = MemoryCache("test", max_entries=3)
|
|
cache.put("a", 1)
|
|
cache.put("b", 2)
|
|
cache.put("c", 3)
|
|
assert cache.size == 3
|
|
cache.put("d", 4) # Should evict 'a' (LRU)
|
|
assert cache.size == 3
|
|
assert cache.get("a") is None
|
|
assert cache.get("d") == 4
|
|
|
|
def test_eviction_by_size(self):
|
|
cache = MemoryCache("test", max_entries=100, max_size_bytes=300)
|
|
cache.put("a", 1, size_bytes=100)
|
|
cache.put("b", 2, size_bytes=100)
|
|
cache.put("c", 3, size_bytes=100)
|
|
assert cache.size_bytes == 300
|
|
cache.put("d", 4, size_bytes=100) # Should evict 'a'
|
|
assert cache.size_bytes == 300
|
|
assert cache.get("a") is None
|
|
|
|
def test_lru_access_order(self):
|
|
cache = MemoryCache("test", max_entries=3)
|
|
cache.put("a", 1)
|
|
cache.put("b", 2)
|
|
cache.put("c", 3)
|
|
# Access 'a' to make it recently used
|
|
cache.get("a")
|
|
# Add 'd' — should evict 'b' (now LRU)
|
|
cache.put("d", 4)
|
|
assert cache.get("a") == 1 # Still there
|
|
assert cache.get("b") is None # Evicted
|
|
|
|
def test_remove(self):
|
|
cache = MemoryCache("test", max_entries=10)
|
|
cache.put("key1", "val", size_bytes=50)
|
|
assert cache.remove("key1") is True
|
|
assert cache.get("key1") is None
|
|
assert cache.size_bytes == 0
|
|
|
|
def test_clear(self):
|
|
cache = MemoryCache("test", max_entries=10)
|
|
cache.put("a", 1, size_bytes=100)
|
|
cache.put("b", 2, size_bytes=100)
|
|
cache.clear()
|
|
assert cache.size == 0
|
|
assert cache.size_bytes == 0
|
|
|
|
def test_stats(self):
|
|
cache = MemoryCache("test", max_entries=10, max_size_bytes=1024)
|
|
cache.put("a", 1, size_bytes=100)
|
|
cache.get("a") # hit
|
|
cache.get("b") # miss
|
|
s = cache.stats()
|
|
assert s["name"] == "test"
|
|
assert s["entries"] == 1
|
|
assert s["hits"] == 1
|
|
assert s["misses"] == 1
|
|
assert s["hit_rate"] == 50.0
|
|
|
|
|
|
class TestCacheManager:
|
|
def test_singleton_structure(self):
|
|
mgr = CacheManager()
|
|
assert mgr.terrain is not None
|
|
assert mgr.buildings is not None
|
|
assert mgr.spatial is not None
|
|
assert mgr.osm_disk is not None
|
|
|
|
def test_stats(self):
|
|
mgr = CacheManager()
|
|
s = mgr.stats()
|
|
assert "terrain" in s
|
|
assert "buildings" in s
|
|
assert "total_memory_mb" in s
|
|
|
|
def test_clear_all(self):
|
|
mgr = CacheManager()
|
|
mgr.terrain.put("test", "data", 100)
|
|
mgr.buildings.put("test", "data", 100)
|
|
mgr.clear_all()
|
|
assert mgr.terrain.size == 0
|
|
assert mgr.buildings.size == 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
for cls_name, cls in [("MemoryCache", TestMemoryCache), ("CacheManager", TestCacheManager)]:
|
|
instance = cls()
|
|
for method_name in [m for m in dir(instance) if m.startswith("test_")]:
|
|
try:
|
|
getattr(instance, method_name)()
|
|
print(f" PASS {cls_name}.{method_name}")
|
|
except Exception as e:
|
|
print(f" FAIL {cls_name}.{method_name}: {e}")
|
|
print("\nAll tests completed.")
|