709 lines
27 KiB
Python
709 lines
27 KiB
Python
|
"""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
|