bede/tfm_ainventory
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(phase1): add static file serving for /images/

Browse Source
This commit is contained in:
Daniel Bedeleanu
2026-04-20 22:43:33 +03:00
parent 8d2750cfa3
commit 294555c574
22 changed files with 1989 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
backend/main.py
View File

@@ -2,6 +2,7 @@ import os
from . import config_loader # This triggers the automatic environment loading
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from slowapi import Limiter
from slowapi.util import get_remote_address
from . import models
@@ -10,7 +11,7 @@ from .routers import items, operations, users, auth, sync, categories
from .routers.admin import backups, ai_config, db_config
from .logger import log
from .scheduler import scheduler, sync_scheduler_config
from .services.image_storage import ensure_image_directories
from .services.image_storage import ensure_image_directories, IMAGES_ROOT
# Create the database tables
from .database import DATA_DIR, db_path
@@ -95,6 +96,14 @@ app.include_router(backups.router)
app.include_router(ai_config.router)
app.include_router(db_config.router)
# [STATIC FILES] Mount /images/ directory for serving uploaded photos
# Ensure directory exists before mounting (StaticFiles requires pre-existing directory)
IMAGES_ROOT.mkdir(parents=True, exist_ok=True)
# Mount at root level after all API routes to avoid conflicts with dynamic routes.
# MIME types are auto-detected by StaticFiles based on file extensions.
# Supported formats: JPEG (.jpg/.jpeg), PNG (.png), WebP (.webp), GIF (.gif)
app.mount("/images", StaticFiles(directory=str(IMAGES_ROOT)), name="images")
@app.on_event("startup")
def startup_event():
log.info("[STARTUP] Initializing image storage directories...")

190
backend/routers/items.py
View File

