feat: enhance messaging system with improved API and tests
- Refactor messaging API endpoints for admin, shop, and vendor - Add message-specific exceptions (ConversationNotFoundException, etc.) - Enhance messaging service with additional helper methods - Add comprehensive test fixtures for messaging - Add integration tests for admin and vendor messaging APIs - Add unit tests for messaging and attachment services 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
387
tests/unit/services/test_message_attachment_service.py
Normal file
387
tests/unit/services/test_message_attachment_service.py
Normal file
@@ -0,0 +1,387 @@
|
||||
# tests/unit/services/test_message_attachment_service.py
|
||||
"""
|
||||
Unit tests for MessageAttachmentService.
|
||||
"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi import UploadFile
|
||||
|
||||
from app.services.message_attachment_service import (
|
||||
ALLOWED_MIME_TYPES,
|
||||
DEFAULT_MAX_FILE_SIZE_MB,
|
||||
IMAGE_MIME_TYPES,
|
||||
MessageAttachmentService,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def attachment_service():
|
||||
"""Create a MessageAttachmentService instance with temp storage."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
yield MessageAttachmentService(storage_base=tmpdir)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_upload_file():
|
||||
"""Create a mock UploadFile."""
|
||||
|
||||
def _create_upload_file(
|
||||
content: bytes = b"test content",
|
||||
filename: str = "test.txt",
|
||||
content_type: str = "text/plain",
|
||||
):
|
||||
file = MagicMock(spec=UploadFile)
|
||||
file.filename = filename
|
||||
file.content_type = content_type
|
||||
file.read = AsyncMock(return_value=content)
|
||||
return file
|
||||
|
||||
return _create_upload_file
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestMessageAttachmentServiceValidation:
|
||||
"""Tests for file validation methods."""
|
||||
|
||||
def test_validate_file_type_allowed_image(self, attachment_service):
|
||||
"""Test image MIME types are allowed."""
|
||||
for mime_type in IMAGE_MIME_TYPES:
|
||||
assert attachment_service.validate_file_type(mime_type) is True
|
||||
|
||||
def test_validate_file_type_allowed_documents(self, attachment_service):
|
||||
"""Test document MIME types are allowed."""
|
||||
document_types = [
|
||||
"application/pdf",
|
||||
"application/msword",
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||||
"application/vnd.ms-excel",
|
||||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||||
]
|
||||
for mime_type in document_types:
|
||||
assert attachment_service.validate_file_type(mime_type) is True
|
||||
|
||||
def test_validate_file_type_allowed_others(self, attachment_service):
|
||||
"""Test other allowed MIME types."""
|
||||
other_types = ["application/zip", "text/plain", "text/csv"]
|
||||
for mime_type in other_types:
|
||||
assert attachment_service.validate_file_type(mime_type) is True
|
||||
|
||||
def test_validate_file_type_not_allowed(self, attachment_service):
|
||||
"""Test disallowed MIME types."""
|
||||
disallowed_types = [
|
||||
"application/javascript",
|
||||
"application/x-executable",
|
||||
"text/html",
|
||||
"video/mp4",
|
||||
"audio/mpeg",
|
||||
]
|
||||
for mime_type in disallowed_types:
|
||||
assert attachment_service.validate_file_type(mime_type) is False
|
||||
|
||||
def test_is_image_true(self, attachment_service):
|
||||
"""Test image detection for actual images."""
|
||||
for mime_type in IMAGE_MIME_TYPES:
|
||||
assert attachment_service.is_image(mime_type) is True
|
||||
|
||||
def test_is_image_false(self, attachment_service):
|
||||
"""Test image detection for non-images."""
|
||||
non_images = ["application/pdf", "text/plain", "application/zip"]
|
||||
for mime_type in non_images:
|
||||
assert attachment_service.is_image(mime_type) is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestMessageAttachmentServiceMaxFileSize:
|
||||
"""Tests for max file size retrieval."""
|
||||
|
||||
def test_get_max_file_size_from_settings(self, db, attachment_service):
|
||||
"""Test retrieving max file size from platform settings."""
|
||||
with patch(
|
||||
"app.services.message_attachment_service.admin_settings_service"
|
||||
) as mock_settings:
|
||||
mock_settings.get_setting_value.return_value = 15
|
||||
max_size = attachment_service.get_max_file_size_bytes(db)
|
||||
assert max_size == 15 * 1024 * 1024 # 15 MB in bytes
|
||||
|
||||
def test_get_max_file_size_default(self, db, attachment_service):
|
||||
"""Test default max file size when setting not found."""
|
||||
with patch(
|
||||
"app.services.message_attachment_service.admin_settings_service"
|
||||
) as mock_settings:
|
||||
mock_settings.get_setting_value.return_value = DEFAULT_MAX_FILE_SIZE_MB
|
||||
max_size = attachment_service.get_max_file_size_bytes(db)
|
||||
assert max_size == DEFAULT_MAX_FILE_SIZE_MB * 1024 * 1024
|
||||
|
||||
def test_get_max_file_size_invalid_value(self, db, attachment_service):
|
||||
"""Test handling of invalid setting value."""
|
||||
with patch(
|
||||
"app.services.message_attachment_service.admin_settings_service"
|
||||
) as mock_settings:
|
||||
mock_settings.get_setting_value.return_value = "invalid"
|
||||
max_size = attachment_service.get_max_file_size_bytes(db)
|
||||
assert max_size == DEFAULT_MAX_FILE_SIZE_MB * 1024 * 1024
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestMessageAttachmentServiceValidateAndStore:
|
||||
"""Tests for validate_and_store method."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_and_store_success(
|
||||
self, db, attachment_service, mock_upload_file
|
||||
):
|
||||
"""Test successful file storage."""
|
||||
file = mock_upload_file(
|
||||
content=b"test file content",
|
||||
filename="document.pdf",
|
||||
content_type="application/pdf",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"app.services.message_attachment_service.admin_settings_service"
|
||||
) as mock_settings:
|
||||
mock_settings.get_setting_value.return_value = 10
|
||||
|
||||
result = await attachment_service.validate_and_store(
|
||||
db=db,
|
||||
file=file,
|
||||
conversation_id=1,
|
||||
)
|
||||
|
||||
assert result["original_filename"] == "document.pdf"
|
||||
assert result["mime_type"] == "application/pdf"
|
||||
assert result["file_size"] == len(b"test file content")
|
||||
assert result["is_image"] is False
|
||||
assert result["filename"].endswith(".pdf")
|
||||
assert os.path.exists(result["file_path"])
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_and_store_image(
|
||||
self, db, attachment_service, mock_upload_file
|
||||
):
|
||||
"""Test storage of image file."""
|
||||
# Create a minimal valid PNG
|
||||
png_header = (
|
||||
b"\x89PNG\r\n\x1a\n" # PNG signature
|
||||
+ b"\x00\x00\x00\rIHDR" # IHDR chunk header
|
||||
+ b"\x00\x00\x00\x01" # width = 1
|
||||
+ b"\x00\x00\x00\x01" # height = 1
|
||||
+ b"\x08\x02" # bit depth = 8, color type = RGB
|
||||
+ b"\x00\x00\x00" # compression, filter, interlace
|
||||
)
|
||||
|
||||
file = mock_upload_file(
|
||||
content=png_header,
|
||||
filename="image.png",
|
||||
content_type="image/png",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"app.services.message_attachment_service.admin_settings_service"
|
||||
) as mock_settings:
|
||||
mock_settings.get_setting_value.return_value = 10
|
||||
|
||||
result = await attachment_service.validate_and_store(
|
||||
db=db,
|
||||
file=file,
|
||||
conversation_id=1,
|
||||
)
|
||||
|
||||
assert result["original_filename"] == "image.png"
|
||||
assert result["mime_type"] == "image/png"
|
||||
assert result["is_image"] is True
|
||||
assert result["filename"].endswith(".png")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_and_store_invalid_type(
|
||||
self, db, attachment_service, mock_upload_file
|
||||
):
|
||||
"""Test rejection of invalid file type."""
|
||||
file = mock_upload_file(
|
||||
content=b"<script>alert('xss')</script>",
|
||||
filename="script.js",
|
||||
content_type="application/javascript",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"app.services.message_attachment_service.admin_settings_service"
|
||||
) as mock_settings:
|
||||
mock_settings.get_setting_value.return_value = 10
|
||||
|
||||
with pytest.raises(ValueError, match="File type.*not allowed"):
|
||||
await attachment_service.validate_and_store(
|
||||
db=db,
|
||||
file=file,
|
||||
conversation_id=1,
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_and_store_file_too_large(
|
||||
self, db, attachment_service, mock_upload_file
|
||||
):
|
||||
"""Test rejection of oversized file."""
|
||||
# Create content larger than max size
|
||||
large_content = b"x" * (11 * 1024 * 1024) # 11 MB
|
||||
file = mock_upload_file(
|
||||
content=large_content,
|
||||
filename="large.pdf",
|
||||
content_type="application/pdf",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"app.services.message_attachment_service.admin_settings_service"
|
||||
) as mock_settings:
|
||||
mock_settings.get_setting_value.return_value = 10 # 10 MB limit
|
||||
|
||||
with pytest.raises(ValueError, match="exceeds maximum allowed size"):
|
||||
await attachment_service.validate_and_store(
|
||||
db=db,
|
||||
file=file,
|
||||
conversation_id=1,
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_and_store_no_filename(
|
||||
self, db, attachment_service, mock_upload_file
|
||||
):
|
||||
"""Test handling of file without filename."""
|
||||
file = mock_upload_file(
|
||||
content=b"test content",
|
||||
filename=None,
|
||||
content_type="text/plain",
|
||||
)
|
||||
file.filename = None # Ensure it's None
|
||||
|
||||
with patch(
|
||||
"app.services.message_attachment_service.admin_settings_service"
|
||||
) as mock_settings:
|
||||
mock_settings.get_setting_value.return_value = 10
|
||||
|
||||
result = await attachment_service.validate_and_store(
|
||||
db=db,
|
||||
file=file,
|
||||
conversation_id=1,
|
||||
)
|
||||
|
||||
assert result["original_filename"] == "attachment"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_validate_and_store_no_content_type(
|
||||
self, db, attachment_service, mock_upload_file
|
||||
):
|
||||
"""Test handling of file without content type (falls back to octet-stream)."""
|
||||
file = mock_upload_file(
|
||||
content=b"test content",
|
||||
filename="file.bin",
|
||||
content_type=None,
|
||||
)
|
||||
file.content_type = None
|
||||
|
||||
with patch(
|
||||
"app.services.message_attachment_service.admin_settings_service"
|
||||
) as mock_settings:
|
||||
mock_settings.get_setting_value.return_value = 10
|
||||
|
||||
# Should reject application/octet-stream as not allowed
|
||||
with pytest.raises(ValueError, match="File type.*not allowed"):
|
||||
await attachment_service.validate_and_store(
|
||||
db=db,
|
||||
file=file,
|
||||
conversation_id=1,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestMessageAttachmentServiceFileOperations:
|
||||
"""Tests for file operation methods."""
|
||||
|
||||
def test_delete_attachment_success(self, attachment_service):
|
||||
"""Test successful attachment deletion."""
|
||||
# Create a temp file
|
||||
with tempfile.NamedTemporaryFile(delete=False) as f:
|
||||
f.write(b"test content")
|
||||
file_path = f.name
|
||||
|
||||
assert os.path.exists(file_path)
|
||||
|
||||
result = attachment_service.delete_attachment(file_path)
|
||||
|
||||
assert result is True
|
||||
assert not os.path.exists(file_path)
|
||||
|
||||
def test_delete_attachment_with_thumbnail(self, attachment_service):
|
||||
"""Test deletion of attachment with thumbnail."""
|
||||
# Create temp files
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as f:
|
||||
f.write(b"image content")
|
||||
file_path = f.name
|
||||
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix="_thumb.png") as f:
|
||||
f.write(b"thumbnail content")
|
||||
thumb_path = f.name
|
||||
|
||||
result = attachment_service.delete_attachment(file_path, thumb_path)
|
||||
|
||||
assert result is True
|
||||
assert not os.path.exists(file_path)
|
||||
assert not os.path.exists(thumb_path)
|
||||
|
||||
def test_delete_attachment_file_not_exists(self, attachment_service):
|
||||
"""Test deletion when file doesn't exist."""
|
||||
result = attachment_service.delete_attachment("/nonexistent/file.pdf")
|
||||
assert result is True # No error, just returns True
|
||||
|
||||
def test_get_download_url(self, attachment_service):
|
||||
"""Test download URL generation."""
|
||||
url = attachment_service.get_download_url("uploads/messages/2025/01/1/abc.pdf")
|
||||
assert url == "/static/uploads/messages/2025/01/1/abc.pdf"
|
||||
|
||||
def test_get_file_content_success(self, attachment_service):
|
||||
"""Test reading file content."""
|
||||
test_content = b"test file content"
|
||||
with tempfile.NamedTemporaryFile(delete=False) as f:
|
||||
f.write(test_content)
|
||||
file_path = f.name
|
||||
|
||||
try:
|
||||
result = attachment_service.get_file_content(file_path)
|
||||
assert result == test_content
|
||||
finally:
|
||||
os.unlink(file_path)
|
||||
|
||||
def test_get_file_content_not_found(self, attachment_service):
|
||||
"""Test reading non-existent file."""
|
||||
result = attachment_service.get_file_content("/nonexistent/file.pdf")
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestMessageAttachmentServiceThumbnail:
|
||||
"""Tests for thumbnail creation."""
|
||||
|
||||
def test_create_thumbnail_pil_not_installed(self, attachment_service):
|
||||
"""Test graceful handling when PIL is not available."""
|
||||
with patch.dict("sys.modules", {"PIL": None}):
|
||||
# This should not raise an error, just return empty dict
|
||||
result = attachment_service._create_thumbnail(
|
||||
b"fake image content", "/tmp/test.png"
|
||||
)
|
||||
# When PIL import fails, it returns empty dict
|
||||
assert isinstance(result, dict)
|
||||
|
||||
def test_create_thumbnail_invalid_image(self, attachment_service):
|
||||
"""Test handling of invalid image data."""
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as f:
|
||||
f.write(b"not an image")
|
||||
file_path = f.name
|
||||
|
||||
try:
|
||||
result = attachment_service._create_thumbnail(b"not an image", file_path)
|
||||
# Should return empty dict on error
|
||||
assert isinstance(result, dict)
|
||||
finally:
|
||||
os.unlink(file_path)
|
||||
Reference in New Issue
Block a user