Add storage utilities and tests for file handling and tokens
Introduced new fixtures and tests for storage functionality, including saving files, generating URLs, and token creation/verification. Refactored `get_storage_provider` into a separate dependency module. Enhanced test coverage for improved reliability.
This commit is contained in:
@@ -19,6 +19,7 @@ from app.models import Event, GiftItem, GiftStatus, GiftPriority, GiftCategory,
|
||||
from app.models.user import User
|
||||
from app.utils.test_utils import setup_test_db, teardown_test_db, setup_async_test_db, teardown_async_test_db
|
||||
|
||||
pytest_plugins = ["pytest_asyncio"]
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def db_session():
|
||||
|
||||
75
backend/tests/core/storage.py
Normal file
75
backend/tests/core/storage.py
Normal file
@@ -0,0 +1,75 @@
|
||||
import os
|
||||
from io import BytesIO
|
||||
|
||||
import pytest
|
||||
from fastapi import UploadFile
|
||||
|
||||
from app.core.storage import LocalStorageProvider
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_storage():
|
||||
"""Create a test storage provider that uses a temp directory."""
|
||||
import tempfile
|
||||
test_dir = tempfile.mkdtemp()
|
||||
provider = LocalStorageProvider(upload_folder=test_dir, files_url_path="/test-files")
|
||||
yield provider
|
||||
# Clean up
|
||||
import shutil
|
||||
shutil.rmtree(test_dir)
|
||||
|
||||
|
||||
@pytest.mark.asyncio # Add this marker to run async tests
|
||||
async def test_save_file(test_storage):
|
||||
"""Test saving a file to storage."""
|
||||
# Create a test file
|
||||
content = b"test file content"
|
||||
test_file = BytesIO(content)
|
||||
|
||||
# Create UploadFile with the correct parameters
|
||||
file = UploadFile(
|
||||
filename="test.txt",
|
||||
file=test_file,
|
||||
)
|
||||
# Set content_type after creation
|
||||
# file.content_type = "text/plain"
|
||||
|
||||
# Save the file
|
||||
relative_path = "test-folder/test.txt"
|
||||
saved_path = await test_storage.save_file(file, relative_path)
|
||||
|
||||
# Verify the file exists
|
||||
full_path = os.path.join(test_storage.upload_folder, relative_path)
|
||||
assert os.path.exists(full_path)
|
||||
|
||||
# Check the content
|
||||
with open(full_path, "rb") as f:
|
||||
saved_content = f.read()
|
||||
assert saved_content == content
|
||||
|
||||
# Check the returned path
|
||||
assert saved_path == relative_path
|
||||
|
||||
|
||||
def test_generate_presigned_url(test_storage):
|
||||
"""Test generating a presigned URL."""
|
||||
file_path = "images/test.jpg"
|
||||
filename = "test.jpg"
|
||||
content_type = "image/jpeg"
|
||||
|
||||
upload_url, file_url = test_storage.generate_presigned_url(
|
||||
file_path, filename, content_type
|
||||
)
|
||||
|
||||
# Check the URLs
|
||||
assert upload_url.startswith("/api/v1/uploads/")
|
||||
assert file_url == f"/test-files/{file_path}"
|
||||
|
||||
|
||||
def test_get_file_url(test_storage):
|
||||
"""Test getting a file URL."""
|
||||
file_path = "images/test.jpg"
|
||||
|
||||
url = test_storage.get_file_url(file_path)
|
||||
|
||||
assert url == f"/test-files/{file_path}"
|
||||
58
backend/tests/utils/test_security.py
Normal file
58
backend/tests/utils/test_security.py
Normal file
@@ -0,0 +1,58 @@
|
||||
import time
|
||||
import pytest
|
||||
from app.utils.security import create_upload_token, verify_upload_token
|
||||
|
||||
|
||||
def test_upload_token_creation():
|
||||
"""Test that upload tokens can be created with expected fields."""
|
||||
file_path = "images/test.jpg"
|
||||
content_type = "image/jpeg"
|
||||
|
||||
token = create_upload_token(file_path, content_type)
|
||||
|
||||
assert token is not None
|
||||
assert isinstance(token, str)
|
||||
assert len(token) > 0
|
||||
|
||||
|
||||
def test_upload_token_verification():
|
||||
"""Test that created tokens can be verified."""
|
||||
file_path = "images/test.jpg"
|
||||
content_type = "image/jpeg"
|
||||
|
||||
token = create_upload_token(file_path, content_type)
|
||||
payload = verify_upload_token(token)
|
||||
|
||||
assert payload is not None
|
||||
assert payload["path"] == file_path
|
||||
assert payload["content_type"] == content_type
|
||||
assert payload["exp"] > int(time.time())
|
||||
|
||||
|
||||
def test_upload_token_expiration():
|
||||
"""Test that expired tokens are rejected."""
|
||||
file_path = "images/test.jpg"
|
||||
content_type = "image/jpeg"
|
||||
|
||||
# Create a token that expires in 1 second
|
||||
token = create_upload_token(file_path, content_type, expires_in=1)
|
||||
|
||||
# Wait for it to expire
|
||||
time.sleep(2)
|
||||
|
||||
payload = verify_upload_token(token)
|
||||
assert payload is None
|
||||
|
||||
|
||||
def test_upload_token_tampered():
|
||||
"""Test that tampered tokens are rejected."""
|
||||
file_path = "images/test.jpg"
|
||||
content_type = "image/jpeg"
|
||||
|
||||
token = create_upload_token(file_path, content_type)
|
||||
|
||||
# Tamper with the token
|
||||
tampered_token = token[:-5] + "XXXXX"
|
||||
|
||||
payload = verify_upload_token(tampered_token)
|
||||
assert payload is None
|
||||
Reference in New Issue
Block a user