agatha
61d923d5be
Short IDs become the canonical identifier in URLs (/i/:short_id), MinIO/R2 storage keys, and all API responses. Hash-based deduplication is preserved. Includes two-phase Alembic migration (003 adds nullable column, 004 enforces NOT NULL) with a backfill script to copy storage objects and populate short_id for existing images. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
132 lines
4.2 KiB
Python
132 lines
4.2 KiB
Python
"""
|
|
T055 — GET /api/v1/i/{short_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 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(authed_client):
|
|
client, token = authed_client
|
|
data = _minimal_webp()
|
|
upload = await client.post(
|
|
"/api/v1/images",
|
|
files={"file": ("img.webp", io.BytesIO(data), "image/webp")},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert upload.status_code in (200, 201)
|
|
upload_body = upload.json()
|
|
image_id = upload_body["short_id"]
|
|
image_hash = upload_body["hash"]
|
|
|
|
response = await client.get(f"/api/v1/i/{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("/api/v1/i/NotFound/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(authed_client):
|
|
client, token = authed_client
|
|
data = _minimal_webp()
|
|
upload = await client.post(
|
|
"/api/v1/images",
|
|
files={"file": ("img.webp", io.BytesIO(data), "image/webp")},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert upload.status_code in (200, 201)
|
|
image_id = upload.json()["short_id"]
|
|
|
|
response = await client.get(f"/api/v1/i/{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(authed_client):
|
|
client, token = authed_client
|
|
data = _real_jpeg()
|
|
upload = await client.post(
|
|
"/api/v1/images",
|
|
files={"file": ("t.jpg", io.BytesIO(data), "image/jpeg")},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert upload.status_code == 201
|
|
body = upload.json()
|
|
image_id = body["short_id"]
|
|
image_hash = body["hash"]
|
|
|
|
response = await client.get(f"/api/v1/i/{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(authed_client, db_session):
|
|
client, token = authed_client
|
|
data = _real_jpeg()
|
|
upload = await client.post(
|
|
"/api/v1/images",
|
|
files={"file": ("fallback.jpg", io.BytesIO(data), "image/jpeg")},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert upload.status_code == 201
|
|
image_id = upload.json()["short_id"]
|
|
|
|
await db_session.execute(
|
|
update(Image).where(Image.short_id == image_id).values(thumbnail_key=None)
|
|
)
|
|
await db_session.flush()
|
|
db_session.expire_all()
|
|
|
|
response = await client.get(f"/api/v1/i/{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("/api/v1/i/NotFound/thumbnail")
|
|
assert response.status_code == 404
|
|
body = response.json()
|
|
assert body["code"] == "image_not_found"
|