feat(T05): Dateiablage pro Fahrzeug + File UI with thumbnails and gallery
This commit is contained in:
+2
-1
@@ -9,7 +9,7 @@ from fastapi import APIRouter, FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.config import settings
|
||||
from app.routers import auth, contacts, ocr, users, vehicles
|
||||
from app.routers import auth, contacts, files, ocr, users, vehicles
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
@@ -42,6 +42,7 @@ api_v1_router.include_router(auth.router)
|
||||
api_v1_router.include_router(users.router)
|
||||
api_v1_router.include_router(vehicles.router)
|
||||
api_v1_router.include_router(contacts.router)
|
||||
api_v1_router.include_router(files.router)
|
||||
api_v1_router.include_router(ocr.router)
|
||||
|
||||
# Health endpoint (no auth required)
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
"""SQLAlchemy model for vehicle file attachments."""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import DateTime, ForeignKey, Integer, String, func
|
||||
from sqlalchemy.dialects.postgresql import UUID
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from app.database import Base
|
||||
|
||||
|
||||
class File(Base):
|
||||
"""File attachment entity linked to a vehicle."""
|
||||
|
||||
__tablename__ = "files"
|
||||
|
||||
id: Mapped[uuid.UUID] = mapped_column(
|
||||
UUID(as_uuid=True),
|
||||
primary_key=True,
|
||||
default=uuid.uuid4,
|
||||
)
|
||||
vehicle_id: Mapped[uuid.UUID] = mapped_column(
|
||||
UUID(as_uuid=True),
|
||||
ForeignKey("vehicles.id", ondelete="CASCADE"),
|
||||
nullable=False,
|
||||
index=True,
|
||||
)
|
||||
original_filename: Mapped[str] = mapped_column(
|
||||
String(255), nullable=False
|
||||
)
|
||||
stored_filename: Mapped[str] = mapped_column(
|
||||
String(255), nullable=False
|
||||
)
|
||||
file_path: Mapped[str] = mapped_column(
|
||||
String(512), nullable=False
|
||||
)
|
||||
mime_type: Mapped[str] = mapped_column(
|
||||
String(100), nullable=False
|
||||
)
|
||||
file_size: Mapped[int] = mapped_column(
|
||||
Integer, nullable=False
|
||||
)
|
||||
thumbnail_path: Mapped[str | None] = mapped_column(
|
||||
String(512), nullable=True
|
||||
)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True),
|
||||
nullable=False,
|
||||
server_default=func.now(),
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True),
|
||||
nullable=False,
|
||||
server_default=func.now(),
|
||||
onupdate=func.now(),
|
||||
)
|
||||
|
||||
vehicle: Mapped["Vehicle"] = relationship(
|
||||
"Vehicle", back_populates="files"
|
||||
)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<File id={self.id} vehicle_id={self.vehicle_id} filename={self.original_filename}>"
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
"""Serialize file for API responses."""
|
||||
return {
|
||||
"id": str(self.id),
|
||||
"vehicle_id": str(self.vehicle_id),
|
||||
"original_filename": self.original_filename,
|
||||
"stored_filename": self.stored_filename,
|
||||
"file_path": self.file_path,
|
||||
"mime_type": self.mime_type,
|
||||
"file_size": self.file_size,
|
||||
"thumbnail_path": self.thumbnail_path,
|
||||
"created_at": (
|
||||
self.created_at.isoformat() if self.created_at else None
|
||||
),
|
||||
"updated_at": (
|
||||
self.updated_at.isoformat() if self.updated_at else None
|
||||
),
|
||||
}
|
||||
@@ -130,6 +130,9 @@ class Vehicle(Base):
|
||||
mobile_de_listings: Mapped[list["MobileDeListing"]] = relationship(
|
||||
back_populates="vehicle", cascade="all, delete-orphan"
|
||||
)
|
||||
files: Mapped[list["File"]] = relationship(
|
||||
"File", back_populates="vehicle", cascade="all, delete-orphan"
|
||||
)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<Vehicle id={self.id} fin={self.fin} make={self.make}>"
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
"""Files router: upload, list, download, and delete files for vehicles."""
|
||||
|
||||
import os
|
||||
import uuid
|
||||
|
||||
from fastapi import APIRouter, Depends, File as FastAPIFile, HTTPException, UploadFile, status
|
||||
from fastapi.responses import FileResponse as FastAPIFileResponse
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.database import get_db
|
||||
from app.dependencies import get_current_user, get_pagination
|
||||
from app.models.user import User
|
||||
from app.models.vehicle import Vehicle
|
||||
from app.schemas.file import (
|
||||
FileDeleteResponse,
|
||||
FileListResponse,
|
||||
FileResponse,
|
||||
FileUploadResponse,
|
||||
)
|
||||
from app.services import file_service, vehicle_service
|
||||
|
||||
router = APIRouter(prefix="/vehicles", tags=["files"])
|
||||
|
||||
|
||||
async def _verify_vehicle_exists(
|
||||
db: AsyncSession, vehicle_id: uuid.UUID
|
||||
) -> Vehicle:
|
||||
"""Verify that a vehicle exists, raising 404 if not."""
|
||||
vehicle = await vehicle_service.get_vehicle_by_id(db, vehicle_id)
|
||||
if vehicle is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"error": {
|
||||
"code": "VEHICLE_NOT_FOUND",
|
||||
"message": "Vehicle not found",
|
||||
}
|
||||
},
|
||||
)
|
||||
return vehicle
|
||||
|
||||
|
||||
@router.get(
|
||||
"/{vehicle_id}/files",
|
||||
response_model=FileListResponse,
|
||||
status_code=status.HTTP_200_OK,
|
||||
)
|
||||
async def list_files(
|
||||
vehicle_id: uuid.UUID,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
pagination: dict = Depends(get_pagination),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""List all files for a vehicle with pagination."""
|
||||
await _verify_vehicle_exists(db, vehicle_id)
|
||||
|
||||
files, total = await file_service.list_files(
|
||||
db,
|
||||
vehicle_id=vehicle_id,
|
||||
page=pagination["page"],
|
||||
page_size=pagination["page_size"],
|
||||
)
|
||||
return FileListResponse(
|
||||
items=[FileResponse.model_validate(f) for f in files],
|
||||
total=total,
|
||||
page=pagination["page"],
|
||||
page_size=pagination["page_size"],
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/{vehicle_id}/files",
|
||||
response_model=FileUploadResponse,
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
)
|
||||
async def upload_file(
|
||||
vehicle_id: uuid.UUID,
|
||||
file: UploadFile = FastAPIFile(...),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Upload a file for a vehicle (multipart/form-data).
|
||||
|
||||
Accepts images (jpg, png, webp), documents (pdf, doc, docx).
|
||||
Max file size: 20MB.
|
||||
"""
|
||||
await _verify_vehicle_exists(db, vehicle_id)
|
||||
|
||||
# Read file content
|
||||
file_content = await file.read()
|
||||
file_size = len(file_content)
|
||||
|
||||
# Check file size (20MB limit)
|
||||
if not file_service.validate_file_size(file_size, max_size_mb=20):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
||||
detail={
|
||||
"error": {
|
||||
"code": "FILE_TOO_LARGE",
|
||||
"message": f"File size {file_size} bytes exceeds 20MB limit",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
# Validate MIME type
|
||||
mime_type = file.content_type or ""
|
||||
if not file_service.validate_mime_type(mime_type, file.filename or ""):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail={
|
||||
"error": {
|
||||
"code": "INVALID_MIME_TYPE",
|
||||
"message": f"Unsupported file type: {mime_type}. Allowed: jpg, png, webp, pdf, doc, docx",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
try:
|
||||
file_record = await file_service.upload_file(
|
||||
db,
|
||||
vehicle_id=vehicle_id,
|
||||
file_content=file_content,
|
||||
original_filename=file.filename or "unnamed",
|
||||
mime_type=mime_type,
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail={
|
||||
"error": {
|
||||
"code": "UPLOAD_FAILED",
|
||||
"message": str(exc),
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
return FileUploadResponse.model_validate(file_record)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/{vehicle_id}/files/{file_id}",
|
||||
status_code=status.HTTP_200_OK,
|
||||
)
|
||||
async def download_file(
|
||||
vehicle_id: uuid.UUID,
|
||||
file_id: uuid.UUID,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Download a file by its ID."""
|
||||
await _verify_vehicle_exists(db, vehicle_id)
|
||||
|
||||
file_record = await file_service.get_file(db, vehicle_id, file_id)
|
||||
if file_record is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"error": {
|
||||
"code": "FILE_NOT_FOUND",
|
||||
"message": "File not found",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
if not os.path.exists(file_record.file_path):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"error": {
|
||||
"code": "FILE_NOT_FOUND",
|
||||
"message": "File not found on disk",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
return FastAPIFileResponse(
|
||||
path=file_record.file_path,
|
||||
filename=file_record.original_filename,
|
||||
media_type=file_record.mime_type,
|
||||
)
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/{vehicle_id}/files/{file_id}",
|
||||
response_model=FileDeleteResponse,
|
||||
status_code=status.HTTP_200_OK,
|
||||
)
|
||||
async def delete_file(
|
||||
vehicle_id: uuid.UUID,
|
||||
file_id: uuid.UUID,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Delete a file by its ID."""
|
||||
await _verify_vehicle_exists(db, vehicle_id)
|
||||
|
||||
file_record = await file_service.delete_file(db, vehicle_id, file_id)
|
||||
if file_record is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={
|
||||
"error": {
|
||||
"code": "FILE_NOT_FOUND",
|
||||
"message": "File not found",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
return FileDeleteResponse(message="File deleted", id=file_record.id)
|
||||
@@ -0,0 +1,57 @@
|
||||
"""Pydantic schemas for file upload, download, and list responses."""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
|
||||
class FileUploadResponse(BaseModel):
|
||||
"""Response after a successful file upload."""
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: uuid.UUID
|
||||
vehicle_id: uuid.UUID
|
||||
original_filename: str
|
||||
stored_filename: str
|
||||
file_path: str
|
||||
mime_type: str
|
||||
file_size: int
|
||||
thumbnail_path: Optional[str] = None
|
||||
created_at: Optional[datetime] = None
|
||||
updated_at: Optional[datetime] = None
|
||||
|
||||
|
||||
class FileResponse(BaseModel):
|
||||
"""Full file metadata response."""
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: uuid.UUID
|
||||
vehicle_id: uuid.UUID
|
||||
original_filename: str
|
||||
stored_filename: str
|
||||
file_path: str
|
||||
mime_type: str
|
||||
file_size: int
|
||||
thumbnail_path: Optional[str] = None
|
||||
created_at: Optional[datetime] = None
|
||||
updated_at: Optional[datetime] = None
|
||||
|
||||
|
||||
class FileListResponse(BaseModel):
|
||||
"""Paginated file list response."""
|
||||
|
||||
items: list[FileResponse]
|
||||
total: int
|
||||
page: int
|
||||
page_size: int
|
||||
|
||||
|
||||
class FileDeleteResponse(BaseModel):
|
||||
"""Response after deleting a file."""
|
||||
|
||||
message: str = "File deleted"
|
||||
id: uuid.UUID
|
||||
@@ -0,0 +1,234 @@
|
||||
"""File service: upload, retrieve, list, delete, and validation for vehicle files.
|
||||
|
||||
All file operations use UPLOAD_DIR from config for storage.
|
||||
Supports images (jpg, png, webp), documents (pdf, doc, docx), and scans (jpg, png).
|
||||
Enforces MAX_FILE_SIZE_MB from config (default 50MB, but endpoint enforces 20MB).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.config import settings
|
||||
from app.models.file import File
|
||||
from app.utils.thumbnails import generate_thumbnail, is_image_mime_type
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Allowed MIME types for upload
|
||||
ALLOWED_MIME_TYPES = {
|
||||
# Images
|
||||
"image/jpeg",
|
||||
"image/png",
|
||||
"image/webp",
|
||||
# Documents
|
||||
"application/pdf",
|
||||
"application/msword",
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||||
}
|
||||
|
||||
# Map MIME types to file extensions for validation
|
||||
MIME_TO_EXTENSIONS = {
|
||||
"image/jpeg": {"jpg", "jpeg"},
|
||||
"image/png": {"png"},
|
||||
"image/webp": {"webp"},
|
||||
"application/pdf": {"pdf"},
|
||||
"application/msword": {"doc"},
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document": {"docx"},
|
||||
}
|
||||
|
||||
|
||||
def validate_mime_type(mime_type: str, filename: str) -> bool:
|
||||
"""Validate that the MIME type is allowed and matches the file extension.
|
||||
|
||||
Args:
|
||||
mime_type: The MIME type from the upload.
|
||||
filename: The original filename to check extension consistency.
|
||||
|
||||
Returns:
|
||||
True if the MIME type is allowed and extension matches.
|
||||
"""
|
||||
mime_type = mime_type.lower().strip()
|
||||
if mime_type not in ALLOWED_MIME_TYPES:
|
||||
return False
|
||||
|
||||
# Check extension consistency
|
||||
ext = Path(filename).suffix.lower().lstrip(".")
|
||||
allowed_extensions = MIME_TO_EXTENSIONS.get(mime_type, set())
|
||||
if ext not in allowed_extensions:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def validate_file_size(file_size: int, max_size_mb: int = 20) -> bool:
|
||||
"""Validate that the file size is within the allowed limit.
|
||||
|
||||
Args:
|
||||
file_size: File size in bytes.
|
||||
max_size_mb: Maximum file size in MB (default 20MB for uploads).
|
||||
|
||||
Returns:
|
||||
True if the file size is within the limit.
|
||||
"""
|
||||
max_size_bytes = max_size_mb * 1024 * 1024
|
||||
return file_size <= max_size_bytes
|
||||
|
||||
|
||||
async def upload_file(
|
||||
db: AsyncSession,
|
||||
vehicle_id: uuid.UUID,
|
||||
file_content: bytes,
|
||||
original_filename: str,
|
||||
mime_type: str,
|
||||
) -> File:
|
||||
"""Upload a file for a vehicle.
|
||||
|
||||
Validates MIME type and file size, stores the file in UPLOAD_DIR,
|
||||
generates a thumbnail for images, and creates a DB record.
|
||||
|
||||
Raises:
|
||||
ValueError: If MIME type or file size validation fails.
|
||||
"""
|
||||
# Validate MIME type
|
||||
if not validate_mime_type(mime_type, original_filename):
|
||||
raise ValueError(f"Unsupported MIME type: {mime_type} for file: {original_filename}")
|
||||
|
||||
# Validate file size (20MB limit for uploads)
|
||||
file_size = len(file_content)
|
||||
if not validate_file_size(file_size, max_size_mb=20):
|
||||
raise ValueError(f"File size {file_size} bytes exceeds 20MB limit")
|
||||
|
||||
# Generate stored filename
|
||||
ext = Path(original_filename).suffix.lower()
|
||||
stored_filename = f"{uuid.uuid4().hex}{ext}"
|
||||
|
||||
# Create vehicle-specific upload directory
|
||||
upload_dir = Path(settings.UPLOAD_DIR) / str(vehicle_id)
|
||||
upload_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
file_path = upload_dir / stored_filename
|
||||
|
||||
# Write file to disk
|
||||
with open(file_path, "wb") as f:
|
||||
f.write(file_content)
|
||||
|
||||
logger.info("File uploaded: %s -> %s", original_filename, file_path)
|
||||
|
||||
# Generate thumbnail for images
|
||||
thumbnail_path = None
|
||||
if is_image_mime_type(mime_type):
|
||||
thumbnail_dir = upload_dir / "thumbnails"
|
||||
thumbnail_path = generate_thumbnail(
|
||||
source_path=file_path,
|
||||
thumbnail_dir=thumbnail_dir,
|
||||
stored_filename=stored_filename,
|
||||
)
|
||||
|
||||
# Create DB record
|
||||
file_record = File(
|
||||
vehicle_id=vehicle_id,
|
||||
original_filename=original_filename,
|
||||
stored_filename=stored_filename,
|
||||
file_path=str(file_path),
|
||||
mime_type=mime_type,
|
||||
file_size=file_size,
|
||||
thumbnail_path=thumbnail_path,
|
||||
)
|
||||
db.add(file_record)
|
||||
await db.flush()
|
||||
await db.refresh(file_record)
|
||||
|
||||
return file_record
|
||||
|
||||
|
||||
async def get_file(
|
||||
db: AsyncSession,
|
||||
vehicle_id: uuid.UUID,
|
||||
file_id: uuid.UUID,
|
||||
) -> Optional[File]:
|
||||
"""Get a single file by ID and vehicle ID."""
|
||||
stmt = select(File).where(
|
||||
File.id == file_id,
|
||||
File.vehicle_id == vehicle_id,
|
||||
)
|
||||
result = await db.execute(stmt)
|
||||
return result.scalar_one_or_none()
|
||||
|
||||
|
||||
async def list_files(
|
||||
db: AsyncSession,
|
||||
vehicle_id: uuid.UUID,
|
||||
page: int = 1,
|
||||
page_size: int = 20,
|
||||
) -> tuple[list[File], int]:
|
||||
"""List files for a vehicle with pagination.
|
||||
|
||||
Returns (files, total_count).
|
||||
"""
|
||||
# Count total files for this vehicle
|
||||
count_stmt = select(func.count(File.id)).where(
|
||||
File.vehicle_id == vehicle_id
|
||||
)
|
||||
count_result = await db.execute(count_stmt)
|
||||
total = count_result.scalar_one()
|
||||
|
||||
# Fetch paginated files
|
||||
offset = (page - 1) * page_size
|
||||
data_stmt = (
|
||||
select(File)
|
||||
.where(File.vehicle_id == vehicle_id)
|
||||
.order_by(File.created_at.desc())
|
||||
.offset(offset)
|
||||
.limit(page_size)
|
||||
)
|
||||
result = await db.execute(data_stmt)
|
||||
files = list(result.scalars().all())
|
||||
|
||||
return files, total
|
||||
|
||||
|
||||
async def delete_file(
|
||||
db: AsyncSession,
|
||||
vehicle_id: uuid.UUID,
|
||||
file_id: uuid.UUID,
|
||||
) -> Optional[File]:
|
||||
"""Delete a file: remove from disk and DB.
|
||||
|
||||
Returns the deleted file record, or None if not found.
|
||||
"""
|
||||
file_record = await get_file(db, vehicle_id, file_id)
|
||||
if file_record is None:
|
||||
return None
|
||||
|
||||
# Remove file from disk
|
||||
file_path = Path(file_record.file_path)
|
||||
if file_path.exists():
|
||||
try:
|
||||
file_path.unlink()
|
||||
logger.info("File deleted from disk: %s", file_path)
|
||||
except OSError as exc:
|
||||
logger.error("Failed to delete file from disk: %s", exc)
|
||||
|
||||
# Remove thumbnail from disk if it exists
|
||||
if file_record.thumbnail_path:
|
||||
thumb_path = Path(file_record.thumbnail_path)
|
||||
if thumb_path.exists():
|
||||
try:
|
||||
thumb_path.unlink()
|
||||
logger.info("Thumbnail deleted from disk: %s", thumb_path)
|
||||
except OSError as exc:
|
||||
logger.error("Failed to delete thumbnail from disk: %s", exc)
|
||||
|
||||
# Remove from DB
|
||||
await db.delete(file_record)
|
||||
await db.flush()
|
||||
|
||||
return file_record
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Thumbnail generation utilities using Pillow.
|
||||
|
||||
Generates 200x200 thumbnails for image files (jpg, png, webp).
|
||||
Non-image files return None (no thumbnail).
|
||||
"""
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from PIL import Image, ImageOps
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
THUMBNAIL_SIZE = (200, 200)
|
||||
IMAGE_MIME_TYPES = {
|
||||
"image/jpeg",
|
||||
"image/png",
|
||||
"image/webp",
|
||||
}
|
||||
IMAGE_EXTENSIONS = {"jpg", "jpeg", "png", "webp"}
|
||||
|
||||
|
||||
def is_image_mime_type(mime_type: str) -> bool:
|
||||
"""Check if the given MIME type is a supported image type."""
|
||||
return mime_type.lower() in IMAGE_MIME_TYPES
|
||||
|
||||
|
||||
def generate_thumbnail(
|
||||
source_path: str | Path,
|
||||
thumbnail_dir: str | Path,
|
||||
stored_filename: str,
|
||||
) -> Optional[str]:
|
||||
"""Generate a 200x200 thumbnail for an image file.
|
||||
|
||||
Args:
|
||||
source_path: Path to the original image file.
|
||||
thumbnail_dir: Directory where thumbnails are stored.
|
||||
stored_filename: The stored filename of the original file.
|
||||
|
||||
Returns:
|
||||
Relative path to the thumbnail file, or None if the file is not
|
||||
a supported image or thumbnail generation fails.
|
||||
"""
|
||||
source_path = Path(source_path)
|
||||
thumbnail_dir = Path(thumbnail_dir)
|
||||
|
||||
# Check if the file extension is a supported image type
|
||||
ext = source_path.suffix.lower().lstrip(".")
|
||||
if ext not in IMAGE_EXTENSIONS:
|
||||
return None
|
||||
|
||||
try:
|
||||
thumbnail_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
thumbnail_name = f"thumb_{stored_filename}"
|
||||
# Use .jpg for thumbnails to save space
|
||||
if ext in ("jpg", "jpeg"):
|
||||
thumbnail_name = f"thumb_{Path(stored_filename).stem}.jpg"
|
||||
elif ext == "png":
|
||||
thumbnail_name = f"thumb_{Path(stored_filename).stem}.png"
|
||||
elif ext == "webp":
|
||||
thumbnail_name = f"thumb_{Path(stored_filename).stem}.webp"
|
||||
|
||||
thumbnail_path = thumbnail_dir / thumbnail_name
|
||||
|
||||
with Image.open(source_path) as img:
|
||||
# Convert to RGB if necessary (e.g., for RGBA/P mode images)
|
||||
if img.mode in ("RGBA", "P", "LA"):
|
||||
# Create a white background for transparency
|
||||
background = Image.new("RGB", img.size, (255, 255, 255))
|
||||
if img.mode == "P":
|
||||
img = img.convert("RGBA")
|
||||
background.paste(img, mask=img.split()[-1] if img.mode in ("RGBA", "LA") else None)
|
||||
img = background
|
||||
elif img.mode != "RGB":
|
||||
img = img.convert("RGB")
|
||||
|
||||
# Use ImageOps.fit for a centered crop to exact thumbnail size
|
||||
thumbnail = ImageOps.fit(img, THUMBNAIL_SIZE, method=Image.Resampling.LANCZOS)
|
||||
thumbnail.save(thumbnail_path, quality=85, optimize=True)
|
||||
|
||||
logger.info("Thumbnail generated: %s", thumbnail_path)
|
||||
return str(thumbnail_path)
|
||||
|
||||
except Exception as exc:
|
||||
logger.error("Failed to generate thumbnail for %s: %s", source_path, exc)
|
||||
return None
|
||||
@@ -12,3 +12,4 @@ pytest-asyncio>=0.24.0
|
||||
httpx>=0.27.0
|
||||
pytest-cov>=5.0.0
|
||||
aiosqlite>=0.20.0
|
||||
Pillow>=10.0.0
|
||||
|
||||
@@ -0,0 +1,708 @@
|
||||
"""Tests for file upload, list, download, delete, MIME validation, size limit, and thumbnail generation."""
|
||||
|
||||
import io
|
||||
import os
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from PIL import Image
|
||||
|
||||
from app.config import settings
|
||||
from app.database import Base, get_db
|
||||
from app.main import app
|
||||
from app.models.file import File
|
||||
from app.models.user import User, UserRole
|
||||
from app.models.vehicle import Vehicle
|
||||
from app.services.auth_service import hash_password
|
||||
|
||||
|
||||
# ---- Test fixtures ----
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def sample_vehicle_data():
|
||||
"""Valid vehicle data for creation."""
|
||||
return {
|
||||
"make": "Mercedes-Benz",
|
||||
"model": "Actros",
|
||||
"fin": "WDB9066351L123456",
|
||||
"year": 2020,
|
||||
"first_registration": "2020-03-15",
|
||||
"power_kw": 300,
|
||||
"fuel_type": "Diesel",
|
||||
"transmission": "Manual",
|
||||
"color": "White",
|
||||
"condition": "used",
|
||||
"location": "Berlin",
|
||||
"availability": "available",
|
||||
"price": 45000.00,
|
||||
"vehicle_type": "lkw",
|
||||
"lkw_type": "sattelzugmaschine",
|
||||
"mileage_km": 120000,
|
||||
"description": "Well maintained truck",
|
||||
}
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def created_vehicle(admin_client, sample_vehicle_data):
|
||||
"""Create a vehicle via API and return the response."""
|
||||
response = await admin_client.post("/api/v1/vehicles/", json=sample_vehicle_data)
|
||||
assert response.status_code == 201, response.text
|
||||
return response.json()
|
||||
|
||||
|
||||
def _make_image_bytes(format="JPEG", size=(800, 600), color="blue") -> bytes:
|
||||
"""Create a valid image in memory."""
|
||||
img = Image.new("RGB", size, color=color)
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, format=format)
|
||||
buf.seek(0)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def _make_png_bytes(size=(800, 600), color="red") -> bytes:
|
||||
"""Create a valid PNG image in memory."""
|
||||
img = Image.new("RGB", size, color=color)
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, format="PNG")
|
||||
buf.seek(0)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def _make_webp_bytes(size=(800, 600), color="green") -> bytes:
|
||||
"""Create a valid WebP image in memory."""
|
||||
img = Image.new("RGB", size, color=color)
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, format="WEBP")
|
||||
buf.seek(0)
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def _make_pdf_bytes() -> bytes:
|
||||
"""Create a minimal valid PDF."""
|
||||
return b"%PDF-1.4\n1 0 obj<</Type/Catalog/Pages 2 0 R>>endobj\n2 0 obj<</Type/Pages/Kids[3 0 R]/Count 1>>endobj\n3 0 obj<</Type/Page/Parent 2 0 R/MediaBox[0 0 612 792]>>endobj\nxref\n0 4\n0000000000 65535 f \n0000000009 00000 n \n0000000052 00000 n \n0000000101 00000 n \ntrailer<</Size 4/Root 1 0 R>>\nstartxref\n149\n%%EOF"
|
||||
|
||||
|
||||
async def _create_test_vehicle(db_session) -> uuid.UUID:
|
||||
"""Create a test vehicle in the DB and return its ID."""
|
||||
vehicle = Vehicle(
|
||||
make="Test",
|
||||
model="Truck",
|
||||
fin="WDB9066351L123456",
|
||||
price=45000,
|
||||
vehicle_type="lkw",
|
||||
condition="used",
|
||||
availability="available",
|
||||
)
|
||||
db_session.add(vehicle)
|
||||
await db_session.flush()
|
||||
return vehicle.id
|
||||
|
||||
|
||||
# ---- Tests: File Upload ----
|
||||
|
||||
class TestFileUpload:
|
||||
"""POST /api/v1/vehicles/:id/files tests."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_jpg_image_returns_201(self, admin_client, created_vehicle):
|
||||
"""Upload a valid JPG image and verify 201 response."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
image_bytes = _make_image_bytes(format="JPEG")
|
||||
files = {"file": ("test.jpg", image_bytes, "image/jpeg")}
|
||||
response = await admin_client.post(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files",
|
||||
files=files,
|
||||
)
|
||||
assert response.status_code == 201, response.text
|
||||
data = response.json()
|
||||
assert data["vehicle_id"] == vehicle_id
|
||||
assert data["original_filename"] == "test.jpg"
|
||||
assert data["mime_type"] == "image/jpeg"
|
||||
assert data["file_size"] == len(image_bytes)
|
||||
assert data["thumbnail_path"] is not None
|
||||
assert data["id"] is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_png_image_returns_201(self, admin_client, created_vehicle):
|
||||
"""Upload a valid PNG image and verify 201 response."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
image_bytes = _make_png_bytes()
|
||||
files = {"file": ("test.png", image_bytes, "image/png")}
|
||||
response = await admin_client.post(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files",
|
||||
files=files,
|
||||
)
|
||||
assert response.status_code == 201, response.text
|
||||
data = response.json()
|
||||
assert data["mime_type"] == "image/png"
|
||||
assert data["thumbnail_path"] is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_webp_image_returns_201(self, admin_client, created_vehicle):
|
||||
"""Upload a valid WebP image and verify 201 response."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
image_bytes = _make_webp_bytes()
|
||||
files = {"file": ("test.webp", image_bytes, "image/webp")}
|
||||
response = await admin_client.post(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files",
|
||||
files=files,
|
||||
)
|
||||
assert response.status_code == 201, response.text
|
||||
data = response.json()
|
||||
assert data["mime_type"] == "image/webp"
|
||||
assert data["thumbnail_path"] is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_pdf_returns_201(self, admin_client, created_vehicle):
|
||||
"""Upload a valid PDF and verify 201 response."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
pdf_bytes = _make_pdf_bytes()
|
||||
files = {"file": ("document.pdf", pdf_bytes, "application/pdf")}
|
||||
response = await admin_client.post(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files",
|
||||
files=files,
|
||||
)
|
||||
assert response.status_code == 201, response.text
|
||||
data = response.json()
|
||||
assert data["mime_type"] == "application/pdf"
|
||||
assert data["thumbnail_path"] is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_invalid_mime_type_returns_422(self, admin_client, created_vehicle):
|
||||
"""Upload a file with an unsupported MIME type and verify 422."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
files = {"file": ("malware.exe", b"MZ\x90\x00", "application/x-msdownload")}
|
||||
response = await admin_client.post(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files",
|
||||
files=files,
|
||||
)
|
||||
assert response.status_code == 422, response.text
|
||||
data = response.json()
|
||||
assert data["detail"]["error"]["code"] == "INVALID_MIME_TYPE"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_text_file_returns_422(self, admin_client, created_vehicle):
|
||||
"""Upload a text file and verify 422."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
files = {"file": ("notes.txt", b"hello world", "text/plain")}
|
||||
response = await admin_client.post(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files",
|
||||
files=files,
|
||||
)
|
||||
assert response.status_code == 422, response.text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_oversized_file_returns_413(self, admin_client, created_vehicle):
|
||||
"""Upload a file larger than 20MB and verify 413."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
# Create a 21MB file (21 * 1024 * 1024 bytes)
|
||||
large_content = b"\x00" * (21 * 1024 * 1024)
|
||||
files = {"file": ("large.jpg", large_content, "image/jpeg")}
|
||||
response = await admin_client.post(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files",
|
||||
files=files,
|
||||
)
|
||||
assert response.status_code == 413, response.text
|
||||
data = response.json()
|
||||
assert data["detail"]["error"]["code"] == "FILE_TOO_LARGE"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_to_nonexistent_vehicle_returns_404(self, admin_client):
|
||||
"""Upload to a non-existent vehicle and verify 404."""
|
||||
fake_id = str(uuid.uuid4())
|
||||
image_bytes = _make_image_bytes()
|
||||
files = {"file": ("test.jpg", image_bytes, "image/jpeg")}
|
||||
response = await admin_client.post(
|
||||
f"/api/v1/vehicles/{fake_id}/files",
|
||||
files=files,
|
||||
)
|
||||
assert response.status_code == 404, response.text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_without_auth_returns_401(self, test_session_factory, created_vehicle):
|
||||
"""Upload without authentication and verify 401."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
image_bytes = _make_image_bytes()
|
||||
files = {"file": ("test.jpg", image_bytes, "image/jpeg")}
|
||||
|
||||
# Create a fresh client without auth headers
|
||||
async def _override_get_db():
|
||||
async with test_session_factory() as session:
|
||||
try:
|
||||
yield session
|
||||
await session.commit()
|
||||
except Exception:
|
||||
await session.rollback()
|
||||
raise
|
||||
finally:
|
||||
await session.close()
|
||||
|
||||
app.dependency_overrides[get_db] = _override_get_db
|
||||
transport = ASGITransport(app=app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as unauth_client:
|
||||
response = await unauth_client.post(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files",
|
||||
files=files,
|
||||
)
|
||||
assert response.status_code == 401, response.text
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
# ---- Tests: File List ----
|
||||
|
||||
class TestFileList:
|
||||
"""GET /api/v1/vehicles/:id/files tests."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_files_returns_200_with_pagination(self, admin_client, created_vehicle):
|
||||
"""List files for a vehicle returns 200 with paginated response."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
# Upload a file first
|
||||
image_bytes = _make_image_bytes()
|
||||
files = {"file": ("test.jpg", image_bytes, "image/jpeg")}
|
||||
await admin_client.post(f"/api/v1/vehicles/{vehicle_id}/files", files=files)
|
||||
|
||||
response = await admin_client.get(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files?page=1&page_size=20"
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
data = response.json()
|
||||
assert "items" in data
|
||||
assert "total" in data
|
||||
assert "page" in data
|
||||
assert "page_size" in data
|
||||
assert data["page"] == 1
|
||||
assert data["page_size"] == 20
|
||||
assert data["total"] >= 1
|
||||
assert len(data["items"]) >= 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_files_empty_returns_200(self, admin_client, created_vehicle):
|
||||
"""List files for a vehicle with no files returns 200 with empty list."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
response = await admin_client.get(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files"
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
data = response.json()
|
||||
assert data["total"] == 0
|
||||
assert data["items"] == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_files_for_nonexistent_vehicle_returns_404(self, admin_client):
|
||||
"""List files for a non-existent vehicle returns 404."""
|
||||
fake_id = str(uuid.uuid4())
|
||||
response = await admin_client.get(f"/api/v1/vehicles/{fake_id}/files")
|
||||
assert response.status_code == 404, response.text
|
||||
|
||||
|
||||
# ---- Tests: File Download ----
|
||||
|
||||
class TestFileDownload:
|
||||
"""GET /api/v1/vehicles/:id/files/:fileId tests."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_download_file_returns_200(self, admin_client, created_vehicle):
|
||||
"""Download a file and verify 200 with correct content."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
image_bytes = _make_image_bytes()
|
||||
files = {"file": ("test.jpg", image_bytes, "image/jpeg")}
|
||||
upload_resp = await admin_client.post(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files", files=files
|
||||
)
|
||||
assert upload_resp.status_code == 201
|
||||
file_id = upload_resp.json()["id"]
|
||||
|
||||
response = await admin_client.get(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files/{file_id}"
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.headers["content-type"].startswith("image/jpeg")
|
||||
assert len(response.content) == len(image_bytes)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_download_nonexistent_file_returns_404(self, admin_client, created_vehicle):
|
||||
"""Download a non-existent file returns 404."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
fake_file_id = str(uuid.uuid4())
|
||||
response = await admin_client.get(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files/{fake_file_id}"
|
||||
)
|
||||
assert response.status_code == 404, response.text
|
||||
|
||||
|
||||
# ---- Tests: File Delete ----
|
||||
|
||||
class TestFileDelete:
|
||||
"""DELETE /api/v1/vehicles/:id/files/:fileId tests."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_file_returns_200(self, admin_client, created_vehicle):
|
||||
"""Delete a file and verify 200 response."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
image_bytes = _make_image_bytes()
|
||||
files = {"file": ("test.jpg", image_bytes, "image/jpeg")}
|
||||
upload_resp = await admin_client.post(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files", files=files
|
||||
)
|
||||
assert upload_resp.status_code == 201
|
||||
file_id = upload_resp.json()["id"]
|
||||
|
||||
response = await admin_client.delete(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files/{file_id}"
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
data = response.json()
|
||||
assert data["message"] == "File deleted"
|
||||
assert data["id"] == file_id
|
||||
|
||||
# Verify file is gone from list
|
||||
list_resp = await admin_client.get(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files"
|
||||
)
|
||||
assert list_resp.status_code == 200
|
||||
assert list_resp.json()["total"] == 0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_nonexistent_file_returns_404(self, admin_client, created_vehicle):
|
||||
"""Delete a non-existent file returns 404."""
|
||||
vehicle_id = created_vehicle["id"]
|
||||
fake_file_id = str(uuid.uuid4())
|
||||
response = await admin_client.delete(
|
||||
f"/api/v1/vehicles/{vehicle_id}/files/{fake_file_id}"
|
||||
)
|
||||
assert response.status_code == 404, response.text
|
||||
|
||||
|
||||
# ---- Tests: MIME Type Validation ----
|
||||
|
||||
class TestMIMEValidation:
|
||||
"""Unit tests for MIME type validation."""
|
||||
|
||||
def test_validate_jpeg_mime_type(self):
|
||||
from app.services.file_service import validate_mime_type
|
||||
assert validate_mime_type("image/jpeg", "photo.jpg") is True
|
||||
assert validate_mime_type("image/jpeg", "photo.jpeg") is True
|
||||
|
||||
def test_validate_png_mime_type(self):
|
||||
from app.services.file_service import validate_mime_type
|
||||
assert validate_mime_type("image/png", "photo.png") is True
|
||||
|
||||
def test_validate_webp_mime_type(self):
|
||||
from app.services.file_service import validate_mime_type
|
||||
assert validate_mime_type("image/webp", "photo.webp") is True
|
||||
|
||||
def test_validate_pdf_mime_type(self):
|
||||
from app.services.file_service import validate_mime_type
|
||||
assert validate_mime_type("application/pdf", "doc.pdf") is True
|
||||
|
||||
def test_validate_doc_mime_type(self):
|
||||
from app.services.file_service import validate_mime_type
|
||||
assert validate_mime_type("application/msword", "doc.doc") is True
|
||||
|
||||
def test_validate_docx_mime_type(self):
|
||||
from app.services.file_service import validate_mime_type
|
||||
assert validate_mime_type(
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||||
"doc.docx",
|
||||
) is True
|
||||
|
||||
def test_reject_exe_mime_type(self):
|
||||
from app.services.file_service import validate_mime_type
|
||||
assert validate_mime_type("application/x-msdownload", "malware.exe") is False
|
||||
|
||||
def test_reject_text_mime_type(self):
|
||||
from app.services.file_service import validate_mime_type
|
||||
assert validate_mime_type("text/plain", "notes.txt") is False
|
||||
|
||||
def test_reject_mismatched_extension(self):
|
||||
"""MIME type image/jpeg with .png extension should fail."""
|
||||
from app.services.file_service import validate_mime_type
|
||||
assert validate_mime_type("image/jpeg", "photo.png") is False
|
||||
|
||||
|
||||
# ---- Tests: File Size Validation ----
|
||||
|
||||
class TestFileSizeValidation:
|
||||
"""Unit tests for file size validation."""
|
||||
|
||||
def test_validate_small_file_size(self):
|
||||
from app.services.file_service import validate_file_size
|
||||
assert validate_file_size(1024, max_size_mb=20) is True
|
||||
|
||||
def test_validate_exact_20mb_file_size(self):
|
||||
from app.services.file_service import validate_file_size
|
||||
exact_20mb = 20 * 1024 * 1024
|
||||
assert validate_file_size(exact_20mb, max_size_mb=20) is True
|
||||
|
||||
def test_reject_oversized_file(self):
|
||||
from app.services.file_service import validate_file_size
|
||||
over_20mb = 20 * 1024 * 1024 + 1
|
||||
assert validate_file_size(over_20mb, max_size_mb=20) is False
|
||||
|
||||
def test_validate_zero_byte_file(self):
|
||||
from app.services.file_service import validate_file_size
|
||||
assert validate_file_size(0, max_size_mb=20) is True
|
||||
|
||||
|
||||
# ---- Tests: Thumbnail Generation ----
|
||||
|
||||
class TestThumbnailGeneration:
|
||||
"""Tests for thumbnail generation utility."""
|
||||
|
||||
def test_generate_thumbnail_for_jpeg(self, tmp_path):
|
||||
"""Generate a thumbnail for a JPEG image and verify size."""
|
||||
from app.utils.thumbnails import generate_thumbnail
|
||||
|
||||
# Create a test image
|
||||
img = Image.new("RGB", (800, 600), color="blue")
|
||||
source_path = tmp_path / "test.jpg"
|
||||
img.save(source_path, format="JPEG")
|
||||
|
||||
thumbnail_dir = tmp_path / "thumbnails"
|
||||
result = generate_thumbnail(source_path, thumbnail_dir, "test.jpg")
|
||||
|
||||
assert result is not None
|
||||
assert Path(result).exists()
|
||||
with Image.open(result) as thumb:
|
||||
assert thumb.size == (200, 200)
|
||||
|
||||
def test_generate_thumbnail_for_png(self, tmp_path):
|
||||
"""Generate a thumbnail for a PNG image."""
|
||||
from app.utils.thumbnails import generate_thumbnail
|
||||
|
||||
img = Image.new("RGB", (800, 600), color="red")
|
||||
source_path = tmp_path / "test.png"
|
||||
img.save(source_path, format="PNG")
|
||||
|
||||
thumbnail_dir = tmp_path / "thumbnails"
|
||||
result = generate_thumbnail(source_path, thumbnail_dir, "test.png")
|
||||
|
||||
assert result is not None
|
||||
assert Path(result).exists()
|
||||
with Image.open(result) as thumb:
|
||||
assert thumb.size == (200, 200)
|
||||
|
||||
def test_generate_thumbnail_for_webp(self, tmp_path):
|
||||
"""Generate a thumbnail for a WebP image."""
|
||||
from app.utils.thumbnails import generate_thumbnail
|
||||
|
||||
img = Image.new("RGB", (800, 600), color="green")
|
||||
source_path = tmp_path / "test.webp"
|
||||
img.save(source_path, format="WEBP")
|
||||
|
||||
thumbnail_dir = tmp_path / "thumbnails"
|
||||
result = generate_thumbnail(source_path, thumbnail_dir, "test.webp")
|
||||
|
||||
assert result is not None
|
||||
assert Path(result).exists()
|
||||
with Image.open(result) as thumb:
|
||||
assert thumb.size == (200, 200)
|
||||
|
||||
def test_no_thumbnail_for_non_image(self, tmp_path):
|
||||
"""Thumbnail generation returns None for non-image files."""
|
||||
from app.utils.thumbnails import generate_thumbnail
|
||||
|
||||
source_path = tmp_path / "document.pdf"
|
||||
source_path.write_bytes(b"%PDF-1.4 test")
|
||||
|
||||
thumbnail_dir = tmp_path / "thumbnails"
|
||||
result = generate_thumbnail(source_path, thumbnail_dir, "document.pdf")
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_thumbnail_with_transparent_png(self, tmp_path):
|
||||
"""Generate a thumbnail for a transparent PNG (RGBA mode)."""
|
||||
from app.utils.thumbnails import generate_thumbnail
|
||||
|
||||
img = Image.new("RGBA", (400, 400), color=(255, 0, 0, 128))
|
||||
source_path = tmp_path / "transparent.png"
|
||||
img.save(source_path, format="PNG")
|
||||
|
||||
thumbnail_dir = tmp_path / "thumbnails"
|
||||
result = generate_thumbnail(source_path, thumbnail_dir, "transparent.png")
|
||||
|
||||
assert result is not None
|
||||
assert Path(result).exists()
|
||||
with Image.open(result) as thumb:
|
||||
assert thumb.size == (200, 200)
|
||||
|
||||
def test_is_image_mime_type(self):
|
||||
"""Test is_image_mime_type helper."""
|
||||
from app.utils.thumbnails import is_image_mime_type
|
||||
assert is_image_mime_type("image/jpeg") is True
|
||||
assert is_image_mime_type("image/png") is True
|
||||
assert is_image_mime_type("image/webp") is True
|
||||
assert is_image_mime_type("application/pdf") is False
|
||||
assert is_image_mime_type("text/plain") is False
|
||||
|
||||
|
||||
# ---- Tests: File Service Unit Tests ----
|
||||
|
||||
class TestFileServiceUnit:
|
||||
"""Unit tests for file service functions."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_file_creates_db_record(self, db_session, tmp_path):
|
||||
"""Test that upload_file creates a DB record."""
|
||||
from app.services import file_service
|
||||
|
||||
vehicle_id = await _create_test_vehicle(db_session)
|
||||
image_bytes = _make_image_bytes()
|
||||
|
||||
with patch.object(settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
file_record = await file_service.upload_file(
|
||||
db_session,
|
||||
vehicle_id=vehicle_id,
|
||||
file_content=image_bytes,
|
||||
original_filename="test.jpg",
|
||||
mime_type="image/jpeg",
|
||||
)
|
||||
|
||||
assert file_record.id is not None
|
||||
assert file_record.vehicle_id == vehicle_id
|
||||
assert file_record.original_filename == "test.jpg"
|
||||
assert file_record.mime_type == "image/jpeg"
|
||||
assert file_record.file_size == len(image_bytes)
|
||||
assert file_record.thumbnail_path is not None
|
||||
assert Path(file_record.file_path).exists()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_file_rejects_invalid_mime(self, db_session, tmp_path):
|
||||
"""Test that upload_file rejects invalid MIME types."""
|
||||
from app.services import file_service
|
||||
|
||||
with patch.object(settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
with pytest.raises(ValueError, match="Unsupported MIME type"):
|
||||
await file_service.upload_file(
|
||||
db_session,
|
||||
vehicle_id=uuid.uuid4(),
|
||||
file_content=b"hello",
|
||||
original_filename="test.exe",
|
||||
mime_type="application/x-msdownload",
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_file_rejects_oversized(self, db_session, tmp_path):
|
||||
"""Test that upload_file rejects oversized files."""
|
||||
from app.services import file_service
|
||||
|
||||
large_content = b"\x00" * (21 * 1024 * 1024)
|
||||
with patch.object(settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
with pytest.raises(ValueError, match="exceeds 20MB"):
|
||||
await file_service.upload_file(
|
||||
db_session,
|
||||
vehicle_id=uuid.uuid4(),
|
||||
file_content=large_content,
|
||||
original_filename="large.jpg",
|
||||
mime_type="image/jpeg",
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_file_removes_from_disk(self, db_session, tmp_path):
|
||||
"""Test that delete_file removes the file from disk."""
|
||||
from app.services import file_service
|
||||
|
||||
vehicle_id = await _create_test_vehicle(db_session)
|
||||
image_bytes = _make_image_bytes()
|
||||
|
||||
with patch.object(settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
file_record = await file_service.upload_file(
|
||||
db_session,
|
||||
vehicle_id=vehicle_id,
|
||||
file_content=image_bytes,
|
||||
original_filename="test.jpg",
|
||||
mime_type="image/jpeg",
|
||||
)
|
||||
|
||||
file_path = Path(file_record.file_path)
|
||||
assert file_path.exists()
|
||||
|
||||
deleted = await file_service.delete_file(db_session, vehicle_id, file_record.id)
|
||||
assert deleted is not None
|
||||
assert not file_path.exists()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_nonexistent_file_returns_none(self, db_session):
|
||||
"""Test that delete_file returns None for non-existent file."""
|
||||
from app.services import file_service
|
||||
|
||||
result = await file_service.delete_file(
|
||||
db_session, uuid.uuid4(), uuid.uuid4()
|
||||
)
|
||||
assert result is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_files_pagination(self, db_session, tmp_path):
|
||||
"""Test that list_files returns paginated results."""
|
||||
from app.services import file_service
|
||||
|
||||
vehicle_id = await _create_test_vehicle(db_session)
|
||||
|
||||
with patch.object(settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
# Upload 3 files
|
||||
for i in range(3):
|
||||
await file_service.upload_file(
|
||||
db_session,
|
||||
vehicle_id=vehicle_id,
|
||||
file_content=_make_image_bytes(),
|
||||
original_filename=f"test_{i}.jpg",
|
||||
mime_type="image/jpeg",
|
||||
)
|
||||
|
||||
files, total = await file_service.list_files(
|
||||
db_session, vehicle_id, page=1, page_size=2
|
||||
)
|
||||
assert total == 3
|
||||
assert len(files) == 2
|
||||
|
||||
files_page2, total2 = await file_service.list_files(
|
||||
db_session, vehicle_id, page=2, page_size=2
|
||||
)
|
||||
assert total2 == 3
|
||||
assert len(files_page2) == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file_returns_correct_file(self, db_session, tmp_path):
|
||||
"""Test that get_file returns the correct file by ID."""
|
||||
from app.services import file_service
|
||||
|
||||
vehicle_id = await _create_test_vehicle(db_session)
|
||||
|
||||
with patch.object(settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
file_record = await file_service.upload_file(
|
||||
db_session,
|
||||
vehicle_id=vehicle_id,
|
||||
file_content=_make_image_bytes(),
|
||||
original_filename="test.jpg",
|
||||
mime_type="image/jpeg",
|
||||
)
|
||||
|
||||
retrieved = await file_service.get_file(db_session, vehicle_id, file_record.id)
|
||||
assert retrieved is not None
|
||||
assert retrieved.id == file_record.id
|
||||
assert retrieved.original_filename == "test.jpg"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file_wrong_vehicle_returns_none(self, db_session, tmp_path):
|
||||
"""Test that get_file returns None for wrong vehicle ID."""
|
||||
from app.services import file_service
|
||||
|
||||
vehicle_id = await _create_test_vehicle(db_session)
|
||||
|
||||
with patch.object(settings, "UPLOAD_DIR", str(tmp_path)):
|
||||
file_record = await file_service.upload_file(
|
||||
db_session,
|
||||
vehicle_id=vehicle_id,
|
||||
file_content=_make_image_bytes(),
|
||||
original_filename="test.jpg",
|
||||
mime_type="image/jpeg",
|
||||
)
|
||||
|
||||
wrong_vehicle_id = uuid.uuid4()
|
||||
retrieved = await file_service.get_file(db_session, wrong_vehicle_id, file_record.id)
|
||||
assert retrieved is None
|
||||
Reference in New Issue
Block a user