kenyonxu
47193575d6
- Cache manager with JSON read/write, TTL, corruption handling - Civitai client with SHA256 hash (sync+async) and API fetch - KaiCivitaiLoRALoader node with trigger words and JSON output - Node registration, HTTP routes, and secure image serving - Frontend extension with image preview, pagination, and NSFW blur - 49 unit tests covering all modules
108 lines
4.3 KiB
Python
108 lines
4.3 KiB
Python
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
from cache_manager import CacheManager
|
|
|
|
|
|
class TestCacheManager:
|
|
@pytest.fixture
|
|
def tmp_cache_dir(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield tmpdir
|
|
|
|
@pytest.fixture
|
|
def manager(self, tmp_cache_dir):
|
|
return CacheManager(tmp_cache_dir)
|
|
|
|
def test_read_cache_missing_returns_none(self, manager):
|
|
result = manager.read_cache("nonexistent")
|
|
assert result is None
|
|
|
|
def test_write_and_read_cache_roundtrip(self, manager):
|
|
data = {"hash": "abc123", "name": "Test Model", "triggerWords": ["style"]}
|
|
manager.write_cache("abc123", data)
|
|
result = manager.read_cache("abc123")
|
|
assert result == data
|
|
|
|
def test_read_corrupted_cache_auto_deletes(self, manager):
|
|
cache_path = os.path.join(manager.cache_dir, "bad.json")
|
|
os.makedirs(manager.cache_dir, exist_ok=True)
|
|
with open(cache_path, "w") as f:
|
|
f.write("not json{")
|
|
result = manager.read_cache("bad")
|
|
assert result is None
|
|
assert not os.path.exists(cache_path)
|
|
|
|
def test_is_stale_expired(self, manager):
|
|
old_time = time.time() - (8 * 24 * 3600) # 8 days ago
|
|
data = {"fetched_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(old_time))}
|
|
assert manager.is_stale(data) is True
|
|
|
|
def test_is_stale_fresh(self, manager):
|
|
recent_time = time.time() - (1 * 24 * 3600) # 1 day ago
|
|
data = {"fetched_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(recent_time))}
|
|
assert manager.is_stale(data) is False
|
|
|
|
def test_is_stale_no_fetched_at(self, manager):
|
|
data = {"name": "No timestamp"}
|
|
assert manager.is_stale(data) is True
|
|
|
|
def test_is_stale_naive_datetime(self, manager):
|
|
old_time = time.time() - (8 * 24 * 3600)
|
|
data = {"fetched_at": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(old_time))}
|
|
assert manager.is_stale(data) is True
|
|
|
|
def test_clear_cache_deletes_files(self, manager):
|
|
data = {"hash": "xyz"}
|
|
manager.write_cache("xyz", data)
|
|
manager.clear_cache("xyz")
|
|
assert manager.read_cache("xyz") is None
|
|
|
|
def test_get_image_path(self, manager):
|
|
path = manager.get_image_path("abc123", 0)
|
|
assert path.endswith("abc123/preview_0.jpg")
|
|
assert os.path.dirname(path).endswith("abc123")
|
|
|
|
def test_clear_cache_deletes_images(self, manager):
|
|
data = {"hash": "xyz"}
|
|
manager.write_cache("xyz", data)
|
|
# Create a fake image file in the image dir
|
|
img_dir = os.path.join(manager.cache_dir, "xyz")
|
|
os.makedirs(img_dir, exist_ok=True)
|
|
img_path = os.path.join(img_dir, "preview_0.jpg")
|
|
with open(img_path, "w") as f:
|
|
f.write("fake image")
|
|
|
|
manager.clear_cache("xyz")
|
|
assert manager.read_cache("xyz") is None
|
|
assert not os.path.exists(img_path)
|
|
assert not os.path.exists(img_dir)
|
|
|
|
def test_get_cache_for_file_sync_uses_hash(self, manager):
|
|
mock_hash_fn = MagicMock(return_value="mockhash")
|
|
with patch("cache_manager._import_compute_file_hash_sync", return_value=mock_hash_fn):
|
|
with patch.object(manager, "read_cache", return_value={"name": "Cached", "hash": "existing"}):
|
|
result = manager.get_cache_for_file_sync("/fake/path/lora.safetensors")
|
|
assert result == {"name": "Cached", "hash": "existing"}
|
|
|
|
def test_get_cache_for_file_sync_injects_hash(self, manager):
|
|
mock_hash_fn = MagicMock(return_value="injected")
|
|
with patch("cache_manager._import_compute_file_hash_sync", return_value=mock_hash_fn):
|
|
with patch.object(manager, "read_cache", return_value={"name": "Cached"}):
|
|
result = manager.get_cache_for_file_sync("/fake/path/lora.safetensors")
|
|
assert result["hash"] == "injected"
|
|
|
|
def test_get_cache_for_file_sync_file_not_found(self, manager):
|
|
mock_hash_fn = MagicMock(side_effect=FileNotFoundError())
|
|
with patch("cache_manager._import_compute_file_hash_sync", return_value=mock_hash_fn):
|
|
result = manager.get_cache_for_file_sync("/missing/file.safetensors")
|
|
assert result is None
|