"""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<>endobj\n2 0 obj<>endobj\n3 0 obj<>endobj\nxref\n0 4\n0000000000 65535 f \n0000000009 00000 n \n0000000052 00000 n \n0000000101 00000 n \ntrailer<>\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