add pytest in makefile

2025-07-24 11:26:35 +08:00
parent e8820839fc
commit 08b2a71f45
9 changed files with 988 additions and 130 deletions
--- a/tests/test_toolkit.py
+++ b/tests/test_toolkit.py
@@ -3,6 +3,10 @@
 RustFS S3 Storage Toolkit 测试文件
 基于成功的 test_s3_flexible.py 创建的测试套件
 已在 RustFS 1.0.0-alpha.34 上完成测试
+
+环境变量控制:
+- RUN_INTEGRATION_TESTS=1: 运行真实的集成测试
+- 默认: 运行模拟测试
 """

 import os
@@ -10,17 +14,21 @@ import tempfile
 import shutil
 from pathlib import Path
 from datetime import datetime
+import pytest

 from rustfs_s3_toolkit import S3StorageToolkit


+# 检查是否运行集成测试
+RUN_INTEGRATION_TESTS = os.getenv('RUN_INTEGRATION_TESTS', '0') == '1'
+
 # 测试配置 - 请根据实际情况修改
 TEST_CONFIG = {
-    "endpoint_url": "https://rfs.jmsu.top",
-    "access_key_id": "lingyuzeng",
-    "secret_access_key": "rustAdminlingyuzeng",
-    "bucket_name": "test",
-    "region_name": "us-east-1"
+    "endpoint_url": os.getenv("S3_ENDPOINT_URL", "https://rfs.jmsu.top"),
+    "access_key_id": os.getenv("S3_ACCESS_KEY_ID", "lingyuzeng"),
+    "secret_access_key": os.getenv("S3_SECRET_ACCESS_KEY", "rustAdminlingyuzeng"),
+    "bucket_name": os.getenv("S3_BUCKET_NAME", "test"),
+    "region_name": os.getenv("S3_REGION_NAME", "us-east-1")
 }


@@ -52,18 +60,20 @@ def create_test_files():
    return test_dir


+@pytest.mark.skipif(not RUN_INTEGRATION_TESTS, reason="集成测试需要设置 RUN_INTEGRATION_TESTS=1")
 def test_all_functions():
-    """测试所有 9 个核心功能"""
+    """测试所有 9 个核心功能（集成测试，需要真实 S3 服务）"""
    print("🚀 RustFS S3 Storage Toolkit 完整功能测试")
    print("🧪 测试环境: RustFS 1.0.0-alpha.34")
    print("=" * 60)
-    
+
    # 初始化工具包
    try:
        toolkit = S3StorageToolkit(**TEST_CONFIG)
        print("✅ 工具包初始化成功")
    except Exception as e:
        print(f"❌ 工具包初始化失败: {e}")
+        pytest.skip(f"无法连接到 S3 服务: {e}")
        return False
    
    test_results = {}
@@ -236,8 +246,70 @@ def test_all_functions():
    else:
        print("❌ 所有测试都失败了。请检查 RustFS 配置和网络连接。")
    
+    # 对于 pytest，使用 assert 而不是 return
+    assert passed_tests == len(test_names), f"只有 {passed_tests}/{len(test_names)} 个测试通过"
+
    return test_results


+def test_toolkit_initialization():
+    """测试工具包初始化（单元测试，不需要真实 S3 服务）"""
+    # 测试基本初始化
+    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"
+
+    # 测试带区域的初始化
+    toolkit_with_region = 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_with_region.bucket_name == "test-bucket"
+    assert toolkit_with_region.endpoint_url == "https://test.example.com"
+
+
+def test_basic_functionality():
+    """测试基本功能（确保代码结构正确）"""
+    # 这个测试确保所有方法都存在且可调用
+    toolkit = S3StorageToolkit(
+        endpoint_url="https://test.example.com",
+        access_key_id="test_key",
+        secret_access_key="test_secret",
+        bucket_name="test-bucket"
+    )
+
+    # 检查所有核心方法是否存在
+    assert hasattr(toolkit, 'test_connection')
+    assert hasattr(toolkit, 'upload_file')
+    assert hasattr(toolkit, 'create_folder')
+    assert hasattr(toolkit, 'upload_directory')
+    assert hasattr(toolkit, 'download_file')
+    assert hasattr(toolkit, 'download_directory')
+    assert hasattr(toolkit, 'list_files')
+    assert hasattr(toolkit, 'delete_file')
+    assert hasattr(toolkit, 'delete_directory')
+
+    # 检查方法是否可调用
+    assert callable(toolkit.test_connection)
+    assert callable(toolkit.upload_file)
+    assert callable(toolkit.create_folder)
+    assert callable(toolkit.upload_directory)
+    assert callable(toolkit.download_file)
+    assert callable(toolkit.download_directory)
+    assert callable(toolkit.list_files)
+    assert callable(toolkit.delete_file)
+    assert callable(toolkit.delete_directory)
+
+
 if __name__ == "__main__":
    test_all_functions()
--- a/tests/test_unit.py
+++ b/tests/test_unit.py
@@ -0,0 +1,180 @@
+#!/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__])