agatha
f3e0021ee8
- conftest.py: pytest_configure guard rejects non-postgresql+asyncpg:// URLs before any test collects (per constitution §2.5/§5.2 v1.3.0) - docker-compose.test.yml: isolated postgres-test (5433) + minio-test (9002) + api-test runner; one command runs the full suite against real PostgreSQL - Makefile: test-unit and test-integration targets - .env.test.example: documents variables needed to run tests outside Docker - Fix pre-existing test bug: integration tests using client fixture (NoOpAuthProvider) for write operations (upload/delete/patch) now use authed_client with Bearer token — these were never caught because tests never ran against a live stack Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
133 lines
4.3 KiB
Python
133 lines
4.3 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(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["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(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()["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(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["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(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()["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"
|