Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions backend/src/find_api/routers/upload.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,12 @@ async def upload_images(
Returns:
List of created media records with job IDs
"""
if len(files) > settings.MAX_BULK_FILES:
raise HTTPException(
413,
f"Request contains more than {settings.MAX_BULK_FILES} files",
)

Comment thread
coderabbitai[bot] marked this conversation as resolved.
results = []

for file in files:
Expand Down
53 changes: 53 additions & 0 deletions backend/tests/test_upload.py
Original file line number Diff line number Diff line change
Expand Up @@ -239,3 +239,56 @@ def test_bulk_upload_oversized_file_skipped(self, client):
huge = next(r for r in results if r["filename"] == "huge.jpg")
assert huge["status"] == "failed"
assert "exceeds max upload size" in huge["error"].lower()

def test_bulk_upload_below_max_files(self, client):
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
Outdated
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
Outdated
"""ZIP with file count below MAX_BULK_FILES should succeed."""
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, "a", zipfile.ZIP_DEFLATED) as zf:
for i in range(5):
zf.writestr(f"image{i}.png", get_valid_image_bytes())
zip_buffer.seek(0)

response = client.post(
"/api/upload/bulk",
files=[("file", ("images.zip", zip_buffer.read(), "application/zip"))],
)
assert response.status_code == 200
results = response.json()["results"]
assert len(results) == 5
assert all(r["status"] == "uploaded" for r in results)

def test_bulk_upload_at_max_files(self, client):
"""ZIP with exactly MAX_BULK_FILES should succeed."""
from find_api.settings import settings

zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, "a", zipfile.ZIP_DEFLATED) as zf:
for i in range(settings.MAX_BULK_FILES):
zf.writestr(f"image{i}.png", get_valid_image_bytes())
zip_buffer.seek(0)

response = client.post(
"/api/upload/bulk",
files=[("file", ("images.zip", zip_buffer.read(), "application/zip"))],
)
assert response.status_code == 200
results = response.json()["results"]
assert len(results) == settings.MAX_BULK_FILES
assert all(r["status"] == "uploaded" for r in results)

def test_bulk_upload_above_max_files(self, client):
"""ZIP with file count exceeding MAX_BULK_FILES should be rejected."""
from find_api.settings import settings

zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, "a", zipfile.ZIP_DEFLATED) as zf:
for i in range(settings.MAX_BULK_FILES + 1):
zf.writestr(f"image{i}.png", get_valid_image_bytes())
zip_buffer.seek(0)

response = client.post(
"/api/upload/bulk",
files=[("file", ("images.zip", zip_buffer.read(), "application/zip"))],
)
assert response.status_code == 400
assert "files" in response.json()["detail"].lower() and "limit" in response.json()["detail"].lower()
Loading