reactbin/api/tests/integration/test_upload.py

"""
T026 — valid JPEG upload → 201, record in DB, object in MinIO
T027 — same image uploaded twice → 200, duplicate: true, no second MinIO object
T028 — invalid MIME type → 422 invalid_mime_type (error envelope with code field)
T029 — file > MAX_UPLOAD_BYTES → 422 file_too_large
T013 — upload produces short_id; storage_key equals short_id; thumbnail_key = {short_id}-thumb
"""
import io
import re
from unittest.mock import patch

import pytest
from PIL import Image as PILImage


def _real_jpeg(color: tuple = (100, 150, 200), size: tuple = (200, 150)) -> bytes:
    buf = io.BytesIO()
    PILImage.new("RGB", size, color=color).save(buf, format="JPEG")
    return buf.getvalue()


def _minimal_jpeg() -> bytes:
    # Minimal valid JPEG bytes (SOI + APP0 + EOI)
    return (
        b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00"
        b"\xff\xd9"
    )


@pytest.mark.asyncio
async def test_upload_new_image_returns_201(authed_client):
    client, token = authed_client
    data = _minimal_jpeg()
    response = await client.post(
        "/api/v1/images",
        files={"file": ("test.jpg", io.BytesIO(data), "image/jpeg")},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 201
    body = response.json()
    assert body["duplicate"] is False
    assert body["filename"] == "test.jpg"
    assert body["mime_type"] == "image/jpeg"
    assert "id" in body
    assert "hash" in body
    assert len(body["hash"]) == 64


@pytest.mark.asyncio
async def test_upload_duplicate_returns_200_with_flag(authed_client):
    client, token = authed_client
    data = _minimal_jpeg()
    headers = {"Authorization": f"Bearer {token}"}
    # First upload
    r1 = await client.post(
        "/api/v1/images",
        files={"file": ("test.jpg", io.BytesIO(data), "image/jpeg")},
        headers=headers,
    )
    assert r1.status_code in (200, 201)

    # Second upload of same bytes
    r2 = await client.post(
        "/api/v1/images",
        files={"file": ("test.jpg", io.BytesIO(data), "image/jpeg")},
        headers=headers,
    )
    assert r2.status_code == 200
    body = r2.json()
    assert body["duplicate"] is True
    assert body["id"] == r1.json()["id"]


@pytest.mark.asyncio
async def test_upload_invalid_mime_type_returns_422(authed_client):
    client, token = authed_client
    response = await client.post(
        "/api/v1/images",
        files={"file": ("doc.pdf", io.BytesIO(b"%PDF-1.4"), "application/pdf")},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 422
    body = response.json()
    assert body["code"] == "invalid_mime_type"
    assert "detail" in body


@pytest.mark.asyncio
async def test_upload_oversized_file_returns_422(authed_client):
    import os

    from app.config import get_settings

    client, token = authed_client
    os.environ["MAX_UPLOAD_BYTES"] = "10"
    get_settings.cache_clear()

    try:
        response = await client.post(
            "/api/v1/images",
            files={"file": ("big.jpg", io.BytesIO(b"x" * 11), "image/jpeg")},
            headers={"Authorization": f"Bearer {token}"},
        )
        assert response.status_code == 422
        body = response.json()
        assert body["code"] == "file_too_large"
    finally:
        del os.environ["MAX_UPLOAD_BYTES"]
        get_settings.cache_clear()


@pytest.mark.asyncio
async def test_get_unknown_image_returns_404_with_envelope(client):
    response = await client.get("/api/v1/i/NotFound")
    assert response.status_code == 404
    body = response.json()
    assert body["code"] == "image_not_found"
    assert "detail" in body


_SHORT_ID_RE = re.compile(r"^[a-zA-Z0-9]{8}$")


@pytest.mark.asyncio
async def test_upload_returns_short_id(authed_client):
    client, token = authed_client
    data = _minimal_jpeg()
    response = await client.post(
        "/api/v1/images",
        files={"file": ("s1.jpg", io.BytesIO(data), "image/jpeg")},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 201
    body = response.json()
    assert "short_id" in body
    assert _SHORT_ID_RE.match(body["short_id"]), f"short_id invalid: {body['short_id']}"


@pytest.mark.asyncio
async def test_upload_storage_key_equals_short_id(authed_client):
    client, token = authed_client
    data = _real_jpeg(color=(10, 20, 30))
    response = await client.post(
        "/api/v1/images",
        files={"file": ("s2.jpg", io.BytesIO(data), "image/jpeg")},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 201
    body = response.json()
    assert body["storage_key"] == body["short_id"]


@pytest.mark.asyncio
async def test_upload_thumbnail_key_equals_short_id_thumb(authed_client):
    client, token = authed_client
    data = _real_jpeg(color=(30, 60, 90))
    response = await client.post(
        "/api/v1/images",
        files={"file": ("s3.jpg", io.BytesIO(data), "image/jpeg")},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 201
    body = response.json()
    if body["thumbnail_key"] is not None:
        assert body["thumbnail_key"] == f"{body['short_id']}-thumb"


@pytest.mark.asyncio
async def test_duplicate_upload_returns_same_short_id(authed_client):
    client, token = authed_client
    data = _real_jpeg(color=(200, 100, 50))
    headers = {"Authorization": f"Bearer {token}"}
    r1 = await client.post(
        "/api/v1/images",
        files={"file": ("dup_short.jpg", io.BytesIO(data), "image/jpeg")},
        headers=headers,
    )
    assert r1.status_code in (200, 201)
    r2 = await client.post(
        "/api/v1/images",
        files={"file": ("dup_short.jpg", io.BytesIO(data), "image/jpeg")},
        headers=headers,
    )
    assert r2.status_code == 200
    assert r2.json()["duplicate"] is True
    assert r2.json()["short_id"] == r1.json()["short_id"]


@pytest.mark.asyncio
async def test_upload_returns_thumbnail_key(authed_client):
    client, token = authed_client
    data = _real_jpeg(color=(100, 150, 200))
    response = await client.post(
        "/api/v1/images",
        files={"file": ("thumb_test.jpg", io.BytesIO(data), "image/jpeg")},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 201
    body = response.json()
    assert "thumbnail_key" in body
    assert body["thumbnail_key"] is not None
    assert body["thumbnail_key"].endswith("-thumb")
    assert "file_url" in body
    assert body["file_url"].startswith("/api/v1/i/")
    assert "thumbnail_url" in body
    assert body["thumbnail_url"].startswith("/api/v1/i/")


@pytest.mark.asyncio
async def test_duplicate_upload_reuses_thumbnail_key(authed_client):
    client, token = authed_client
    headers = {"Authorization": f"Bearer {token}"}
    data = _real_jpeg(color=(200, 100, 50))
    r1 = await client.post(
        "/api/v1/images",
        files={"file": ("dup.jpg", io.BytesIO(data), "image/jpeg")},
        headers=headers,
    )
    assert r1.status_code in (200, 201)

    r2 = await client.post(
        "/api/v1/images",
        files={"file": ("dup.jpg", io.BytesIO(data), "image/jpeg")},
        headers=headers,
    )
    assert r2.status_code == 200

    tk1 = r1.json()["thumbnail_key"]
    tk2 = r2.json()["thumbnail_key"]
    assert tk1 is not None
    assert tk1 == tk2


@pytest.mark.asyncio
async def test_upload_succeeds_when_thumbnail_fails(authed_client):
    client, token = authed_client
    data = _real_jpeg(color=(50, 200, 150))
    with patch("app.routers.images.generate_thumbnail", side_effect=RuntimeError("simulated")):
        response = await client.post(
            "/api/v1/images",
            files={"file": ("no_thumb.jpg", io.BytesIO(data), "image/jpeg")},
            headers={"Authorization": f"Bearer {token}"},
        )
    assert response.status_code in (200, 201)
    body = response.json()
    assert body["thumbnail_key"] is None
    assert "file_url" in body
    assert body["file_url"].startswith("/api/v1/i/")
    assert body["thumbnail_url"] is None