""" 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)