316 lines
11 KiB
Python
316 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
RustFS S3 Storage Toolkit 测试文件
|
||
基于成功的 test_s3_flexible.py 创建的测试套件
|
||
已在 RustFS 1.0.0-alpha.34 上完成测试
|
||
|
||
环境变量控制:
|
||
- RUN_INTEGRATION_TESTS=1: 运行真实的集成测试
|
||
- 默认: 运行模拟测试
|
||
"""
|
||
|
||
import os
|
||
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": 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")
|
||
}
|
||
|
||
|
||
def create_test_files():
|
||
"""创建测试文件和目录"""
|
||
test_dir = tempfile.mkdtemp(prefix='s3_toolkit_test_')
|
||
|
||
# 创建测试文件
|
||
test_files = {
|
||
"test.txt": "这是一个测试文件\n时间: " + datetime.now().isoformat(),
|
||
"data.json": '{"name": "test", "value": 123, "timestamp": "' + datetime.now().isoformat() + '"}',
|
||
"readme.md": "# 测试文件\n\n这是一个测试用的 Markdown 文件。\n\n- 项目: S3 存储测试\n- 时间: " + datetime.now().isoformat(),
|
||
}
|
||
|
||
# 创建子目录和文件
|
||
sub_dir = Path(test_dir) / "subdir"
|
||
sub_dir.mkdir()
|
||
|
||
for filename, content in test_files.items():
|
||
file_path = Path(test_dir) / filename
|
||
with open(file_path, 'w', encoding='utf-8') as f:
|
||
f.write(content)
|
||
|
||
# 在子目录中创建文件
|
||
sub_file = sub_dir / "sub_test.txt"
|
||
with open(sub_file, 'w', encoding='utf-8') as f:
|
||
f.write("子目录中的测试文件\n时间: " + datetime.now().isoformat())
|
||
|
||
return test_dir
|
||
|
||
|
||
@pytest.mark.skipif(not RUN_INTEGRATION_TESTS, reason="集成测试需要设置 RUN_INTEGRATION_TESTS=1")
|
||
def test_all_functions():
|
||
"""测试所有 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 = {}
|
||
|
||
# 1. 测试连接
|
||
print("\n1. 测试连接...")
|
||
result = toolkit.test_connection()
|
||
test_results['connection'] = result['success']
|
||
if result['success']:
|
||
print(f"✅ 连接成功: {result['message']}")
|
||
else:
|
||
print(f"❌ 连接失败: {result['error']}")
|
||
return test_results
|
||
|
||
# 2. 测试上传文件
|
||
print("\n2. 测试上传文件...")
|
||
test_content = f"测试文件内容\n时间: {datetime.now().isoformat()}"
|
||
with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt', encoding='utf-8') as f:
|
||
f.write(test_content)
|
||
test_file = f.name
|
||
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
remote_key = f"test_files/single_file_{timestamp}.txt"
|
||
|
||
result = toolkit.upload_file(test_file, remote_key)
|
||
test_results['upload_file'] = result['success']
|
||
if result['success']:
|
||
print(f"✅ 文件上传成功: {remote_key}")
|
||
uploaded_file_key = remote_key
|
||
else:
|
||
print(f"❌ 文件上传失败: {result['error']}")
|
||
uploaded_file_key = None
|
||
|
||
os.unlink(test_file)
|
||
|
||
# 3. 测试创建文件夹
|
||
print("\n3. 测试创建文件夹...")
|
||
folder_path = f"test_folders/folder_{timestamp}/"
|
||
result = toolkit.create_folder(folder_path)
|
||
test_results['create_folder'] = result['success']
|
||
if result['success']:
|
||
print(f"✅ 文件夹创建成功: {folder_path}")
|
||
else:
|
||
print(f"❌ 文件夹创建失败: {result['error']}")
|
||
|
||
# 4. 测试上传目录
|
||
print("\n4. 测试上传目录...")
|
||
test_dir = create_test_files()
|
||
remote_prefix = f"test_directories/dir_{timestamp}/"
|
||
|
||
result = toolkit.upload_directory(test_dir, remote_prefix)
|
||
test_results['upload_directory'] = result['success']
|
||
if result['success']:
|
||
print(f"✅ 目录上传成功: 上传了 {result['file_count']} 个文件")
|
||
uploaded_dir_prefix = remote_prefix
|
||
else:
|
||
print(f"❌ 目录上传失败: {result['error']}")
|
||
uploaded_dir_prefix = None
|
||
|
||
shutil.rmtree(test_dir)
|
||
|
||
# 5. 测试下载文件
|
||
print("\n5. 测试下载文件...")
|
||
if uploaded_file_key:
|
||
download_path = tempfile.mktemp(suffix='_downloaded.txt')
|
||
result = toolkit.download_file(uploaded_file_key, download_path)
|
||
test_results['download_file'] = result['success']
|
||
if result['success']:
|
||
print(f"✅ 文件下载成功: {download_path}")
|
||
# 验证内容
|
||
with open(download_path, 'r', encoding='utf-8') as f:
|
||
downloaded_content = f.read()
|
||
if test_content == downloaded_content:
|
||
print("✅ 文件内容验证成功")
|
||
else:
|
||
print("❌ 文件内容验证失败")
|
||
os.unlink(download_path)
|
||
else:
|
||
print(f"❌ 文件下载失败: {result['error']}")
|
||
else:
|
||
test_results['download_file'] = False
|
||
print("❌ 跳过文件下载测试（没有可下载的文件）")
|
||
|
||
# 6. 测试下载目录
|
||
print("\n6. 测试下载目录...")
|
||
if uploaded_dir_prefix:
|
||
download_dir = tempfile.mkdtemp(prefix='s3_download_')
|
||
result = toolkit.download_directory(uploaded_dir_prefix, download_dir)
|
||
test_results['download_directory'] = result['success']
|
||
if result['success']:
|
||
print(f"✅ 目录下载成功: 下载了 {result['file_count']} 个文件")
|
||
else:
|
||
print(f"❌ 目录下载失败: {result['error']}")
|
||
shutil.rmtree(download_dir)
|
||
else:
|
||
test_results['download_directory'] = False
|
||
print("❌ 跳过目录下载测试（没有可下载的目录）")
|
||
|
||
# 7. 测试列出文件
|
||
print("\n7. 测试列出文件...")
|
||
result = toolkit.list_files(max_keys=20)
|
||
test_results['list_files'] = result['success']
|
||
if result['success']:
|
||
print(f"✅ 文件列表获取成功: 找到 {result['file_count']} 个文件")
|
||
if result['files']:
|
||
print("📄 前几个文件:")
|
||
for i, file_info in enumerate(result['files'][:5], 1):
|
||
size_mb = file_info['size'] / (1024 * 1024)
|
||
print(f" {i}. {file_info['key']} ({size_mb:.2f} MB)")
|
||
else:
|
||
print(f"❌ 文件列表获取失败: {result['error']}")
|
||
|
||
# 8. 测试删除文件
|
||
print("\n8. 测试删除文件...")
|
||
if uploaded_file_key:
|
||
result = toolkit.delete_file(uploaded_file_key)
|
||
test_results['delete_file'] = result['success']
|
||
if result['success']:
|
||
print(f"✅ 文件删除成功: {uploaded_file_key}")
|
||
else:
|
||
print(f"❌ 文件删除失败: {result['error']}")
|
||
else:
|
||
test_results['delete_file'] = False
|
||
print("❌ 跳过文件删除测试（没有可删除的文件）")
|
||
|
||
# 9. 测试删除目录
|
||
print("\n9. 测试删除目录...")
|
||
if uploaded_dir_prefix:
|
||
result = toolkit.delete_directory(uploaded_dir_prefix)
|
||
test_results['delete_directory'] = result['success']
|
||
if result['success']:
|
||
print(f"✅ 目录删除成功: 删除了 {result['deleted_count']} 个文件")
|
||
else:
|
||
print(f"❌ 目录删除失败: {result['error']}")
|
||
else:
|
||
test_results['delete_directory'] = False
|
||
print("❌ 跳过目录删除测试（没有可删除的目录）")
|
||
|
||
# 打印测试摘要
|
||
print("\n" + "="*60)
|
||
print("🎯 测试结果摘要")
|
||
print("="*60)
|
||
|
||
test_names = {
|
||
'connection': 'Connection',
|
||
'upload_file': 'Upload File',
|
||
'create_folder': 'Create Folder',
|
||
'upload_directory': 'Upload Directory',
|
||
'download_file': 'Download File',
|
||
'download_directory': 'Download Directory',
|
||
'list_files': 'List Files',
|
||
'delete_file': 'Delete File',
|
||
'delete_directory': 'Delete Directory'
|
||
}
|
||
|
||
passed_tests = 0
|
||
for test_key, test_name in test_names.items():
|
||
status = "✅ 通过" if test_results.get(test_key, False) else "❌ 失败"
|
||
print(f"{test_name:20} : {status}")
|
||
if test_results.get(test_key, False):
|
||
passed_tests += 1
|
||
|
||
print("-" * 60)
|
||
print(f"总计: {passed_tests}/{len(test_names)} 个测试通过")
|
||
|
||
if passed_tests == len(test_names):
|
||
print("🎉 所有测试都通过了！RustFS S3 Storage Toolkit 工作正常。")
|
||
elif passed_tests > 0:
|
||
print("⚠️ 部分测试通过。请检查失败的测试项。")
|
||
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()
|