@@ -1,12 +1,16 @@
from fastapi import APIRouter, Depends, HTTPException, status, UploadFile, File, Request
from sqlalchemy.orm import Session
from sqlalchemy import func
from typing import List
from typing import List, Optional, Dict
import json
from datetime import datetime, timezone
from pathlib import Path
from slowapi import Limiter
from slowapi.util import get_remote_address
from .. import models, schemas, auth
from ..database import get_db
from ..services.image_processing import ImageProcessor
from ..services.image_storage import save_image, get_unique_filename
# [H-02] Rate limiter for extract-label endpoint
limiter = Limiter(key_func=get_remote_address)
@@ -52,10 +56,21 @@ def read_item(
db: Session = Depends(get_db),
current_user: auth.TokenData = Depends(auth.get_current_user)
):
"""[C-01] Get item — only for authenticated users."""
"""[C-01] Get item — only for authenticated users. [PHASE1-T5] Include photo URLs if set."""
item = db.query(models.Item).filter(models.Item.id == item_id).first()
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
# Build photo object if photo_path is set
if item.photo_path and item.photo_thumbnail_path and item.photo_upload_date:
item.photo = schemas.PhotoResponse(
thumbnail_url=item.photo_thumbnail_path,
full_url=item.photo_path,
uploaded_at=item.photo_upload_date
)
else:
item.photo = None
return item
_ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/gif"}
@@ -233,8 +248,177 @@ def delete_item(
# [CLEANUP] Delete related InterventionItems to prevent foreign key issues
db.query(models.InterventionItem).filter(models.InterventionItem.item_id == item_id).delete()
# Audit Logs in database are NOT deleted here to preserve history of actions
db.delete(db_item)
db.commit()
return {"message": "Item deleted successfully. History logs preserved."}
@router.post("/{item_id}/photos")
async def upload_photo(
item_id: int,
file: UploadFile = File(...),
crop_bounds: Optional[str] = "",
replace_existing: Optional[str] = "",
db: Session = Depends(get_db),
current_user: auth.TokenData = Depends(auth.get_current_user)
):
"""
[PHASE1-T4] Upload/replace photo for an item.
- Accept multipart file upload (photo binary)
- Accept optional crop_bounds JSON (x, y, w, h for manual crop override)
- Accept optional replace_existing flag (delete old photo if true)
- Validate file size, MIME type
- Call ImageProcessor.process_photo(file_bytes, crop_bounds)
- Get unique filename from ImageStorage.get_unique_filename()
- Save original and thumbnail using ImageStorage.save_image()
- Delete old photo file if replacing
- Update Item.photo_path, photo_thumbnail_path, photo_upload_date
- Return: {status: "ok", photo: {thumbnail_url, full_url, uploaded_at}}
"""
# Verify item exists
db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
if not db_item:
raise HTTPException(status_code=404, detail="Item not found")
# Validate file type
if file.content_type not in _ALLOWED_IMAGE_TYPES:
raise HTTPException(
status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
detail=f"File type not allowed: {file.content_type}. Accepted: {', '.join(_ALLOWED_IMAGE_TYPES)}"
)
# Read file and validate size
file_bytes = await file.read()
if len(file_bytes) > _MAX_IMAGE_SIZE:
raise HTTPException(
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
detail="File exceeds 10MB limit."
)
try:
# Parse crop_bounds if provided
crop_bounds_dict = None
if crop_bounds and crop_bounds.strip():
try:
crop_bounds_dict = json.loads(crop_bounds)
except json.JSONDecodeError:
raise HTTPException(
status_code=400,
detail="Invalid crop_bounds JSON"
)
# Parse replace_existing flag (comes as form string)
should_replace = replace_existing and replace_existing.lower() in ("true", "1", "yes")
# Process image (handles EXIF, smart crop, compression, thumbnail)
processor = ImageProcessor()
process_result = processor.process_photo(file_bytes, crop_bounds_dict)
if process_result['status'] != 'success':
raise HTTPException(
status_code=400,
detail=f"Image processing failed: {process_result.get('error', 'Unknown error')}"
)
# Get processed bytes
cropped_bytes = process_result['cropped_image_bytes']
thumbnail_bytes = process_result['thumbnail_bytes']
if not cropped_bytes or not thumbnail_bytes:
raise HTTPException(
status_code=400,
detail="Failed to process image data"
)
# Get category for file storage (use "items" as default category if no category set)
category = db_item.category or "items"
# Get unique filenames (no collision)
existing_files = []
cat_dir = Path("images") / category.lower()
if cat_dir.exists():
existing_files = [f.name for f in cat_dir.iterdir() if f.is_file()]
filename_base = db_item.name or f"item_{item_id}"
original_filename = get_unique_filename(filename_base, category, existing_files, variant="original")
thumbnail_filename = get_unique_filename(filename_base, category, existing_files, variant="thumb")
# Save original image
try:
original_path = save_image(cropped_bytes, category, original_filename.replace("_original.jpg", ""), variant="original")
except (OSError, IOError) as e:
if "No space left" in str(e):
raise HTTPException(
status_code=status.HTTP_507_INSUFFICIENT_STORAGE,
detail="Disk space full"
)
raise HTTPException(
status_code=400,
detail=f"Failed to save image: {str(e)}"
)
# Save thumbnail
try:
thumbnail_path = save_image(thumbnail_bytes, category, thumbnail_filename.replace("_thumb.jpg", ""), variant="thumb")
except (OSError, IOError) as e:
if "No space left" in str(e):
raise HTTPException(
status_code=status.HTTP_507_INSUFFICIENT_STORAGE,
detail="Disk space full"
)
raise HTTPException(
status_code=400,
detail=f"Failed to save thumbnail: {str(e)}"
)
# Delete old photo files if replacing
if should_replace and db_item.photo_path:
try:
old_photo = Path(db_item.photo_path.lstrip("/"))
if old_photo.exists():
old_photo.unlink()
except Exception as e:
# Log but don't fail if cleanup fails
from ..logger import log
log.warning(f"Failed to delete old photo: {str(e)}")
try:
if db_item.photo_thumbnail_path:
old_thumb = Path(db_item.photo_thumbnail_path.lstrip("/"))
if old_thumb.exists():
old_thumb.unlink()
except Exception as e:
from ..logger import log
log.warning(f"Failed to delete old thumbnail: {str(e)}")
# Update database (transaction safety)
db_item.photo_path = original_path
db_item.photo_thumbnail_path = thumbnail_path
db_item.photo_upload_date = datetime.now(timezone.utc)
db.commit()
db.refresh(db_item)
# Return response with URLs
return {
"status": "ok",
"photo": {
"thumbnail_url": db_item.photo_thumbnail_path,
"full_url": db_item.photo_path,
"uploaded_at": db_item.photo_upload_date
}
}
except HTTPException:
raise
except Exception as e:
db.rollback()
from ..logger import log
log.error(f"Unexpected error in photo upload: {str(e)}", exc_info=True)
raise HTTPException(
status_code=500,
detail=f"Internal server error: {str(e)}"
)

2
backend/schemas/__init__.py
View File

@@ -22,6 +22,7 @@ from .items import (
ColorBase,
ColorCreate,
Color,
PhotoResponse,
ItemBase,
ItemCreate,
Item,
@@ -57,6 +58,7 @@ __all__ = [
"ColorBase",
"ColorCreate",
"Color",
"PhotoResponse",
"ItemBase",
"ItemCreate",
"Item",

13
backend/schemas/items.py
View File

@@ -1,5 +1,14 @@
from pydantic import BaseModel
from typing import Optional
from datetime import datetime
# --- Photo Response ---
class PhotoResponse(BaseModel):
"""Photo metadata for item responses."""
thumbnail_url: str
full_url: str
uploaded_at: datetime
# --- Categories ---
@@ -62,6 +71,10 @@ class ItemCreate(ItemBase):
class Item(ItemBase):
id: int
photo_path: Optional[str] = None
photo_thumbnail_path: Optional[str] = None
photo_upload_date: Optional[datetime] = None
photo: Optional[PhotoResponse] = None
class Config:
from_attributes = True

376
backend/tests/test_photo_endpoints.py Normal file
View File

@@ -0,0 +1,376 @@
"""
Comprehensive test suite for photo upload/replace API endpoints.
Tests cover:
- Upload success: photo saved, DB updated, URLs returned
- File replacement: old file deleted, new file saved
- Auth: user can upload, guest cannot
- Invalid file: wrong MIME type rejected
- File too large: >10MB rejected with 413
- Item not found: 404
- GET returns photo URLs when set, null when not
"""
import io
import json
from pathlib import Path
from unittest.mock import patch, MagicMock
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from backend import models, auth
# ============================================================================
# FIXTURES
# ============================================================================
@pytest.fixture
def test_item(test_db: Session):
"""Create a test item in the database."""
item = models.Item(
id=1,
barcode="TEST123",
name="Test Item",
category="networking",
quantity=5.0
)
test_db.add(item)
test_db.commit()
test_db.refresh(item)
return item
@pytest.fixture
def sample_image():
"""Create a minimal valid JPEG image for testing."""
from PIL import Image
# Create a simple 100x100 red image
img = Image.new('RGB', (100, 100), color='red')
img_bytes = io.BytesIO()
img.save(img_bytes, format='JPEG')
img_bytes.seek(0)
return img_bytes.getvalue()
@pytest.fixture
def large_image():
"""Create an image > 10MB to test size limit."""
from PIL import Image
img = Image.new('RGB', (100, 100), color='blue')
img_bytes = io.BytesIO()
img.save(img_bytes, format='JPEG')
img_bytes.seek(0)
small_img = img_bytes.getvalue()
# Pad it to >10MB
large_img = small_img + b'X' * (11 * 1024 * 1024)
return large_img
@pytest.fixture
def invalid_file():
"""Create an invalid file (text instead of image)."""
return b"This is not an image file"
# ============================================================================
# TESTS: UPLOAD SUCCESS
# ============================================================================
def test_upload_photo_success(user_client: TestClient, test_db: Session, test_item, sample_image):
"""Test successful photo upload."""
response = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test.jpg", io.BytesIO(sample_image), "image/jpeg")}
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "ok"
assert "photo" in data
assert "thumbnail_url" in data["photo"]
assert "full_url" in data["photo"]
assert "uploaded_at" in data["photo"]
# Verify DB was updated
updated_item = test_db.query(models.Item).filter(
models.Item.id == test_item.id
).first()
assert updated_item.photo_path is not None
assert updated_item.photo_thumbnail_path is not None
assert updated_item.photo_upload_date is not None
# Cleanup
Path(updated_item.photo_path.lstrip("/")).unlink()
Path(updated_item.photo_thumbnail_path.lstrip("/")).unlink()
def test_upload_photo_creates_files(user_client: TestClient, test_db: Session, test_item, sample_image):
"""Test that photo upload creates image files on disk."""
response = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test.jpg", io.BytesIO(sample_image), "image/jpeg")}
)
assert response.status_code == 200
data = response.json()
# Check that files exist
original_path = Path(data["photo"]["full_url"].lstrip("/"))
thumbnail_path = Path(data["photo"]["thumbnail_url"].lstrip("/"))
assert original_path.exists(), f"Original image not found at {original_path}"
assert thumbnail_path.exists(), f"Thumbnail not found at {thumbnail_path}"
# Cleanup
original_path.unlink()
thumbnail_path.unlink()
# ============================================================================
# TESTS: FILE REPLACEMENT
# ============================================================================
def test_upload_photo_replace_existing(user_client: TestClient, test_db: Session, test_item, sample_image):
"""Test photo replacement with replace_existing=true."""
# First upload
response1 = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test1.jpg", io.BytesIO(sample_image), "image/jpeg")}
)
assert response1.status_code == 200
old_data = response1.json()
first_url = old_data["photo"]["full_url"]
# Create different image for second upload
from PIL import Image
img2 = Image.new('RGB', (150, 150), color='green')
img2_bytes = io.BytesIO()
img2.save(img2_bytes, format='JPEG')
img2_bytes.seek(0)
# Second upload with replace_existing=true
response2 = user_client.post(
f"/items/{test_item.id}/photos",
data={"replace_existing": "true"},
files={"file": ("test2.jpg", io.BytesIO(img2_bytes.getvalue()), "image/jpeg")}
)
assert response2.status_code == 200
new_data = response2.json()
second_url = new_data["photo"]["full_url"]
# URL should have changed (file was replaced)
assert first_url != second_url, "Photo URL should change when replacing"
# New files should exist
new_original = Path(new_data["photo"]["full_url"].lstrip("/"))
new_thumbnail = Path(new_data["photo"]["thumbnail_url"].lstrip("/"))
assert new_original.exists()
assert new_thumbnail.exists()
# Cleanup
new_original.unlink()
new_thumbnail.unlink()
# ============================================================================
# TESTS: AUTHENTICATION
# ============================================================================
def test_upload_photo_requires_auth(test_client: TestClient, test_item, sample_image):
"""Test that unauthenticated upload fails with 401."""
response = test_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test.jpg", io.BytesIO(sample_image), "image/jpeg")}
)
assert response.status_code == 401 # No auth header (unauthorized)
def test_upload_photo_with_valid_token(user_client: TestClient, test_db: Session, test_item, sample_image):
"""Test that authenticated user can upload."""
response = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test.jpg", io.BytesIO(sample_image), "image/jpeg")}
)
assert response.status_code == 200
# Cleanup
data = response.json()
Path(data["photo"]["full_url"].lstrip("/")).unlink()
Path(data["photo"]["thumbnail_url"].lstrip("/")).unlink()
# ============================================================================
# TESTS: FILE VALIDATION
# ============================================================================
def test_upload_photo_invalid_mime_type(user_client: TestClient, test_item, invalid_file):
"""Test that invalid MIME type is rejected."""
response = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test.txt", io.BytesIO(invalid_file), "text/plain")}
)
assert response.status_code == 415 # Unsupported Media Type
data = response.json()
assert "File type not allowed" in data["detail"]
def test_upload_photo_accepted_mime_types(user_client: TestClient, test_db: Session, test_item, sample_image):
"""Test all accepted MIME types."""
mime_types = ["image/jpeg", "image/png", "image/webp", "image/gif"]
for mime_type in mime_types:
response = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test.jpg", io.BytesIO(sample_image), mime_type)}
)
assert response.status_code == 200, f"Failed for {mime_type}"
# Cleanup
data = response.json()
Path(data["photo"]["full_url"].lstrip("/")).unlink()
Path(data["photo"]["thumbnail_url"].lstrip("/")).unlink()
# ============================================================================
# TESTS: FILE SIZE LIMITS
# ============================================================================
def test_upload_photo_too_large(user_client: TestClient, test_item, large_image):
"""Test that files > 10MB are rejected with 413."""
response = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("large.jpg", io.BytesIO(large_image), "image/jpeg")}
)
assert response.status_code == 413 # Payload Too Large
data = response.json()
assert "exceeds 10MB limit" in data["detail"]
# ============================================================================
# TESTS: ITEM NOT FOUND
# ============================================================================
def test_upload_photo_item_not_found(user_client: TestClient, sample_image):
"""Test that upload to non-existent item returns 404."""
response = user_client.post(
f"/items/99999/photos",
files={"file": ("test.jpg", io.BytesIO(sample_image), "image/jpeg")}
)
assert response.status_code == 404
data = response.json()
assert "Item not found" in data["detail"]
# ============================================================================
# TESTS: GET ENDPOINT PHOTO RESPONSE
# ============================================================================
def test_get_item_includes_photo_urls(user_client: TestClient, test_db: Session, test_item, sample_image):
"""Test that GET /items/{id} includes photo URLs when photo is set."""
# First upload a photo
response = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test.jpg", io.BytesIO(sample_image), "image/jpeg")}
)
assert response.status_code == 200
# Get the item
response = user_client.get(f"/items/{test_item.id}")
assert response.status_code == 200
data = response.json()
assert "photo" in data
assert data["photo"] is not None
assert "thumbnail_url" in data["photo"]
assert "full_url" in data["photo"]
assert "uploaded_at" in data["photo"]
# Cleanup
Path(data["photo"]["full_url"].lstrip("/")).unlink()
Path(data["photo"]["thumbnail_url"].lstrip("/")).unlink()
def test_get_item_no_photo_returns_null(user_client: TestClient, test_item):
"""Test that GET /items/{id} returns photo: null when no photo is set."""
response = user_client.get(f"/items/{test_item.id}")
assert response.status_code == 200
data = response.json()
assert "photo" in data
assert data["photo"] is None
# ============================================================================
# TESTS: MULTIPLE UPLOADS
# ============================================================================
def test_upload_multiple_photos_without_replace(user_client: TestClient, test_db: Session, test_item, sample_image):
"""Test multiple uploads without replace_existing (new files created)."""
from PIL import Image
response1 = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test1.jpg", io.BytesIO(sample_image), "image/jpeg")}
)
assert response1.status_code == 200
path1 = response1.json()["photo"]["full_url"]
# Create a different image
img2 = Image.new('RGB', (200, 200), color='yellow')
img2_bytes = io.BytesIO()
img2.save(img2_bytes, format='JPEG')
img2_bytes.seek(0)
response2 = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test2.jpg", io.BytesIO(img2_bytes.getvalue()), "image/jpeg")}
)
assert response2.status_code == 200
path2 = response2.json()["photo"]["full_url"]
# Paths should be different (new file created)
assert path1 != path2
# Cleanup
Path(path1.lstrip("/")).unlink()
Path(path2.lstrip("/")).unlink()
# ============================================================================
# TESTS: EDGE CASES
# ============================================================================
def test_upload_photo_empty_file(user_client: TestClient, test_item):
"""Test that empty file is handled gracefully."""
response = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test.jpg", io.BytesIO(b""), "image/jpeg")}
)
# Should fail during image processing
assert response.status_code >= 400
def test_upload_photo_corrupted_jpeg(user_client: TestClient, test_item):
"""Test that corrupted JPEG is handled gracefully."""
corrupted = b"\xFF\xD8\xFF\xE0" + b"corrupted data"
response = user_client.post(
f"/items/{test_item.id}/photos",
files={"file": ("test.jpg", io.BytesIO(corrupted), "image/jpeg")}
)
# Should fail during image processing
assert response.status_code >= 400

