| import asyncio |
| import errno |
| import io |
| import json |
| from threading import Event |
| import unittest |
| from unittest import mock |
|
|
| from fastapi import FastAPI, HTTPException |
| from fastapi.responses import Response |
| import httpx |
| from PIL import Image |
|
|
| from api.errors import install_exception_handlers |
| import api.system as system_module |
| import services.image_service as images |
|
|
|
|
| class ImageReadErrorTests(unittest.IsolatedAsyncioTestCase): |
| def setUp(self): |
| self.store = mock.Mock() |
| self.store.get_bytes.return_value = b"complete-image" |
| self.store.get_cached_bytes.return_value = None |
| self.store.get_cached_thumbnail.return_value = None |
| for module in (images, system_module): |
| patcher = mock.patch.object(module, "image_storage_service", self.store) |
| patcher.start() |
| self.addCleanup(patcher.stop) |
| self.app = FastAPI() |
| install_exception_handlers(self.app) |
| self.app.include_router(system_module.create_router("regression-test")) |
|
|
| @self.app.get("/test/ping") |
| async def ping(): |
| return {"ok": True} |
|
|
| async def get(self, url): |
| async with httpx.AsyncClient(transport=httpx.ASGITransport(app=self.app), base_url="http://test") as client: |
| return await client.get(url) |
|
|
| async def test_original_is_fully_read_before_response_is_created(self): |
| response = images.get_image_response("2026/10/29/image.webp") |
| self.assertIsInstance(response, Response) |
| self.assertEqual(response.body, b"complete-image") |
| self.assertEqual(response.media_type, "image/webp") |
| self.assertIn("immutable", response.headers["cache-control"]) |
| self.store.get_bytes.assert_called_once_with("2026/10/29/image.webp") |
|
|
| async def test_eio_returns_retryable_uncached_503_instead_of_truncated_200(self): |
| self.store.get_bytes.side_effect = OSError(errno.EIO, "private-path-must-not-leak") |
| response = await self.get("/images/2026/10/29/a.png") |
| self.assertEqual(response.status_code, 503) |
| self.assertEqual(response.headers["cache-control"], "no-store") |
| self.assertEqual(response.headers["retry-after"], "10") |
| self.assertNotIn("private-path", response.text) |
|
|
| async def test_missing_image_remains_404(self): |
| self.store.get_bytes.side_effect = FileNotFoundError() |
| self.assertEqual((await self.get("/images/a.png")).status_code, 404) |
|
|
| async def test_unknown_bug_still_propagates(self): |
| self.store.get_bytes.side_effect = RuntimeError("unexpected bug") |
| with self.assertRaisesRegex(RuntimeError, "unexpected bug"): |
| await self.get("/images/a.png") |
|
|
| async def test_thumbnail_bytes_are_read_before_success(self): |
| buffer = io.BytesIO() |
| Image.new("RGB", (640, 480), "blue").save(buffer, format="PNG") |
| self.store.get_bytes.return_value = buffer.getvalue() |
| with mock.patch.object(images, "ensure_thumbnail", side_effect=AssertionError("mount IO")): |
| response = await self.get("/image-thumbnails/a.png") |
| self.assertEqual(response.status_code, 200) |
| with Image.open(io.BytesIO(response.content)) as result: |
| self.assertEqual(result.size, (320, 240)) |
| self.store.cache_thumbnail.assert_called_once_with("a.png", response.content) |
|
|
| async def test_thumbnail_read_eio_is_503(self): |
| self.store.get_bytes.side_effect = OSError(errno.EIO, "io") |
| response = await self.get("/image-thumbnails/a.png") |
| self.assertEqual(response.status_code, 503) |
| self.assertEqual(response.headers["cache-control"], "no-store") |
|
|
| async def test_invalid_thumbnail_is_422_not_success(self): |
| response = await self.get("/image-thumbnails/a.png") |
| self.assertEqual(response.status_code, 422) |
|
|
| async def test_cached_original_and_download_do_not_enter_io_queue(self): |
| self.store.get_cached_bytes.return_value = b"hot-image" |
| with (mock.patch.object(system_module, "_run_image_io", side_effect=AssertionError("IO queue")), |
| mock.patch.object(system_module, "require_admin")): |
| original = await self.get("/images/a.png") |
| download = await self.get("/api/images/download/a.png") |
| self.assertEqual(original.content, b"hot-image") |
| self.assertEqual(download.content, b"hot-image") |
| self.assertEqual(original.status_code, 200) |
| self.assertIn("attachment", download.headers["content-disposition"]) |
| self.store.get_bytes.assert_not_called() |
|
|
| async def test_cached_thumbnail_does_not_enter_io_queue(self): |
| self.store.get_cached_thumbnail.return_value = b"hot-thumbnail" |
| with mock.patch.object(system_module, "_run_image_io", side_effect=AssertionError("IO queue")): |
| response = await self.get("/image-thumbnails/a.png") |
| self.assertEqual(response.content, b"hot-thumbnail") |
| self.store.get_bytes.assert_not_called() |
|
|
| async def test_hot_original_thumbnail_uses_separate_transform_pool(self): |
| buffer = io.BytesIO() |
| Image.new("RGBA", (512, 512), "green").save(buffer, format="PNG") |
| self.store.get_cached_bytes.return_value = buffer.getvalue() |
| self.store.get_bytes.side_effect = AssertionError("mount read") |
| with mock.patch.object(system_module, "_IMAGE_IO_POOL") as slow_pool: |
| slow_pool.submit.side_effect = AssertionError("blocked IO pool") |
| response = await self.get("/image-thumbnails/a.png") |
| self.assertEqual(response.status_code, 200) |
| with Image.open(io.BytesIO(response.content)) as result: |
| self.assertEqual(result.size, (320, 320)) |
|
|
| async def test_image_io_helper_forwards_keyword_arguments(self): |
| result = await system_module._run_image_io(lambda *, limit: limit, limit=3) |
| self.assertEqual(result, 3) |
|
|
| async def test_download_keeps_bytes_and_attachment_header(self): |
| with mock.patch.object(system_module, "require_admin"): |
| response = await self.get("/api/images/download/a.webp") |
| self.assertEqual(response.status_code, 200) |
| self.assertEqual(response.content, b"complete-image") |
| self.assertIn("attachment", response.headers["content-disposition"]) |
| self.assertIn("a.webp", response.headers["content-disposition"]) |
|
|
| async def test_download_eio_is_not_success(self): |
| self.store.get_bytes.side_effect = OSError(errno.EIO, "io") |
| with mock.patch.object(system_module, "require_admin"): |
| self.assertEqual((await self.get("/api/images/download/a.png")).status_code, 503) |
|
|
| async def test_slow_actual_read_is_isolated_and_has_route_timeout(self): |
| release, entered = Event(), Event() |
|
|
| def slow_read(_rel): |
| entered.set() |
| release.wait(2) |
| return b"image" |
|
|
| self.store.get_bytes.side_effect = slow_read |
| try: |
| with mock.patch.object(system_module, "_IMAGE_IO_TIMEOUT_SECONDS", 0.05): |
| pending = asyncio.create_task(self.get("/images/a.png")) |
| for _ in range(200): |
| if entered.is_set(): |
| break |
| await asyncio.sleep(0.001) |
| self.assertTrue(entered.is_set()) |
| self.assertEqual((await self.get("/test/ping")).status_code, 200) |
| response = await pending |
| self.assertEqual(response.status_code, 503) |
| finally: |
| release.set() |
|
|
| async def test_timeout_diagnostic_is_bounded_rate_limited_and_value_free(self): |
| release = Event() |
| def slow_read(private_value): |
| release.wait(2) |
| return private_value |
| try: |
| with (mock.patch.object(system_module, "_IMAGE_IO_TIMEOUT_SECONDS", 0.03), |
| mock.patch.object(system_module, "_last_image_timeout_log_at", float("-inf")), |
| mock.patch.object(system_module.logger, "warning") as warning): |
| for _ in range(2): |
| with self.assertRaises(HTTPException) as caught: |
| await system_module._run_image_io(slow_read, "private-path-and-secret") |
| self.assertEqual(caught.exception.status_code, 503) |
| warning.assert_called_once() |
| entry = warning.call_args.args[0] |
| self.assertEqual(entry["event"], "image_io_timeout") |
| self.assertEqual(entry["worker_state"], "running") |
| self.assertLessEqual(len(entry["stack"]), 10) |
| self.assertIn("slow_read", [f["function"] for f in entry["stack"]]) |
| self.assertNotIn("private-path-and-secret", json.dumps(entry)) |
| self.assertTrue(all(set(f) == {"module", "function", "line"} for f in entry["stack"])) |
| finally: |
| release.set() |
|
|
