CoreRasurae
399 lines
16 KiB
Python
399 lines
16 KiB
Python
"""
|
|
Client Tests
|
|
==========
|
|
|
|
This module contains tests for the StorageClient class.
|
|
"""
|
|
|
|
import os
|
|
import pytest
|
|
import tempfile
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from clevercloud_storage_framework.client import StorageClient
|
|
from clevercloud_storage_framework.exceptions import StorageError
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_provider():
|
|
"""Create a mock storage provider for testing."""
|
|
provider = MagicMock()
|
|
provider.get_provider_name.return_value = 'mock'
|
|
return provider
|
|
|
|
|
|
@pytest.fixture
|
|
def client(mock_provider):
|
|
"""Create a StorageClient instance for testing."""
|
|
return StorageClient(mock_provider)
|
|
|
|
|
|
class TestStorageClient:
|
|
"""Tests for the StorageClient class."""
|
|
|
|
def test_list_files(self, client, mock_provider):
|
|
"""Test listing files."""
|
|
# Set up the mock provider
|
|
mock_provider.list_files.return_value = [
|
|
{'name': 'file1.txt', 'path': '/path/to/file1.txt'},
|
|
{'name': 'file2.txt', 'path': '/path/to/file2.txt'}
|
|
]
|
|
|
|
# Call the method
|
|
files = client.list_files('/path/to')
|
|
|
|
# Check the result
|
|
assert len(files) == 2
|
|
assert files[0]['name'] == 'file1.txt'
|
|
assert files[1]['name'] == 'file2.txt'
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.list_files.assert_called_once_with('/path/to', False)
|
|
|
|
def test_file_exists(self, client, mock_provider):
|
|
"""Test checking if a file exists."""
|
|
# Set up the mock provider
|
|
mock_provider.file_exists.return_value = True
|
|
|
|
# Call the method
|
|
result = client.file_exists('/path/to/file.txt')
|
|
|
|
# Check the result
|
|
assert result is True
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.file_exists.assert_called_once_with('/path/to/file.txt')
|
|
|
|
def test_get_file_size(self, client, mock_provider):
|
|
"""Test getting a file's size."""
|
|
# Set up the mock provider
|
|
mock_provider.get_file_size.return_value = 1024
|
|
|
|
# Call the method
|
|
size = client.get_file_size('/path/to/file.txt')
|
|
|
|
# Check the result
|
|
assert size == 1024
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.get_file_size.assert_called_once_with('/path/to/file.txt')
|
|
|
|
def test_get_file_metadata(self, client, mock_provider):
|
|
"""Test getting a file's metadata."""
|
|
# Set up the mock provider
|
|
mock_provider.get_file_metadata.return_value = {
|
|
'name': 'file.txt',
|
|
'path': '/path/to/file.txt',
|
|
'size': 1024
|
|
}
|
|
|
|
# Call the method
|
|
metadata = client.get_file_metadata('/path/to/file.txt')
|
|
|
|
# Check the result
|
|
assert metadata['name'] == 'file.txt'
|
|
assert metadata['path'] == '/path/to/file.txt'
|
|
assert metadata['size'] == 1024
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.get_file_metadata.assert_called_once_with('/path/to/file.txt')
|
|
|
|
def test_read_file_to_bytes(self, client, mock_provider):
|
|
"""Test reading a file to bytes."""
|
|
# Set up the mock provider
|
|
mock_provider.read_file.return_value = [b'chunk1', b'chunk2']
|
|
|
|
# Call the method
|
|
content = client.read_file('/path/to/file.txt')
|
|
|
|
# Check the result
|
|
assert content == b'chunk1chunk2'
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.read_file.assert_called_once_with('/path/to/file.txt')
|
|
|
|
def test_read_file_to_local(self, client, mock_provider):
|
|
"""Test reading a file to a local file."""
|
|
# Set up the mock provider
|
|
mock_provider.read_file.return_value = [b'chunk1', b'chunk2']
|
|
|
|
# Create a temporary file
|
|
with tempfile.NamedTemporaryFile(delete=False) as f:
|
|
local_path = f.name
|
|
|
|
try:
|
|
# Call the method
|
|
result = client.read_file('/path/to/file.txt', local_path)
|
|
|
|
# Check the result
|
|
assert result == local_path
|
|
|
|
# Check that the file was written correctly
|
|
with open(local_path, 'rb') as f:
|
|
assert f.read() == b'chunk1chunk2'
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.read_file.assert_called_once_with('/path/to/file.txt')
|
|
finally:
|
|
# Clean up
|
|
os.unlink(local_path)
|
|
|
|
def test_write_file_from_bytes(self, client, mock_provider):
|
|
"""Test writing a file from bytes."""
|
|
# Call the method
|
|
client.write_file('/path/to/file.txt', b'content')
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.write_file.assert_called_once_with('/path/to/file.txt', b'content')
|
|
|
|
def test_write_file_from_string_path(self, client, mock_provider):
|
|
"""Test writing a file from a string path."""
|
|
# Create a temporary file
|
|
with tempfile.NamedTemporaryFile(delete=False) as f:
|
|
f.write(b'content')
|
|
local_path = f.name
|
|
|
|
try:
|
|
# Call the method
|
|
with patch('os.path.exists', return_value=True):
|
|
client.write_file('/path/to/file.txt', local_path)
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.write_file.assert_called_once()
|
|
|
|
# The first argument should be the destination path
|
|
assert mock_provider.write_file.call_args[0][0] == '/path/to/file.txt'
|
|
|
|
# The second argument should be a file-like object
|
|
assert hasattr(mock_provider.write_file.call_args[0][1], 'read')
|
|
finally:
|
|
# Clean up
|
|
os.unlink(local_path)
|
|
|
|
def test_write_file_from_string_content(self, client, mock_provider):
|
|
"""Test writing a file from a string content."""
|
|
# Call the method
|
|
with patch('os.path.exists', return_value=False):
|
|
client.write_file('/path/to/file.txt', 'content')
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.write_file.assert_called_once_with('/path/to/file.txt', b'content')
|
|
|
|
def test_delete_file(self, client, mock_provider):
|
|
"""Test deleting a file."""
|
|
# Call the method
|
|
client.delete_file('/path/to/file.txt')
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.delete_file.assert_called_once_with('/path/to/file.txt')
|
|
|
|
def test_copy_same_provider(self, client, mock_provider):
|
|
"""Test copying a file with the same provider."""
|
|
# Set up the mock provider
|
|
mock_provider.supports_path.return_value = True
|
|
|
|
# Call the method
|
|
client.copy('/source/path', '/dest/path')
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.copy_file.assert_called_once_with('/source/path', '/dest/path')
|
|
|
|
@patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path')
|
|
def test_copy_different_providers(self, mock_get_provider, client, mock_provider):
|
|
"""Test copying a file with different providers."""
|
|
# Set up the mock providers
|
|
mock_provider.supports_path.side_effect = [True, False]
|
|
mock_provider.supports_destination_protocol.return_value = False
|
|
|
|
source_provider = mock_provider
|
|
dest_provider = MagicMock()
|
|
mock_get_provider.return_value = dest_provider
|
|
dest_provider.supports_destination_protocol.return_value = True
|
|
|
|
source_provider.read_file.return_value = [b'chunk1', b'chunk2']
|
|
|
|
# Call the method
|
|
client.copy('/source/path', '/dest/path')
|
|
|
|
# Check that the destination provider's write_file was called
|
|
dest_provider.write_file.assert_called_once()
|
|
|
|
# The first argument should be the destination path
|
|
assert dest_provider.write_file.call_args[0][0] == '/dest/path'
|
|
|
|
def test_move_same_provider(self, client, mock_provider):
|
|
"""Test moving a file with the same provider."""
|
|
# Set up the mock provider
|
|
mock_provider.supports_path.return_value = True
|
|
|
|
# Call the method
|
|
client.move('/source/path', '/dest/path')
|
|
|
|
# Check that the provider methods were called
|
|
mock_provider.move_file.assert_called_once_with('/source/path', '/dest/path')
|
|
|
|
def test_create_directory(self, client, mock_provider):
|
|
"""Test creating a directory."""
|
|
# Call the method
|
|
client.create_directory('/path/to/dir')
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.create_directory.assert_called_once_with('/path/to/dir')
|
|
|
|
def test_delete_directory(self, client, mock_provider):
|
|
"""Test deleting a directory."""
|
|
# Call the method
|
|
client.delete_directory('/path/to/dir', recursive=True)
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.delete_directory.assert_called_once_with('/path/to/dir', True)
|
|
|
|
def test_get_provider_name(self, client, mock_provider):
|
|
"""Test getting the provider name."""
|
|
# Set up the mock provider
|
|
mock_provider.get_provider_name.return_value = 'mock'
|
|
|
|
# Call the method
|
|
name = client.get_provider_name()
|
|
|
|
# Check the result
|
|
assert name == 'mock'
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.get_provider_name.assert_called_once()
|
|
|
|
def test_write_file_with_unsupported_type(self, client, mock_provider):
|
|
"""Test writing a file with an unsupported type."""
|
|
# Call the method with an unsupported type
|
|
with pytest.raises(ValueError) as excinfo:
|
|
client.write_file('/path/to/file.txt', 123) # Not a valid content type
|
|
|
|
# Verify the error message
|
|
assert "Unsupported content type" in str(excinfo.value)
|
|
|
|
@patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path')
|
|
def test_move_different_providers(self, mock_get_provider, client, mock_provider):
|
|
"""Test moving a file with different providers."""
|
|
# Set up the mock providers:
|
|
# First call to supports_path checks if both paths are supported by the current provider
|
|
# Second and third calls check individual paths
|
|
mock_provider.supports_path.side_effect = [False, False, True]
|
|
|
|
source_provider = MagicMock()
|
|
mock_get_provider.return_value = source_provider
|
|
|
|
# Set up the source provider to return content chunks
|
|
source_provider.read_file.return_value = [b'chunk1', b'chunk2']
|
|
|
|
# Mock the copy method to track its calls
|
|
with patch.object(client, 'copy') as mock_copy:
|
|
# Call the method
|
|
client.move('/source/path', '/dest/path')
|
|
|
|
# Verify that copy was called with the correct arguments
|
|
mock_copy.assert_called_once_with('/source/path', '/dest/path')
|
|
|
|
# Verify that the source provider's delete_file was called
|
|
source_provider.delete_file.assert_called_once_with('/source/path')
|
|
|
|
@patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path', side_effect=StorageError("No provider found"))
|
|
def test_copy_fallback_to_current_provider(self, mock_get_provider, client, mock_provider):
|
|
"""Test copying a file when no provider is found for the path."""
|
|
# Set up the mock provider
|
|
mock_provider.supports_path.return_value = False
|
|
mock_provider.read_file.return_value = [b'chunk1', b'chunk2']
|
|
|
|
# Call the method
|
|
client.copy('/source/path', '/dest/path')
|
|
|
|
# Check that the provider methods were called
|
|
mock_provider.read_file.assert_called_once_with('/source/path')
|
|
mock_provider.write_file.assert_called_once_with('/dest/path', mock_provider.read_file.return_value)
|
|
|
|
def test_write_file_with_file_path(self, client, mock_provider):
|
|
"""Test writing a file with a file path."""
|
|
# Create a temporary file
|
|
with tempfile.NamedTemporaryFile(delete=False) as f:
|
|
f.write(b'test content')
|
|
temp_path = f.name
|
|
|
|
try:
|
|
# Call the method with patch to make os.path.exists return True
|
|
with patch('os.path.exists', return_value=True):
|
|
# Mock open to return our temp file
|
|
with patch('builtins.open', return_value=open(temp_path, 'rb')) as mock_open:
|
|
client.write_file('/path/to/file.txt', '/local/file.txt')
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.write_file.assert_called_once()
|
|
finally:
|
|
# Clean up
|
|
os.unlink(temp_path)
|
|
|
|
def test_write_file_with_string_as_content(self, client, mock_provider):
|
|
"""Test writing a file with a string as content."""
|
|
# Call the method with a string that doesn't exist as a file
|
|
with patch('os.path.exists', return_value=False):
|
|
client.write_file('/path/to/file.txt', 'string content')
|
|
|
|
# Check that the provider method was called with encoded bytes
|
|
mock_provider.write_file.assert_called_once_with('/path/to/file.txt', b'string content')
|
|
|
|
@patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path')
|
|
def test_copy_with_unsupported_destination_protocol(self, mock_get_provider, client, mock_provider):
|
|
"""Test copying a file when the destination provider doesn't support the source protocol."""
|
|
# Set up the mock providers to trigger the error path
|
|
mock_provider.supports_path.side_effect = [True, False]
|
|
mock_provider.supports_destination_protocol.return_value = True
|
|
|
|
dest_provider = MagicMock()
|
|
mock_get_provider.return_value = dest_provider
|
|
dest_provider.supports_destination_protocol.return_value = False
|
|
|
|
# Call the method
|
|
client.copy('/source/path', '/dest/path')
|
|
|
|
# Verify that the provider's copy_file method was called
|
|
mock_provider.copy_file.assert_called_once_with('/source/path', '/dest/path')
|
|
|
|
def test_supports_presigned_urls(self, client, mock_provider):
|
|
"""Test checking if the provider supports pre-signed URLs."""
|
|
# Set up the mock provider
|
|
mock_provider.supports_presigned_urls.return_value = True
|
|
|
|
# Call the method
|
|
result = client.supports_presigned_urls()
|
|
|
|
# Check the result
|
|
assert result is True
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.supports_presigned_urls.assert_called_once()
|
|
|
|
def test_generate_presigned_url(self, client, mock_provider):
|
|
"""Test generating a pre-signed URL."""
|
|
# Set up the mock provider
|
|
mock_provider.generate_presigned_url.return_value = 'https://example.com/presigned-url'
|
|
|
|
# Call the method
|
|
url = client.generate_presigned_url('/path/to/file.txt', expiration=1800)
|
|
|
|
# Check the result
|
|
assert url == 'https://example.com/presigned-url'
|
|
|
|
# Check that the provider method was called
|
|
mock_provider.generate_presigned_url.assert_called_once_with('/path/to/file.txt', 1800)
|
|
|
|
def test_generate_presigned_url_default_expiration(self, client, mock_provider):
|
|
"""Test generating a pre-signed URL with default expiration."""
|
|
# Set up the mock provider
|
|
mock_provider.generate_presigned_url.return_value = 'https://example.com/presigned-url'
|
|
|
|
# Call the method without expiration parameter
|
|
url = client.generate_presigned_url('/path/to/file.txt')
|
|
|
|
# Check the result
|
|
assert url == 'https://example.com/presigned-url'
|
|
|
|
# Check that the provider method was called with default expiration
|
|
mock_provider.generate_presigned_url.assert_called_once_with('/path/to/file.txt', 3600) |