| 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) |
|
|
