rustfs-s3-toolkit/tests/test_unit.py

#!/usr/bin/env python3
"""
RustFS S3 Storage Toolkit 单元测试
不依赖外部服务的基础测试
"""

import pytest
from unittest.mock import Mock, patch
from rustfs_s3_toolkit import S3StorageToolkit


class TestS3StorageToolkit:
    """S3StorageToolkit 单元测试类"""
    
    def test_init(self):
        """测试工具包初始化"""
        toolkit = S3StorageToolkit(
            endpoint_url="https://test.example.com",
            access_key_id="test_key",
            secret_access_key="test_secret",
            bucket_name="test-bucket"
        )
        
        assert toolkit.bucket_name == "test-bucket"
        assert toolkit.endpoint_url == "https://test.example.com"
    
    def test_init_with_region(self):
        """测试带区域的初始化"""
        toolkit = S3StorageToolkit(
            endpoint_url="https://test.example.com",
            access_key_id="test_key",
            secret_access_key="test_secret",
            bucket_name="test-bucket",
            region_name="us-west-2"
        )
        
        assert toolkit.bucket_name == "test-bucket"
        assert toolkit.endpoint_url == "https://test.example.com"
    
    @patch('boto3.client')
    def test_test_connection_success(self, mock_boto_client):
        """测试连接成功的情况"""
        # 模拟 boto3 客户端
        mock_client = Mock()
        mock_boto_client.return_value = mock_client
        
        # 模拟 list_buckets 响应
        mock_client.list_buckets.return_value = {
            'Buckets': [
                {'Name': 'test-bucket'},
                {'Name': 'other-bucket'}
            ]
        }
        
        # 模拟 head_bucket 成功
        mock_client.head_bucket.return_value = {}
        
        toolkit = S3StorageToolkit(
            endpoint_url="https://test.example.com",
            access_key_id="test_key",
            secret_access_key="test_secret",
            bucket_name="test-bucket"
        )
        
        result = toolkit.test_connection()
        
        assert result['success'] is True
        assert result['bucket_count'] == 2
        assert 'test-bucket' in result['bucket_names']
        assert result['target_bucket_exists'] is True
    
    @patch('boto3.client')
    def test_test_connection_failure(self, mock_boto_client):
        """测试连接失败的情况"""
        # 模拟 boto3 客户端
        mock_client = Mock()
        mock_boto_client.return_value = mock_client
        
        # 模拟连接异常
        mock_client.list_buckets.side_effect = Exception("Connection failed")
        
        toolkit = S3StorageToolkit(
            endpoint_url="https://test.example.com",
            access_key_id="test_key",
            secret_access_key="test_secret",
            bucket_name="test-bucket"
        )
        
        result = toolkit.test_connection()
        
        assert result['success'] is False
        assert 'error' in result
    
    @patch('boto3.client')
    def test_upload_file_success(self, mock_boto_client):
        """测试文件上传成功"""
        # 模拟 boto3 客户端
        mock_client = Mock()
        mock_boto_client.return_value = mock_client
        
        # 模拟上传成功
        mock_client.upload_file.return_value = None
        
        toolkit = S3StorageToolkit(
            endpoint_url="https://test.example.com",
            access_key_id="test_key",
            secret_access_key="test_secret",
            bucket_name="test-bucket"
        )
        
        # 创建临时文件进行测试
        import tempfile
        import os
        
        with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
            f.write("test content")
            temp_file = f.name
        
        try:
            result = toolkit.upload_file(temp_file, "test/file.txt")
            
            assert result['success'] is True
            assert result['key'] == "test/file.txt"
            assert result['bucket'] == "test-bucket"
            
        finally:
            os.unlink(temp_file)
    
    @patch('boto3.client')
    def test_create_folder_success(self, mock_boto_client):
        """测试文件夹创建成功"""
        # 模拟 boto3 客户端
        mock_client = Mock()
        mock_boto_client.return_value = mock_client
        
        # 模拟上传成功
        mock_client.put_object.return_value = {}
        
        toolkit = S3StorageToolkit(
            endpoint_url="https://test.example.com",
            access_key_id="test_key",
            secret_access_key="test_secret",
            bucket_name="test-bucket"
        )
        
        result = toolkit.create_folder("test/folder/")
        
        assert result['success'] is True
        assert result['folder_path'] == "test/folder/"
        assert result['bucket'] == "test-bucket"
    
    def test_folder_path_normalization(self):
        """测试文件夹路径规范化"""
        toolkit = S3StorageToolkit(
            endpoint_url="https://test.example.com",
            access_key_id="test_key",
            secret_access_key="test_secret",
            bucket_name="test-bucket"
        )
        
        # 测试路径规范化（这个方法可能需要在实际类中实现）
        # 这里只是示例，实际实现可能不同
        test_paths = [
            ("folder", "folder/"),
            ("folder/", "folder/"),
            ("folder/subfolder", "folder/subfolder/"),
            ("folder/subfolder/", "folder/subfolder/"),
        ]
        
        for input_path, expected in test_paths:
            # 假设有一个内部方法来规范化路径
            if not input_path.endswith('/'):
                normalized = input_path + '/'
            else:
                normalized = input_path
            assert normalized == expected


if __name__ == "__main__":
    pytest.main([__file__])