File size: 5,925 Bytes
3ab01a7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | import io
import tempfile
from pathlib import Path
from types import SimpleNamespace
import unittest
from unittest import mock
from fastapi import HTTPException
from PIL import Image
from services.image_bytes_cache import ImageBytesCache
from services.image_storage_service import ImageStorageService
class ImageBytesCacheTests(unittest.TestCase):
def test_byte_bound_evicts_least_recently_used(self):
cache = ImageBytesCache(max_bytes=6, max_entries=10)
cache.put("a", b"aaa")
cache.put("b", b"bbb")
self.assertEqual(cache.get("a"), b"aaa")
cache.put("c", b"cc")
self.assertIsNone(cache.get("b"))
self.assertEqual(cache._bytes, 5)
def test_entry_bound(self):
cache = ImageBytesCache(max_bytes=100, max_entries=2)
for key in "abc":
cache.put(key, b"x")
self.assertIsNone(cache.get("a"))
self.assertEqual(len(cache._items), 2)
def test_ttl_does_not_extend_on_read(self):
cache = ImageBytesCache(max_bytes=100, max_entries=2, ttl=5)
with mock.patch("services.image_bytes_cache.time.monotonic", return_value=10):
cache.put("a", b"image")
with mock.patch("services.image_bytes_cache.time.monotonic", return_value=14):
self.assertEqual(cache.get("a"), b"image")
with mock.patch("services.image_bytes_cache.time.monotonic", return_value=15):
self.assertIsNone(cache.get("a"))
self.assertEqual(cache._bytes, 0)
def test_replacement_updates_byte_count(self):
cache = ImageBytesCache(max_bytes=10, max_entries=2)
cache.put("a", b"old")
cache.put("a", b"x")
self.assertEqual(cache._bytes, 1)
self.assertEqual(cache.get("a"), b"x")
cache.discard("a")
cache.discard("a")
self.assertEqual(cache._bytes, 0)
def test_oversize_or_empty_replacement_invalidates_old_value(self):
for payload in (b"", b"too-large"):
cache = ImageBytesCache(max_bytes=4, max_entries=2)
cache.put("a", b"old")
cache.put("a", payload)
self.assertIsNone(cache.get("a"))
self.assertEqual(cache._bytes, 0)
class DurableImageCacheTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name).resolve()
self.images = self.root / "images"
self.cfg = SimpleNamespace(
images_dir=self.images, base_url="http://test",
get_image_storage_settings=lambda: {"mode": "local"},
)
patcher = mock.patch("services.image_storage_service.config", self.cfg)
patcher.start()
self.addCleanup(patcher.stop)
self.store = ImageStorageService(self.root / "index.json")
buffer = io.BytesIO()
Image.new("RGB", (2, 2), "blue").save(buffer, format="PNG")
self.payload = buffer.getvalue()
def test_save_warms_cache_after_durable_write(self):
original = Path.write_bytes
def write(path, payload):
if path.suffix == ".png":
self.assertIsNone(self.store.get_cached_bytes(path.relative_to(self.images).as_posix()))
return original(path, payload)
with mock.patch.object(Path, "write_bytes", write):
stored = self.store.save(self.payload, "http://test")
self.assertEqual((self.images / stored.rel).read_bytes(), self.payload)
self.assertEqual(self.store.get_cached_bytes(stored.rel), self.payload)
def test_failed_durable_write_does_not_warm_cache(self):
with (mock.patch.object(self.store, "make_relative_path", return_value="failed.png"),
mock.patch.object(Path, "write_bytes", side_effect=OSError("disk failed"))):
with self.assertRaises(OSError):
self.store.save(self.payload, "http://test")
self.assertIsNone(self.store.get_cached_bytes("failed.png"))
def test_cache_hit_does_not_resolve_or_access_storage_configuration(self):
self.store._bytes_cache.put("hot.png", self.payload)
with (mock.patch("services.image_storage_service.config", new=None),
mock.patch("services.image_storage_service._local_image_path", side_effect=AssertionError("mount"))):
self.assertEqual(self.store.get_cached_bytes("hot.png"), self.payload)
self.assertEqual(self.store.get_bytes("hot.png"), self.payload)
self.assertIsNone(self.store.get_cached_bytes("cold.png"))
def test_cold_successful_read_warms_cache(self):
self.images.mkdir()
(self.images / "cold.png").write_bytes(self.payload)
self.assertEqual(self.store.get_bytes("cold.png"), self.payload)
with mock.patch.object(Path, "read_bytes", side_effect=AssertionError("second mount read")):
self.assertEqual(self.store.get_bytes("cold.png"), self.payload)
def test_delete_invalidates_original_and_thumbnail(self):
stored = self.store.save(self.payload, "http://test")
self.store.cache_thumbnail(stored.rel, b"thumbnail")
self.assertTrue(self.store.delete(stored.rel))
self.assertIsNone(self.store.get_cached_bytes(stored.rel))
self.assertIsNone(self.store.get_cached_thumbnail(stored.rel))
def test_new_instance_does_not_depend_on_previous_memory_cache(self):
stored = self.store.save(self.payload, "http://test")
restarted = ImageStorageService(self.root / "index.json")
self.assertIsNone(restarted.get_cached_bytes(stored.rel))
self.assertEqual(restarted.get_bytes(stored.rel), self.payload)
def test_traversal_and_non_image_rejected_before_cache(self):
for rel in ("../private.png", "folder/../../private.png", "config.json"):
with self.assertRaises(HTTPException):
self.store.get_cached_bytes(rel)
|