""" Test suite for StorageClient auto-selection functionality ======================================================= This module contains tests for the auto-selection functionality added to StorageClient, which automatically selects appropriate providers based on path format. """ import pytest from unittest.mock import patch, MagicMock, call from clevercloud_storage_framework.client import StorageClient from clevercloud_storage_framework.exceptions import StorageError class TestStorageClientAutoSelection: """Test suite for StorageClient auto-selection functionality.""" @pytest.fixture def mock_provider(self): """Create a mock storage provider.""" provider = MagicMock() provider.get_provider_name.return_value = "mock" provider.supports_path.return_value = True provider.close.return_value = None return provider @pytest.fixture def mock_s3_provider(self): """Create a mock S3 provider.""" provider = MagicMock() provider.get_provider_name.return_value = "s3" provider.supports_path.return_value = True provider.close.return_value = None return provider @pytest.fixture def client(self, mock_provider): """Create a StorageClient with a mock provider.""" return StorageClient(mock_provider) def test_helper_find_provider_if_needed_supported_path(self, client, mock_provider): """Test _helper_find_provider_if_needed when current provider supports the path.""" # Setup path = "mock://bucket/file.txt" mock_provider.supports_path.return_value = True # Execute result = client._helper_find_provider_if_needed(path) # Verify assert result == mock_provider mock_provider.supports_path.assert_called_once_with(path) @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_helper_find_provider_if_needed_unsupported_path_no_sas(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test _helper_find_provider_if_needed when current provider doesn't support path and no SAS token.""" # Setup path = "s3://bucket/file.txt" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider # Execute result = client._helper_find_provider_if_needed(path) # Verify assert result == mock_s3_provider mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_helper_find_provider_if_needed_unsupported_path_with_sas(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test _helper_find_provider_if_needed when current provider doesn't support path and SAS token present.""" # Setup path = "s3://bucket/file.txt?sp=rl&st=2023-01-01T00:00:00Z&se=2023-01-02T00:00:00Z&sv=2022-02-02&sr=b&sig=test" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider # Execute result = client._helper_find_provider_if_needed(path) # Verify assert result == mock_s3_provider mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path, "?sp=rl&st=2023-01-01T00:00:00Z&se=2023-01-02T00:00:00Z&sv=2022-02-02&sr=b&sig=test") @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_list_files_auto_selection_success(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test list_files with auto-selection when alternative provider succeeds.""" # Setup path = "s3://bucket/file.txt" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider mock_s3_provider.list_files.return_value = [{"name": "file1.txt", "size": 100}] # Execute result = client.list_files(path, recursive=True) # Verify assert result == [{"name": "file1.txt", "size": 100}] mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) mock_s3_provider.list_files.assert_called_once_with(path, True) mock_s3_provider.close.assert_called_once() mock_provider.list_files.assert_not_called() @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_list_files_auto_selection_fallback(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test list_files with auto-selection when alternative provider fails and falls back.""" # Setup path = "s3://bucket/file.txt" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider mock_s3_provider.list_files.side_effect = StorageError("S3 error") mock_provider.list_files.return_value = [{"name": "file2.txt", "size": 200}] # Execute result = client.list_files(path, recursive=False) # Verify assert result == [{"name": "file2.txt", "size": 200}] mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) mock_s3_provider.list_files.assert_called_once_with(path, False) mock_s3_provider.close.assert_called_once() mock_provider.list_files.assert_called_once_with(path, False) @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_file_exists_auto_selection_success(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test file_exists with auto-selection when alternative provider succeeds.""" # Setup path = "s3://bucket/file.txt" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider mock_s3_provider.file_exists.return_value = True # Execute result = client.file_exists(path) # Verify assert result is True mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) mock_s3_provider.file_exists.assert_called_once_with(path) mock_s3_provider.close.assert_called_once() mock_provider.file_exists.assert_not_called() @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_file_exists_auto_selection_fallback(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test file_exists with auto-selection when alternative provider fails and falls back.""" # Setup path = "s3://bucket/file.txt" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider mock_s3_provider.file_exists.side_effect = StorageError("S3 error") mock_provider.file_exists.return_value = False # Execute result = client.file_exists(path) # Verify assert result is False mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) mock_s3_provider.file_exists.assert_called_once_with(path) mock_s3_provider.close.assert_called_once() mock_provider.file_exists.assert_called_once_with(path) @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_get_file_size_auto_selection_success(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test get_file_size with auto-selection when alternative provider succeeds.""" # Setup path = "s3://bucket/file.txt" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider mock_s3_provider.get_file_size.return_value = 1024 # Execute result = client.get_file_size(path) # Verify assert result == 1024 mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) mock_s3_provider.get_file_size.assert_called_once_with(path) mock_s3_provider.close.assert_called_once() mock_provider.get_file_size.assert_not_called() @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_get_file_metadata_auto_selection_success(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test get_file_metadata with auto-selection when alternative provider succeeds.""" # Setup path = "s3://bucket/file.txt" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider mock_s3_provider.get_file_metadata.return_value = {"size": 1024, "modified": "2023-01-01"} # Execute result = client.get_file_metadata(path) # Verify assert result == {"size": 1024, "modified": "2023-01-01"} mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) mock_s3_provider.get_file_metadata.assert_called_once_with(path) mock_s3_provider.close.assert_called_once() mock_provider.get_file_metadata.assert_not_called() @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_read_file_auto_selection_success(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test read_file with auto-selection when alternative provider succeeds.""" # Setup path = "s3://bucket/file.txt" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider mock_s3_provider.read_file.return_value = iter([b"test content"]) # Execute result = client.read_file(path, as_iterator=True) # Verify assert list(result) == [b"test content"] mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) mock_s3_provider.read_file.assert_called_once_with(path) mock_s3_provider.close.assert_called_once() @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_write_file_auto_selection_success(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test write_file with auto-selection when alternative provider succeeds.""" # Setup path = "s3://bucket/file.txt" content = b"test content" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider # Execute client.write_file(path, content) # Verify mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) mock_s3_provider.write_file.assert_called_once_with(path, content) mock_s3_provider.close.assert_called_once() mock_provider.write_file.assert_not_called() @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_create_directory_auto_selection_success(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test create_directory with auto-selection when alternative provider succeeds.""" # Setup path = "s3://bucket/folder/" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider # Execute client.create_directory(path) # Verify mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) mock_s3_provider.create_directory.assert_called_once_with(path) mock_s3_provider.close.assert_called_once() mock_provider.create_directory.assert_not_called() @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_delete_directory_auto_selection_success(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test delete_directory with auto-selection when alternative provider succeeds.""" # Setup path = "s3://bucket/folder/" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider # Execute client.delete_directory(path, recursive=True) # Verify mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) mock_s3_provider.delete_directory.assert_called_once_with(path, True) mock_s3_provider.close.assert_called_once() mock_provider.delete_directory.assert_not_called() def test_close_method(self, client, mock_provider): """Test the close method.""" # Execute client.close() # Verify mock_provider.close.assert_called_once() @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_auto_selection_with_none_provider(self, mock_get_provider, client, mock_provider): """Test auto-selection when get_provider_for_path returns None.""" # Setup path = "s3://bucket/file.txt" mock_provider.supports_path.return_value = False mock_get_provider.return_value = None mock_provider.list_files.return_value = [{"name": "file.txt", "size": 100}] # Execute result = client.list_files(path) # Verify assert result == [{"name": "file.txt", "size": 100}] mock_provider.supports_path.assert_called_once_with(path) mock_get_provider.assert_called_once_with(path) mock_provider.list_files.assert_called_once_with(path, False) @patch('clevercloud_storage_framework.factory.StorageClientFactory.get_provider_for_path') def test_auto_selection_provider_close_on_exception(self, mock_get_provider, client, mock_provider, mock_s3_provider): """Test that provider is closed even when an exception occurs.""" # Setup path = "s3://bucket/file.txt" mock_provider.supports_path.return_value = False mock_get_provider.return_value = mock_s3_provider mock_s3_provider.list_files.side_effect = StorageError("S3 error") mock_provider.list_files.return_value = [{"name": "file.txt", "size": 100}] # Execute result = client.list_files(path) # Verify assert result == [{"name": "file.txt", "size": 100}] mock_s3_provider.close.assert_called_once() mock_provider.list_files.assert_called_once_with(path, False)