521 lines
18 KiB
Python
521 lines
18 KiB
Python
"""
|
|
Tests for S3StorageBackend.
|
|
|
|
TDD Phase 1: RED - Write tests first, then implement to pass.
|
|
"""
|
|
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch, call
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir() -> Path:
|
|
"""Create a temporary directory for tests."""
|
|
temp_dir = Path(tempfile.mkdtemp())
|
|
yield temp_dir
|
|
shutil.rmtree(temp_dir, ignore_errors=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_file(temp_dir: Path) -> Path:
|
|
"""Create a sample file for testing."""
|
|
file_path = temp_dir / "sample.txt"
|
|
file_path.write_text("Hello, World!")
|
|
return file_path
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_boto3_client():
|
|
"""Create a mock boto3 S3 client."""
|
|
with patch("boto3.client") as mock_client_func:
|
|
mock_client = MagicMock()
|
|
mock_client_func.return_value = mock_client
|
|
yield mock_client
|
|
|
|
|
|
class TestS3StorageBackendCreation:
|
|
"""Tests for S3StorageBackend instantiation."""
|
|
|
|
def test_create_with_bucket_name(self, mock_boto3_client: MagicMock) -> None:
|
|
"""Test creating backend with bucket name."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
assert backend.bucket_name == "test-bucket"
|
|
|
|
def test_create_with_region(self, mock_boto3_client: MagicMock) -> None:
|
|
"""Test creating backend with region."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
with patch("boto3.client") as mock_client:
|
|
S3StorageBackend(
|
|
bucket_name="test-bucket",
|
|
region_name="us-west-2",
|
|
)
|
|
|
|
mock_client.assert_called_once()
|
|
call_kwargs = mock_client.call_args[1]
|
|
assert call_kwargs.get("region_name") == "us-west-2"
|
|
|
|
def test_create_with_credentials(self, mock_boto3_client: MagicMock) -> None:
|
|
"""Test creating backend with explicit credentials."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
with patch("boto3.client") as mock_client:
|
|
S3StorageBackend(
|
|
bucket_name="test-bucket",
|
|
access_key_id="AKIATEST",
|
|
secret_access_key="secret123",
|
|
)
|
|
|
|
mock_client.assert_called_once()
|
|
call_kwargs = mock_client.call_args[1]
|
|
assert call_kwargs.get("aws_access_key_id") == "AKIATEST"
|
|
assert call_kwargs.get("aws_secret_access_key") == "secret123"
|
|
|
|
def test_create_with_endpoint_url(self, mock_boto3_client: MagicMock) -> None:
|
|
"""Test creating backend with custom endpoint (for S3-compatible services)."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
with patch("boto3.client") as mock_client:
|
|
S3StorageBackend(
|
|
bucket_name="test-bucket",
|
|
endpoint_url="http://localhost:9000",
|
|
)
|
|
|
|
mock_client.assert_called_once()
|
|
call_kwargs = mock_client.call_args[1]
|
|
assert call_kwargs.get("endpoint_url") == "http://localhost:9000"
|
|
|
|
def test_create_bucket_when_requested(self, mock_boto3_client: MagicMock) -> None:
|
|
"""Test that bucket is created when create_bucket=True."""
|
|
from botocore.exceptions import ClientError
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_bucket.side_effect = ClientError(
|
|
{"Error": {"Code": "404"}}, "HeadBucket"
|
|
)
|
|
|
|
S3StorageBackend(
|
|
bucket_name="test-bucket",
|
|
create_bucket=True,
|
|
)
|
|
|
|
mock_boto3_client.create_bucket.assert_called_once()
|
|
|
|
def test_is_storage_backend_subclass(self, mock_boto3_client: MagicMock) -> None:
|
|
"""Test that S3StorageBackend is a StorageBackend."""
|
|
from shared.storage.base import StorageBackend
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
assert isinstance(backend, StorageBackend)
|
|
|
|
|
|
class TestS3StorageBackendUpload:
|
|
"""Tests for S3StorageBackend.upload method."""
|
|
|
|
def test_upload_file(
|
|
self, mock_boto3_client: MagicMock, temp_dir: Path, sample_file: Path
|
|
) -> None:
|
|
"""Test uploading a file."""
|
|
from botocore.exceptions import ClientError
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
# Object does not exist
|
|
mock_boto3_client.head_object.side_effect = ClientError(
|
|
{"Error": {"Code": "404"}}, "HeadObject"
|
|
)
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
result = backend.upload(sample_file, "uploads/sample.txt")
|
|
|
|
assert result == "uploads/sample.txt"
|
|
mock_boto3_client.upload_file.assert_called_once()
|
|
|
|
def test_upload_fails_if_exists_without_overwrite(
|
|
self, mock_boto3_client: MagicMock, sample_file: Path
|
|
) -> None:
|
|
"""Test that upload fails if object exists and overwrite is False."""
|
|
from shared.storage.base import StorageError
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.return_value = {} # Object exists
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
with pytest.raises(StorageError, match="already exists"):
|
|
backend.upload(sample_file, "sample.txt", overwrite=False)
|
|
|
|
def test_upload_succeeds_with_overwrite(
|
|
self, mock_boto3_client: MagicMock, sample_file: Path
|
|
) -> None:
|
|
"""Test that upload succeeds with overwrite=True."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.return_value = {} # Object exists
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
result = backend.upload(sample_file, "sample.txt", overwrite=True)
|
|
|
|
assert result == "sample.txt"
|
|
mock_boto3_client.upload_file.assert_called_once()
|
|
|
|
def test_upload_nonexistent_file_fails(
|
|
self, mock_boto3_client: MagicMock, temp_dir: Path
|
|
) -> None:
|
|
"""Test that uploading nonexistent file fails."""
|
|
from shared.storage.base import FileNotFoundStorageError
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
with pytest.raises(FileNotFoundStorageError):
|
|
backend.upload(temp_dir / "nonexistent.txt", "sample.txt")
|
|
|
|
|
|
class TestS3StorageBackendDownload:
|
|
"""Tests for S3StorageBackend.download method."""
|
|
|
|
def test_download_file(
|
|
self, mock_boto3_client: MagicMock, temp_dir: Path
|
|
) -> None:
|
|
"""Test downloading a file."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.return_value = {} # Object exists
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
local_path = temp_dir / "downloaded.txt"
|
|
|
|
result = backend.download("sample.txt", local_path)
|
|
|
|
assert result == local_path
|
|
mock_boto3_client.download_file.assert_called_once()
|
|
|
|
def test_download_creates_parent_directories(
|
|
self, mock_boto3_client: MagicMock, temp_dir: Path
|
|
) -> None:
|
|
"""Test that download creates parent directories."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.return_value = {}
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
local_path = temp_dir / "deep" / "nested" / "downloaded.txt"
|
|
|
|
backend.download("sample.txt", local_path)
|
|
|
|
assert local_path.parent.exists()
|
|
|
|
def test_download_nonexistent_object_fails(
|
|
self, mock_boto3_client: MagicMock, temp_dir: Path
|
|
) -> None:
|
|
"""Test that downloading nonexistent object fails."""
|
|
from botocore.exceptions import ClientError
|
|
from shared.storage.base import FileNotFoundStorageError
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.side_effect = ClientError(
|
|
{"Error": {"Code": "404"}}, "HeadObject"
|
|
)
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
with pytest.raises(FileNotFoundStorageError):
|
|
backend.download("nonexistent.txt", temp_dir / "file.txt")
|
|
|
|
|
|
class TestS3StorageBackendExists:
|
|
"""Tests for S3StorageBackend.exists method."""
|
|
|
|
def test_exists_returns_true_for_existing_object(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test exists returns True for existing object."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.return_value = {}
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
assert backend.exists("sample.txt") is True
|
|
|
|
def test_exists_returns_false_for_nonexistent_object(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test exists returns False for nonexistent object."""
|
|
from botocore.exceptions import ClientError
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.side_effect = ClientError(
|
|
{"Error": {"Code": "404"}}, "HeadObject"
|
|
)
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
assert backend.exists("nonexistent.txt") is False
|
|
|
|
|
|
class TestS3StorageBackendListFiles:
|
|
"""Tests for S3StorageBackend.list_files method."""
|
|
|
|
def test_list_files_returns_objects(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test listing objects."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.list_objects_v2.return_value = {
|
|
"Contents": [
|
|
{"Key": "file1.txt"},
|
|
{"Key": "file2.txt"},
|
|
{"Key": "subdir/file3.txt"},
|
|
]
|
|
}
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
files = backend.list_files("")
|
|
|
|
assert len(files) == 3
|
|
assert "file1.txt" in files
|
|
assert "file2.txt" in files
|
|
assert "subdir/file3.txt" in files
|
|
|
|
def test_list_files_with_prefix(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test listing objects with prefix filter."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.list_objects_v2.return_value = {
|
|
"Contents": [
|
|
{"Key": "images/a.png"},
|
|
{"Key": "images/b.png"},
|
|
]
|
|
}
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
files = backend.list_files("images/")
|
|
|
|
mock_boto3_client.list_objects_v2.assert_called_with(
|
|
Bucket="test-bucket", Prefix="images/"
|
|
)
|
|
|
|
def test_list_files_empty_bucket(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test listing files in empty bucket."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.list_objects_v2.return_value = {} # No Contents key
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
files = backend.list_files("")
|
|
|
|
assert files == []
|
|
|
|
|
|
class TestS3StorageBackendDelete:
|
|
"""Tests for S3StorageBackend.delete method."""
|
|
|
|
def test_delete_existing_object(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test deleting an existing object."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.return_value = {}
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
result = backend.delete("sample.txt")
|
|
|
|
assert result is True
|
|
mock_boto3_client.delete_object.assert_called_once()
|
|
|
|
def test_delete_nonexistent_object_returns_false(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test deleting nonexistent object returns False."""
|
|
from botocore.exceptions import ClientError
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.side_effect = ClientError(
|
|
{"Error": {"Code": "404"}}, "HeadObject"
|
|
)
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
result = backend.delete("nonexistent.txt")
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestS3StorageBackendGetUrl:
|
|
"""Tests for S3StorageBackend.get_url method."""
|
|
|
|
def test_get_url_returns_s3_url(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test get_url returns S3 URL."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.return_value = {}
|
|
mock_boto3_client.generate_presigned_url.return_value = (
|
|
"https://test-bucket.s3.amazonaws.com/sample.txt"
|
|
)
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
url = backend.get_url("sample.txt")
|
|
|
|
assert "sample.txt" in url
|
|
|
|
def test_get_url_nonexistent_object_raises(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test get_url raises for nonexistent object."""
|
|
from botocore.exceptions import ClientError
|
|
from shared.storage.base import FileNotFoundStorageError
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.side_effect = ClientError(
|
|
{"Error": {"Code": "404"}}, "HeadObject"
|
|
)
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
with pytest.raises(FileNotFoundStorageError):
|
|
backend.get_url("nonexistent.txt")
|
|
|
|
|
|
class TestS3StorageBackendUploadBytes:
|
|
"""Tests for S3StorageBackend.upload_bytes method."""
|
|
|
|
def test_upload_bytes(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test uploading bytes directly."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
from botocore.exceptions import ClientError
|
|
mock_boto3_client.head_object.side_effect = ClientError(
|
|
{"Error": {"Code": "404"}}, "HeadObject"
|
|
)
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
data = b"Binary content here"
|
|
|
|
result = backend.upload_bytes(data, "binary.dat")
|
|
|
|
assert result == "binary.dat"
|
|
mock_boto3_client.put_object.assert_called_once()
|
|
|
|
def test_upload_bytes_fails_if_exists_without_overwrite(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test upload_bytes fails if object exists and overwrite is False."""
|
|
from shared.storage.base import StorageError
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.return_value = {} # Object exists
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
with pytest.raises(StorageError, match="already exists"):
|
|
backend.upload_bytes(b"content", "sample.txt", overwrite=False)
|
|
|
|
|
|
class TestS3StorageBackendDownloadBytes:
|
|
"""Tests for S3StorageBackend.download_bytes method."""
|
|
|
|
def test_download_bytes(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test downloading object as bytes."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.read.return_value = b"Hello, World!"
|
|
mock_boto3_client.get_object.return_value = {"Body": mock_response}
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
data = backend.download_bytes("sample.txt")
|
|
|
|
assert data == b"Hello, World!"
|
|
|
|
def test_download_bytes_nonexistent_raises(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test downloading nonexistent object as bytes."""
|
|
from botocore.exceptions import ClientError
|
|
from shared.storage.base import FileNotFoundStorageError
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.get_object.side_effect = ClientError(
|
|
{"Error": {"Code": "NoSuchKey"}}, "GetObject"
|
|
)
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
with pytest.raises(FileNotFoundStorageError):
|
|
backend.download_bytes("nonexistent.txt")
|
|
|
|
|
|
class TestS3StorageBackendPresignedUrl:
|
|
"""Tests for S3StorageBackend.get_presigned_url method."""
|
|
|
|
def test_get_presigned_url_generates_url(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test get_presigned_url generates presigned URL."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.return_value = {}
|
|
mock_boto3_client.generate_presigned_url.return_value = (
|
|
"https://test-bucket.s3.amazonaws.com/sample.txt?X-Amz-Algorithm=..."
|
|
)
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
url = backend.get_presigned_url("sample.txt")
|
|
|
|
assert "X-Amz-Algorithm" in url or "sample.txt" in url
|
|
mock_boto3_client.generate_presigned_url.assert_called_once()
|
|
|
|
def test_get_presigned_url_with_custom_expiry(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test get_presigned_url uses custom expiry."""
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.return_value = {}
|
|
mock_boto3_client.generate_presigned_url.return_value = "https://..."
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
backend.get_presigned_url("sample.txt", expires_in_seconds=7200)
|
|
|
|
call_args = mock_boto3_client.generate_presigned_url.call_args
|
|
assert call_args[1].get("ExpiresIn") == 7200
|
|
|
|
def test_get_presigned_url_nonexistent_raises(
|
|
self, mock_boto3_client: MagicMock
|
|
) -> None:
|
|
"""Test get_presigned_url raises for nonexistent object."""
|
|
from botocore.exceptions import ClientError
|
|
from shared.storage.base import FileNotFoundStorageError
|
|
from shared.storage.s3 import S3StorageBackend
|
|
|
|
mock_boto3_client.head_object.side_effect = ClientError(
|
|
{"Error": {"Code": "404"}}, "HeadObject"
|
|
)
|
|
|
|
backend = S3StorageBackend(bucket_name="test-bucket")
|
|
|
|
with pytest.raises(FileNotFoundStorageError):
|
|
backend.get_presigned_url("nonexistent.txt")
|