CoreRasurae
139 lines
6.0 KiB
Python
139 lines
6.0 KiB
Python
"""
|
|
Factory Tests
|
|
===========
|
|
|
|
This module contains tests for the StorageClientFactory class.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from clevercloud_storage_framework.factory import StorageClientFactory
|
|
from clevercloud_storage_framework.client import StorageClient
|
|
from clevercloud_storage_framework.providers import (
|
|
S3StorageProvider, EFSStorageProvider, LocalStorageProvider
|
|
)
|
|
from clevercloud_storage_framework.exceptions import StorageError
|
|
|
|
|
|
class TestStorageClientFactory:
|
|
"""Tests for the StorageClientFactory class."""
|
|
|
|
@patch('boto3.client')
|
|
def test_create_client(self, mock_boto3_client):
|
|
"""Test creating a client."""
|
|
# Mock boto3 client to avoid AWS API calls
|
|
mock_boto3_client.return_value = MagicMock()
|
|
|
|
# Create an S3 client
|
|
s3_client = StorageClientFactory.create_client('s3')
|
|
assert isinstance(s3_client, StorageClient)
|
|
assert isinstance(s3_client.provider, S3StorageProvider)
|
|
assert s3_client.get_provider_name() == 's3'
|
|
|
|
# Create an EFS client
|
|
efs_client = StorageClientFactory.create_client('efs', region_name='us-east-1')
|
|
assert isinstance(efs_client, StorageClient)
|
|
assert isinstance(efs_client.provider, EFSStorageProvider)
|
|
assert efs_client.get_provider_name() == 'efs'
|
|
|
|
# Create a local client
|
|
local_client = StorageClientFactory.create_client('local')
|
|
assert isinstance(local_client, StorageClient)
|
|
assert isinstance(local_client.provider, LocalStorageProvider)
|
|
assert local_client.get_provider_name() == 'local'
|
|
|
|
# Invalid provider type
|
|
with pytest.raises(StorageError):
|
|
StorageClientFactory.create_client('invalid')
|
|
|
|
@patch('boto3.client')
|
|
def test_create_provider(self, mock_boto3_client):
|
|
"""Test creating a provider."""
|
|
# Mock boto3 client to avoid AWS API calls
|
|
mock_boto3_client.return_value = MagicMock()
|
|
|
|
# Create an S3 provider
|
|
s3_provider = StorageClientFactory.create_provider('s3')
|
|
assert isinstance(s3_provider, S3StorageProvider)
|
|
assert s3_provider.get_provider_name() == 's3'
|
|
|
|
# Create an EFS provider
|
|
efs_provider = StorageClientFactory.create_provider('efs', region_name='us-east-1')
|
|
assert isinstance(efs_provider, EFSStorageProvider)
|
|
assert efs_provider.get_provider_name() == 'efs'
|
|
|
|
# Create a local provider
|
|
local_provider = StorageClientFactory.create_provider('local')
|
|
assert isinstance(local_provider, LocalStorageProvider)
|
|
assert local_provider.get_provider_name() == 'local'
|
|
|
|
# Invalid provider type
|
|
with pytest.raises(StorageError):
|
|
StorageClientFactory.create_provider('invalid')
|
|
|
|
@patch('boto3.client')
|
|
def test_get_provider_for_path(self, mock_boto3_client):
|
|
"""Test getting a provider for a path."""
|
|
# Mock boto3 client to avoid AWS API calls
|
|
mock_boto3_client.return_value = MagicMock()
|
|
|
|
# S3 path
|
|
provider = StorageClientFactory.get_provider_for_path('s3://bucket/key')
|
|
assert isinstance(provider, S3StorageProvider)
|
|
assert provider.get_provider_name() == 's3'
|
|
|
|
# EFS path
|
|
provider = StorageClientFactory.get_provider_for_path('efs://filesystem/path', region_name='us-east-1')
|
|
assert isinstance(provider, EFSStorageProvider)
|
|
assert provider.get_provider_name() == 'efs'
|
|
|
|
# Local path
|
|
provider = StorageClientFactory.get_provider_for_path('/local/path')
|
|
assert isinstance(provider, LocalStorageProvider)
|
|
assert provider.get_provider_name() == 'local'
|
|
|
|
# Test with unsupported path - need to patch all provider supports_path_format methods
|
|
# to return False to ensure the error is raised
|
|
with patch.object(S3StorageProvider, 'supports_path_format', return_value=False), \
|
|
patch.object(EFSStorageProvider, 'supports_path_format', return_value=False), \
|
|
patch.object(LocalStorageProvider, 'supports_path_format', return_value=False):
|
|
with pytest.raises(StorageError):
|
|
StorageClientFactory.get_provider_for_path('unsupported://path')
|
|
|
|
@patch('boto3.client')
|
|
def test_register_provider(self, mock_boto3_client):
|
|
"""Test registering a custom provider."""
|
|
# Mock boto3 client to avoid AWS API calls
|
|
mock_boto3_client.return_value = MagicMock()
|
|
|
|
# Create a custom provider class
|
|
class CustomProvider(LocalStorageProvider):
|
|
def get_provider_name(self):
|
|
return 'custom'
|
|
|
|
def supports_path(self, path):
|
|
return path.startswith('custom://')
|
|
|
|
@staticmethod
|
|
def supports_path_format(path):
|
|
return path.startswith('custom://')
|
|
|
|
# Register the custom provider
|
|
StorageClientFactory.register_provider('custom', CustomProvider)
|
|
|
|
# Create a client with the custom provider
|
|
client = StorageClientFactory.create_client('custom')
|
|
assert isinstance(client.provider, CustomProvider)
|
|
assert client.get_provider_name() == 'custom'
|
|
|
|
# Get a provider for a custom path - need to patch all provider classes
|
|
# to ensure our custom provider is selected
|
|
with patch.object(S3StorageProvider, 'supports_path_format', return_value=False), \
|
|
patch.object(EFSStorageProvider, 'supports_path_format', return_value=False), \
|
|
patch.object(LocalStorageProvider, 'supports_path_format', return_value=False), \
|
|
patch.object(CustomProvider, 'supports_path_format', return_value=True):
|
|
provider = StorageClientFactory.get_provider_for_path('custom://path')
|
|
assert isinstance(provider, CustomProvider)
|
|
assert provider.get_provider_name() == 'custom'
|