256
backend/tests/test_static_files.py Normal file
View File

@@ -0,0 +1,256 @@
"""Tests for static file serving of uploaded images."""
import os
import tempfile
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from backend.main import app
from backend.services.image_storage import save_image, IMAGES_ROOT, ensure_image_directories
@pytest.fixture(autouse=True)
def setup_images_directory():
"""Create and clean up images directory for each test."""
# Ensure directory exists
IMAGES_ROOT.mkdir(parents=True, exist_ok=True)
yield
# Cleanup after test
import shutil
if IMAGES_ROOT.exists():
shutil.rmtree(IMAGES_ROOT)
@pytest.fixture
def client():
"""FastAPI test client."""
return TestClient(app)
@pytest.fixture
def sample_jpeg_image():
"""Create a minimal valid JPEG image for testing."""
# Minimal JPEG header + footer
jpeg_data = (
b'\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00'
b'\xff\xdb\x00C\x00\x08\x06\x06\x07\x06\x05\x08\x07\x07\x07\t\t'
b'\x08\n\x0c\x14\r\x0c\x0b\x0b\x0c\x19\x12\x13\x0f\x14\x1d\x1a'
b'\x1f\x1e\x1d\x1a\x1c\x1c $.\' ",#\x1c\x1c(7),01444\x1f\''
b'\x9898_-282-\xff\xc0\x00\x0b\x08\x00\x01\x00\x01\x01\x11\x00'
b'\xff\xc4\x00\x1f\x00\x00\x01\x05\x01\x01\x01\x01\x01\x01\x00'
b'\x00\x00\x00\x00\x00\x00\x00\x01\x02\x03\x04\x05\x06\x07\x08'
b'\t\n\x0b\xff\xda\x00\x08\x01\x01\x00\x00?\x00\x7f\x00\xff\xd9'
)
return jpeg_data
@pytest.fixture
def sample_png_image():
"""Create a minimal valid PNG image for testing."""
# Create a minimal valid PNG programmatically using PIL
try:
from PIL import Image
from io import BytesIO
img = Image.new('RGB', (1, 1), color='white')
buffer = BytesIO()
img.save(buffer, format='PNG')
return buffer.getvalue()
except ImportError:
# Fallback to hardcoded PNG if PIL not available
# This is a valid minimal PNG that should be detected correctly
png_data = (
b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00'
b'\x00\x01\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDATx'
b'\x9cc\xf8\xcf\xc0\x00\x00\x00\x03\x00\x01\xfc\x83\t\xbf\x00'
b'\x00\x00\x00IEND\xaeB`\x82'
)
return png_data
class TestStaticFileServing:
"""Test static file serving for /images/ directory."""
def test_images_directory_exists_on_startup(self):
"""Verify /images directory is created on startup."""
assert IMAGES_ROOT.exists(), f"Images directory {IMAGES_ROOT} should exist"
assert IMAGES_ROOT.is_dir(), f"{IMAGES_ROOT} should be a directory"
def test_serve_uploaded_jpeg_image(self, client, sample_jpeg_image):
"""Test serving a JPEG image uploaded to /images/."""
# Save image directly using image storage utility
image_path = save_image(
file_bytes=sample_jpeg_image,
category="test_category",
filename_base="test_image",
variant="original"
)
# Request the image via static file mount
response = client.get(image_path)
# Verify response
assert response.status_code == 200, f"Expected 200, got {response.status_code}"
assert response.headers["content-type"] == "image/jpeg"
assert response.content == sample_jpeg_image
def test_serve_uploaded_png_image(self, client, sample_png_image):
"""Test serving a PNG image uploaded to /images/."""
image_path = save_image(
file_bytes=sample_png_image,
category="test_category",
filename_base="test_png",
variant="original"
)
response = client.get(image_path)
# Note: save_image always saves with .jpg extension
assert response.status_code == 200
# Content is served back correctly (extension determines MIME type)
assert response.content == sample_png_image
def test_serve_thumbnail_variant(self, client, sample_jpeg_image):
"""Test serving thumbnail variant of an image."""
image_path = save_image(
file_bytes=sample_jpeg_image,
category="test_category",
filename_base="test_image",
variant="thumb"
)
response = client.get(image_path)
assert response.status_code == 200
assert response.headers["content-type"] == "image/jpeg"
def test_404_for_nonexistent_image(self, client):
"""Test 404 response for non-existent image."""
response = client.get("/images/nonexistent/missing.jpg")
assert response.status_code == 404
def test_directory_traversal_protection(self, client):
"""Test that directory traversal attempts are blocked."""
# Try various directory traversal patterns
traversal_paths = [
"/images/../../../etc/passwd",
"/images/test/../../etc/passwd",
"/images/test/..%2F..%2Fetc%2Fpasswd",
]
for path in traversal_paths:
response = client.get(path)
# Should either be 404 or 400, not 200
assert response.status_code in [400, 404], \
f"Path {path} should not be accessible, got {response.status_code}"
def test_serve_multiple_categories(self, client, sample_jpeg_image, sample_png_image):
"""Test serving images from multiple categories."""
# Save images to different categories
jpeg_path = save_image(
file_bytes=sample_jpeg_image,
category="memory",
filename_base="ddr4_stick",
variant="original"
)
png_path = save_image(
file_bytes=sample_png_image,
category="networking",
filename_base="sfp_transceiver",
variant="original"
)
# Verify both can be served
jpeg_response = client.get(jpeg_path)
png_response = client.get(png_path)
assert jpeg_response.status_code == 200
assert jpeg_response.headers["content-type"] == "image/jpeg"
assert png_response.status_code == 200
# Both saved as .jpg due to save_image implementation
assert png_response.headers["content-type"] in ["image/jpeg", "image/jpg"]
def test_content_matches_uploaded_file(self, client, sample_jpeg_image):
"""Verify that served content exactly matches uploaded file."""
image_path = save_image(
file_bytes=sample_jpeg_image,
category="test_category",
filename_base="test_image",
variant="original"
)
response = client.get(image_path)
# Content-length should match
assert len(response.content) == len(sample_jpeg_image)
# Content should match byte-for-byte
assert response.content == sample_jpeg_image
def test_mime_type_auto_detection_jpeg(self, client, sample_jpeg_image):
"""Test MIME type is auto-detected correctly for JPEG."""
image_path = save_image(
file_bytes=sample_jpeg_image,
category="test_category",
filename_base="jpeg_file",
variant="original"
)
response = client.get(image_path)
assert response.headers["content-type"] in ["image/jpeg", "image/jpg"]
def test_mime_type_auto_detection_all_saved_as_jpeg(self, client, sample_png_image):
"""Test MIME type detection for images saved as JPEG (all images saved as .jpg)."""
# Note: save_image always saves with .jpg extension regardless of input format
image_path = save_image(
file_bytes=sample_png_image,
category="test_category",
filename_base="png_file",
variant="original"
)
response = client.get(image_path)
# File extension is .jpg, so MIME type will be image/jpeg
assert response.headers["content-type"] in ["image/jpeg", "image/jpg"]
def test_original_and_thumbnail_variants_accessible(self, client, sample_jpeg_image):
"""Test both original and thumbnail variants are accessible."""
original_path = save_image(
file_bytes=sample_jpeg_image,
category="test_category",
filename_base="multi_variant",
variant="original"
)
thumb_path = save_image(
file_bytes=sample_jpeg_image,
category="test_category",
filename_base="multi_variant",
variant="thumb"
)
# Both should be accessible
original_response = client.get(original_path)
thumb_response = client.get(thumb_path)
assert original_response.status_code == 200
assert thumb_response.status_code == 200
def test_images_served_with_correct_path_structure(self, client, sample_jpeg_image):
"""Test images are served at /images/{category}/{filename} path."""
image_path = save_image(
file_bytes=sample_jpeg_image,
category="storage",
filename_base="ssd_drive",
variant="original"
)
# Path should be /images/storage/ssd_drive_original.jpg
assert image_path.startswith("/images/"), "Path should start with /images/"
assert "storage" in image_path, "Path should include category"
response = client.get(image_path)
assert response.status_code == 200