agatha
cd89ba5dea
Replace the presigned-URL redirect (302) in GET /api/v1/images/{id}/file
with a direct proxy that fetches bytes from S3 server-side and returns them
to the client. The browser never contacts the storage backend, eliminating
the /etc/hosts workaround needed in local development.
- StorageBackend: swap get_presigned_url for get(key) -> bytes
- S3StorageBackend: implement get() via aiobotocore get_object
- serve_image_file: return Response with ETag + Cache-Control: immutable
- test_serving: assert 200 + content-type + ETag; add no-storage-details test
- Spec Kit artifacts for feature 002-api-image-proxy
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
65 lines
2.0 KiB
Python
65 lines
2.0 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
|
|
|
|
|
|
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()
|