CoreRasurae
159 lines
6.8 KiB
Python
159 lines
6.8 KiB
Python
"""
|
|
Test module for StorageClient with SAS token handling
|
|
=====================================================
|
|
|
|
This module contains tests for the StorageClient class with a focus on
|
|
SAS token handling when working with S3 provider.
|
|
"""
|
|
|
|
import os
|
|
import io
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock, call
|
|
|
|
from clevercloud_storage_framework.client import StorageClient
|
|
from clevercloud_storage_framework.providers.s3_provider import S3StorageProvider
|
|
from clevercloud_storage_framework.exceptions import StorageError
|
|
|
|
|
|
class TestStorageClientSASToken:
|
|
"""Test suite for StorageClient with SAS token handling."""
|
|
|
|
@pytest.fixture
|
|
def mock_s3_provider(self):
|
|
"""Create a mock S3 provider with SAS token."""
|
|
provider = MagicMock(spec=S3StorageProvider)
|
|
provider.sas_token = "?sig=abc123&se=2023-01-01"
|
|
provider.get_provider_name.return_value = "s3"
|
|
return provider
|
|
|
|
@pytest.fixture
|
|
def client(self, mock_s3_provider):
|
|
"""Create a StorageClient with the mock S3 provider."""
|
|
return StorageClient(mock_s3_provider)
|
|
|
|
def test_read_file_with_sas_token_as_bytes(self, client, mock_s3_provider):
|
|
"""Test reading a file with SAS token and returning as bytes."""
|
|
# Setup mock provider to return chunks
|
|
mock_s3_provider.read_file.return_value = iter([b"chunk1", b"chunk2"])
|
|
|
|
# Call the method
|
|
result = client.read_file("s3://bucket/file.txt")
|
|
|
|
# Verify the result
|
|
assert result == b"chunk1chunk2"
|
|
mock_s3_provider.read_file.assert_called_once_with("s3://bucket/file.txt")
|
|
|
|
def test_read_file_with_sas_token_as_iterator(self, client, mock_s3_provider):
|
|
"""Test reading a file with SAS token and returning as iterator."""
|
|
# Setup mock provider to return chunks
|
|
mock_chunks = iter([b"chunk1", b"chunk2"])
|
|
mock_s3_provider.read_file.return_value = mock_chunks
|
|
|
|
# Call the method
|
|
result = client.read_file("s3://bucket/file.txt", as_iterator=True)
|
|
|
|
# Verify the result is the same iterator
|
|
assert result is mock_chunks
|
|
mock_s3_provider.read_file.assert_called_once_with("s3://bucket/file.txt")
|
|
|
|
def test_read_file_with_sas_token_to_local_file(self, client, mock_s3_provider, tmpdir):
|
|
"""Test reading a file with SAS token and saving to a local file."""
|
|
# Setup mock provider to return chunks
|
|
mock_s3_provider.read_file.return_value = iter([b"chunk1", b"chunk2"])
|
|
|
|
# Create a temporary file path
|
|
local_path = os.path.join(tmpdir, "downloaded_file.txt")
|
|
|
|
# Call the method
|
|
result = client.read_file("s3://bucket/file.txt", local_path=local_path)
|
|
|
|
# Verify the result
|
|
assert result == local_path
|
|
assert os.path.exists(local_path)
|
|
with open(local_path, 'rb') as f:
|
|
content = f.read()
|
|
assert content == b"chunk1chunk2"
|
|
mock_s3_provider.read_file.assert_called_once_with("s3://bucket/file.txt")
|
|
|
|
def test_write_file_with_sas_token_from_bytes(self, client, mock_s3_provider):
|
|
"""Test writing bytes to a file with SAS token."""
|
|
# Call the method
|
|
client.write_file("s3://bucket/file.txt", b"content")
|
|
|
|
# Verify the provider method was called correctly
|
|
mock_s3_provider.write_file.assert_called_once_with("s3://bucket/file.txt", b"content")
|
|
|
|
def test_write_file_with_sas_token_from_iterator(self, client, mock_s3_provider):
|
|
"""Test writing an iterator of bytes to a file with SAS token."""
|
|
# Create an iterator of bytes
|
|
content_iter = iter([b"chunk1", b"chunk2"])
|
|
|
|
# Call the method
|
|
client.write_file("s3://bucket/file.txt", content_iter)
|
|
|
|
# Verify the provider method was called correctly
|
|
mock_s3_provider.write_file.assert_called_once_with("s3://bucket/file.txt", content_iter)
|
|
|
|
def test_write_file_with_sas_token_from_file_object(self, client, mock_s3_provider):
|
|
"""Test writing a file-like object to a file with SAS token."""
|
|
# Create a file-like object
|
|
file_obj = io.BytesIO(b"file content")
|
|
|
|
# Call the method
|
|
client.write_file("s3://bucket/file.txt", file_obj)
|
|
|
|
# Verify the provider method was called correctly
|
|
mock_s3_provider.write_file.assert_called_once()
|
|
# Check that the first argument is correct
|
|
assert mock_s3_provider.write_file.call_args[0][0] == "s3://bucket/file.txt"
|
|
# The second argument should be a file-like object
|
|
assert hasattr(mock_s3_provider.write_file.call_args[0][1], 'read')
|
|
|
|
def test_copy_with_sas_token(self, client, mock_s3_provider):
|
|
"""Test copying a file with SAS token."""
|
|
# Setup the provider to handle the copy
|
|
mock_s3_provider.supports_path.return_value = True
|
|
|
|
# Call the method
|
|
client.copy("s3://bucket/source.txt", "s3://bucket/dest.txt")
|
|
|
|
# Verify the provider method was called correctly
|
|
mock_s3_provider.copy_file.assert_called_once_with(
|
|
"s3://bucket/source.txt", "s3://bucket/dest.txt"
|
|
)
|
|
|
|
def test_move_with_sas_token(self, client, mock_s3_provider):
|
|
"""Test moving a file with SAS token."""
|
|
# Setup the provider to handle the move
|
|
mock_s3_provider.supports_path.return_value = True
|
|
|
|
# Call the method
|
|
client.move("s3://bucket/source.txt", "s3://bucket/dest.txt")
|
|
|
|
# Verify the provider method was called correctly
|
|
mock_s3_provider.move_file.assert_called_once_with(
|
|
"s3://bucket/source.txt", "s3://bucket/dest.txt"
|
|
)
|
|
|
|
def test_cross_provider_copy_with_sas_token(self, client, mock_s3_provider):
|
|
"""Test copying between providers with SAS token."""
|
|
# Setup the source provider to not support the destination path
|
|
mock_s3_provider.supports_path.side_effect = lambda path: path.startswith("s3://bucket/source")
|
|
mock_s3_provider.supports_destination_protocol.return_value = False
|
|
|
|
# Setup the factory to return our mock provider
|
|
with patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path',
|
|
return_value=mock_s3_provider):
|
|
|
|
# Setup the read and write methods
|
|
mock_chunks = iter([b"chunk1", b"chunk2"])
|
|
mock_s3_provider.read_file.return_value = mock_chunks
|
|
|
|
# Call the method
|
|
client.copy("s3://bucket/source.txt", "s3://another-bucket/dest.txt")
|
|
|
|
# Verify the provider methods were called correctly
|
|
mock_s3_provider.read_file.assert_called_once_with("s3://bucket/source.txt")
|
|
mock_s3_provider.write_file.assert_called_once_with("s3://another-bucket/dest.txt", mock_chunks)
|