agatha
f953c88984
- Add Pillow dependency and thumbnail.py with generate_thumbnail() — produces
WebP ≤400px, preserves aspect ratio, never upscales, handles GIF frame 0
- Alembic migration 002 adds nullable thumbnail_key column to images table
- Upload route generates thumbnail via asyncio.to_thread (non-blocking),
stores at {hash}-thumb; failure is tolerated and upload succeeds with null key
- New GET /api/v1/images/{id}/thumbnail endpoint: serves WebP thumbnail or
falls back to original for pre-feature images; ETag + immutable cache headers
- Delete route cleans up thumbnail storage object alongside original
- Library grid switches from /file to /thumbnail for all image src bindings
- 59 tests passing (46 existing + 13 new across unit, upload, serving, delete)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
"""
|
|
T055 — GET /api/v1/images/{id}/file → 200 with binary content, ETag, Cache-Control
|
|
T056 — /file for unknown ID → 404 image_not_found
|
|
T057 — /file response exposes no storage-specific details
|
|
"""
|
|
import io
|
|
import uuid
|
|
|
|
import pytest
|
|
from PIL import Image as PILImage
|
|
from sqlalchemy import update
|
|
|
|
from app.models import Image
|
|
|
|
|
|
def _real_jpeg() -> bytes:
|
|
buf = io.BytesIO()
|
|
PILImage.new("RGB", (200, 150), color=(120, 80, 200)).save(buf, format="JPEG")
|
|
return buf.getvalue()
|
|
|
|
|
|
def _minimal_webp() -> bytes:
|
|
# Minimal VP8L WebP
|
|
return (
|
|
b"RIFF$\x00\x00\x00WEBPVP8L\x18\x00\x00\x00"
|
|
b"/\x00\x00\x00\x00\x18\xf0\x1f\xfe\xff\x02\xfe\x00"
|
|
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_returns_200_with_content(client):
|
|
data = _minimal_webp()
|
|
upload = await client.post(
|
|
"/api/v1/images",
|
|
files={"file": ("img.webp", io.BytesIO(data), "image/webp")},
|
|
)
|
|
assert upload.status_code in (200, 201)
|
|
upload_body = upload.json()
|
|
image_id = upload_body["id"]
|
|
image_hash = upload_body["hash"]
|
|
|
|
response = await client.get(f"/api/v1/images/{image_id}/file")
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"].startswith("image/")
|
|
assert response.headers["etag"] == f'"{image_hash}"'
|
|
assert "immutable" in response.headers["cache-control"]
|
|
assert len(response.content) > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_unknown_id_returns_404(client):
|
|
response = await client.get(f"/api/v1/images/{uuid.uuid4()}/file")
|
|
assert response.status_code == 404
|
|
body = response.json()
|
|
assert body["code"] == "image_not_found"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_response_exposes_no_storage_details(client):
|
|
data = _minimal_webp()
|
|
upload = await client.post(
|
|
"/api/v1/images",
|
|
files={"file": ("img.webp", io.BytesIO(data), "image/webp")},
|
|
)
|
|
assert upload.status_code in (200, 201)
|
|
image_id = upload.json()["id"]
|
|
|
|
response = await client.get(f"/api/v1/images/{image_id}/file")
|
|
assert response.status_code == 200
|
|
assert "location" not in response.headers
|
|
assert "minio" not in response.text.lower()
|
|
assert "s3://" not in response.text.lower()
|
|
assert "amazonaws.com" not in response.text.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_thumbnail_returns_webp(client):
|
|
data = _real_jpeg()
|
|
upload = await client.post(
|
|
"/api/v1/images",
|
|
files={"file": ("t.jpg", io.BytesIO(data), "image/jpeg")},
|
|
)
|
|
assert upload.status_code == 201
|
|
body = upload.json()
|
|
image_id = body["id"]
|
|
image_hash = body["hash"]
|
|
|
|
response = await client.get(f"/api/v1/images/{image_id}/thumbnail")
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "image/webp"
|
|
assert response.headers["etag"] == f'"{image_hash}"'
|
|
assert "immutable" in response.headers["cache-control"]
|
|
assert len(response.content) > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_thumbnail_fallback_returns_original(client, db_session):
|
|
data = _real_jpeg()
|
|
upload = await client.post(
|
|
"/api/v1/images",
|
|
files={"file": ("fallback.jpg", io.BytesIO(data), "image/jpeg")},
|
|
)
|
|
assert upload.status_code == 201
|
|
image_id = upload.json()["id"]
|
|
|
|
await db_session.execute(
|
|
update(Image).where(Image.id == uuid.UUID(image_id)).values(thumbnail_key=None)
|
|
)
|
|
await db_session.flush()
|
|
db_session.expire_all()
|
|
|
|
response = await client.get(f"/api/v1/images/{image_id}/thumbnail")
|
|
assert response.status_code == 200
|
|
assert "image/jpeg" in response.headers["content-type"]
|
|
assert len(response.content) > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_thumbnail_unknown_id_returns_404(client):
|
|
response = await client.get(f"/api/v1/images/{uuid.uuid4()}/thumbnail")
|
|
assert response.status_code == 404
|
|
body = response.json()
|
|
assert body["code"] == "image_not_found"
|