Replaced calls to get_file_url with get_uploaded_file_url where appropriate. Added the get_uploaded_file_url method to the StorageProvider class and its implementations for more specific URL generation. Updated corresponding tests and API routes to reflect this change.
76 lines
2.0 KiB
Python
76 lines
2.0 KiB
Python
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_uploaded_file_url(file_path)
|
|
|
|
assert url == f"/test-files/{file_path}"
|