from unittest import mock
|
from unittest.mock import Mock
|
|
import pytest
|
from data_import.api import DownloadStorageData
|
from data_import.models import FileUpload
|
from django.conf import settings
|
from django.http import HttpResponse
|
from organizations.models import Organization
|
from rest_framework import status
|
from rest_framework.test import APIRequestFactory
|
from users.models import User
|
|
"""
|
Comprehensive test suite for DownloadStorageData API
|
|
This test module validates the complete functionality of the DownloadStorageData view,
|
which handles secure file downloads from persistent storage (S3, GCS, Azure, etc.).
|
|
## Test Coverage:
|
|
### Authentication & Authorization Tests:
|
- Missing filepath parameter validation
|
- Invalid filepath security checks
|
- FileUpload permission validation
|
- Organization-based avatar access control
|
- Cross-organization access prevention
|
|
### File Serving Mode Tests:
|
- **NGINX Mode**: X-Accel-Redirect header generation for high-performance serving
|
- **Direct Mode**: RangedFileResponse streaming with range request support
|
- Mode switching via USE_NGINX_FOR_UPLOADS setting
|
|
### File Type Support Tests:
|
- Upload files (settings.UPLOAD_DIR) with permission checking
|
- Avatar files (settings.AVATAR_PATH) with organization validation
|
- Content-Type detection for various formats (PDF, MP3, MP4, JPG, unknown)
|
- Content-Disposition headers (inline vs attachment)
|
|
### Edge Case & Security Tests:
|
- URL encoding/decoding support
|
- NGINX redirect URL construction
|
- Mock response handling for different scenarios
|
- Error response codes (403, 404)
|
|
## Test Structure:
|
Each test method focuses on a specific aspect of the API:
|
1. Sets up mock objects and request data
|
2. Calls the DownloadStorageData.get() method
|
3. Validates response status, headers, and behavior
|
4. Verifies security constraints and permissions
|
|
## Mocking Strategy:
|
- FileUpload.objects.filter() for database queries
|
- User.objects.filter() for avatar access
|
- RangedFileResponse for direct file serving
|
- Settings overrides for mode switching
|
- Organization membership checks
|
|
This comprehensive test suite ensures the API correctly handles all file types,
|
security scenarios, and serving modes while maintaining backward compatibility.
|
"""
|
|
pytestmark = pytest.mark.django_db
|
|
|
class TestDownloadStorageData:
|
@pytest.fixture
|
def api_factory(self):
|
return APIRequestFactory()
|
|
@pytest.fixture
|
def user(self):
|
"""Create a test user with organization"""
|
org = Organization.objects.create(title='Test Org')
|
user = User.objects.create_user(email='test@example.com', password='test123')
|
user.active_organization = org
|
user.save()
|
return user
|
|
@pytest.fixture
|
def other_user(self):
|
"""Create another user with different organization"""
|
other_org = Organization.objects.create(title='Other Org')
|
other_user = User.objects.create_user(email='other@example.com', password='test123')
|
other_user.active_organization = other_org
|
other_user.save()
|
return other_user
|
|
@pytest.fixture
|
def mock_file_upload(self, user):
|
"""Create a mock FileUpload object"""
|
file_upload = Mock(spec=FileUpload)
|
file_upload.has_permission = Mock(return_value=True)
|
|
# Mock the file object and storage
|
mock_file = Mock()
|
mock_storage = Mock()
|
mock_storage.url = Mock(return_value='http://example.com/test.pdf')
|
mock_file.storage = mock_storage
|
mock_file.name = 'test.pdf'
|
mock_file.open = Mock(return_value=Mock())
|
|
file_upload.file = mock_file
|
return file_upload
|
|
@pytest.fixture
|
def view(self):
|
return DownloadStorageData()
|
|
def test_missing_filepath_returns_404(self, api_factory, user, view):
|
"""Test that missing filepath parameter returns 404"""
|
request = api_factory.get('/storage-data/uploaded/')
|
request.user = user
|
|
response = view.get(request)
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
def test_invalid_filepath_returns_403(self, api_factory, user, view):
|
"""Test that filepath not starting with UPLOAD_DIR or AVATAR_PATH returns 403"""
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': 'invalid/path/file.pdf'})
|
request.user = user
|
|
response = view.get(request)
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
@mock.patch('data_import.api.FileUpload.objects.filter')
|
def test_upload_file_not_found_returns_403(self, mock_filter, api_factory, user, view):
|
"""Test that non-existent upload file returns 403"""
|
mock_filter.return_value.last.return_value = None
|
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': f'{settings.UPLOAD_DIR}/test.pdf'})
|
request.user = user
|
|
response = view.get(request)
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
@mock.patch('data_import.api.FileUpload.objects.filter')
|
def test_upload_file_no_permission_returns_403(self, mock_filter, api_factory, user, view, mock_file_upload):
|
"""Test that upload file without permission returns 403"""
|
mock_file_upload.has_permission.return_value = False
|
mock_filter.return_value.last.return_value = mock_file_upload
|
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': f'{settings.UPLOAD_DIR}/test.pdf'})
|
request.user = user
|
|
response = view.get(request)
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
mock_file_upload.has_permission.assert_called_once_with(user)
|
|
@mock.patch('data_import.api.User.objects.filter')
|
def test_avatar_user_not_found_returns_403(self, mock_filter, api_factory, user, view):
|
"""Test that non-existent avatar user returns 403"""
|
mock_filter.return_value.first.return_value = None
|
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': f'{settings.AVATAR_PATH}/avatar.jpg'})
|
request.user = user
|
|
response = view.get(request)
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
@mock.patch('data_import.api.User.objects.filter')
|
def test_avatar_user_no_organization_access_returns_403(self, mock_filter, api_factory, user, other_user, view):
|
"""Test that avatar user from different organization returns 403"""
|
mock_avatar_user = Mock()
|
mock_avatar_user.avatar = Mock()
|
mock_filter.return_value.first.return_value = mock_avatar_user
|
|
# Mock organization access check to return False
|
user.active_organization.has_user = Mock(return_value=False)
|
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': f'{settings.AVATAR_PATH}/avatar.jpg'})
|
request.user = user
|
|
response = view.get(request)
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
@mock.patch('data_import.api.FileUpload.objects.filter')
|
@mock.patch('data_import.api.settings.USE_NGINX_FOR_UPLOADS', True)
|
def test_upload_file_nginx_serving(self, mock_filter, api_factory, user, view, mock_file_upload):
|
"""Test upload file serving through NGINX"""
|
mock_filter.return_value.last.return_value = mock_file_upload
|
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': f'{settings.UPLOAD_DIR}/test.pdf'})
|
request.user = user
|
|
response = view.get(request)
|
|
assert isinstance(response, HttpResponse)
|
assert response.status_code == 200
|
assert 'X-Accel-Redirect' in response
|
assert 'Content-Disposition' in response
|
assert 'inline' in response['Content-Disposition']
|
assert 'test.pdf' in response['Content-Disposition']
|
|
@mock.patch('data_import.api.FileUpload.objects.filter')
|
@mock.patch('data_import.api.settings.USE_NGINX_FOR_UPLOADS', False)
|
@mock.patch('data_import.api.RangedFileResponse')
|
def test_upload_file_direct_serving(
|
self, mock_ranged_response, mock_filter, api_factory, user, view, mock_file_upload
|
):
|
"""Test upload file serving directly (without NGINX)"""
|
mock_filter.return_value.last.return_value = mock_file_upload
|
mock_response_instance = Mock()
|
mock_response_instance.__setitem__ = Mock() # Allow item assignment
|
mock_ranged_response.return_value = mock_response_instance
|
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': f'{settings.UPLOAD_DIR}/test.pdf'})
|
request.user = user
|
|
response = view.get(request)
|
|
assert response == mock_response_instance
|
mock_ranged_response.assert_called_once()
|
|
@mock.patch('data_import.api.User.objects.filter')
|
@mock.patch('data_import.api.settings.USE_NGINX_FOR_UPLOADS', True)
|
def test_avatar_file_nginx_serving(self, mock_filter, api_factory, user, view):
|
"""Test avatar file serving through NGINX"""
|
mock_avatar_user = Mock()
|
mock_avatar_file = Mock()
|
mock_storage = Mock()
|
mock_storage.url = Mock(return_value='http://example.com/avatar.jpg')
|
mock_avatar_file.storage = mock_storage
|
mock_avatar_file.name = 'avatar.jpg'
|
mock_avatar_user.avatar = mock_avatar_file
|
mock_filter.return_value.first.return_value = mock_avatar_user
|
|
# Mock organization access check to return True
|
user.active_organization.has_user = Mock(return_value=True)
|
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': f'{settings.AVATAR_PATH}/avatar.jpg'})
|
request.user = user
|
|
response = view.get(request)
|
|
assert isinstance(response, HttpResponse)
|
assert response.status_code == 200
|
assert 'X-Accel-Redirect' in response
|
assert 'Content-Disposition' in response
|
|
@mock.patch('data_import.api.User.objects.filter')
|
@mock.patch('data_import.api.settings.USE_NGINX_FOR_UPLOADS', False)
|
@mock.patch('data_import.api.RangedFileResponse')
|
def test_avatar_file_direct_serving(self, mock_ranged_response, mock_filter, api_factory, user, view):
|
"""Test avatar file serving directly (without NGINX)"""
|
mock_avatar_user = Mock()
|
mock_avatar_file = Mock()
|
mock_avatar_file.open = Mock(return_value=Mock())
|
mock_avatar_user.avatar = mock_avatar_file
|
mock_filter.return_value.first.return_value = mock_avatar_user
|
mock_response_instance = Mock()
|
mock_response_instance.__setitem__ = Mock() # Allow item assignment
|
mock_ranged_response.return_value = mock_response_instance
|
|
# Mock organization access check to return True
|
user.active_organization.has_user = Mock(return_value=True)
|
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': f'{settings.AVATAR_PATH}/avatar.jpg'})
|
request.user = user
|
|
response = view.get(request)
|
|
assert response == mock_response_instance
|
mock_ranged_response.assert_called_once()
|
|
@pytest.mark.parametrize(
|
'file_extension,expected_content_type',
|
[
|
('test.pdf', 'application/pdf'),
|
('test.mp3', 'audio/mpeg'),
|
('test.mp4', 'video/mp4'),
|
('test.jpg', 'image/jpeg'),
|
('unknown.unknownext', 'application/octet-stream'),
|
],
|
)
|
@mock.patch('data_import.api.FileUpload.objects.filter')
|
@mock.patch('data_import.api.settings.USE_NGINX_FOR_UPLOADS', False)
|
@mock.patch('data_import.api.RangedFileResponse')
|
def test_content_type_detection(
|
self,
|
mock_ranged_response,
|
mock_filter,
|
file_extension,
|
expected_content_type,
|
api_factory,
|
user,
|
view,
|
mock_file_upload,
|
):
|
"""Test that content types are properly detected for different file extensions"""
|
mock_filter.return_value.last.return_value = mock_file_upload
|
mock_response_instance = Mock()
|
mock_response_instance.__setitem__ = Mock() # Allow item assignment
|
mock_ranged_response.return_value = mock_response_instance
|
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': f'{settings.UPLOAD_DIR}/{file_extension}'})
|
request.user = user
|
|
view.get(request)
|
|
# Check that RangedFileResponse was called with the expected content type
|
args, kwargs = mock_ranged_response.call_args
|
assert kwargs['content_type'] == expected_content_type
|
|
@mock.patch('data_import.api.FileUpload.objects.filter')
|
@mock.patch('data_import.api.settings.USE_NGINX_FOR_UPLOADS', True)
|
def test_nginx_redirect_url_construction(self, mock_filter, api_factory, user, view, mock_file_upload):
|
"""Test that NGINX redirect URL is properly constructed"""
|
# Set up mock to return a specific URL
|
mock_file_upload.file.storage.url.return_value = 'https://s3.amazonaws.com/bucket/test.pdf'
|
mock_filter.return_value.last.return_value = mock_file_upload
|
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': f'{settings.UPLOAD_DIR}/test.pdf'})
|
request.user = user
|
|
response = view.get(request)
|
|
# Check that X-Accel-Redirect header contains the expected path
|
redirect_header = response['X-Accel-Redirect']
|
assert redirect_header == '/file_download/https/s3.amazonaws.com/bucket/test.pdf'
|
|
@mock.patch('data_import.api.unquote')
|
@mock.patch('data_import.api.FileUpload.objects.filter')
|
def test_filepath_url_decoding(self, mock_filter, mock_unquote, api_factory, user, view, mock_file_upload):
|
"""Test that filepath is properly URL-decoded"""
|
mock_unquote.return_value = f'{settings.UPLOAD_DIR}/decoded file.pdf'
|
mock_filter.return_value.last.return_value = mock_file_upload
|
|
encoded_filepath = f'{settings.UPLOAD_DIR}/encoded%20file.pdf'
|
request = api_factory.get('/storage-data/uploaded/', {'filepath': encoded_filepath})
|
request.user = user
|
|
view.get(request)
|
|
mock_unquote.assert_called_once_with(encoded_filepath)
|
