171 lines
5.8 KiB
Python
171 lines
5.8 KiB
Python
import unittest
|
|
import os
|
|
import tempfile
|
|
import hashlib
|
|
from datetime import datetime
|
|
from utils.minio_agent import MinIOAgent # 导入之前的MinIO操作类
|
|
|
|
|
|
class TestMinIOAgent(unittest.TestCase):
|
|
# 测试配置 - 本地MinIO社区版
|
|
MINIO_CONFIG = {
|
|
'endpoint': '127.0.0.1:9005',
|
|
'access_key': 'admin', # 默认账号
|
|
'secret_key': 'abc88888888', # 默认密码
|
|
'secure': False # 社区版默认不启用SSL
|
|
}
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
"""初始化测试环境"""
|
|
# 创建唯一测试桶（避免冲突）
|
|
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
|
|
cls.test_bucket = f"test-bucket-{timestamp}"
|
|
cls.test_object = "test-data/sample.txt"
|
|
cls.test_content = b"this is MinIO test data: 1234567890"
|
|
|
|
# 初始化客户端
|
|
cls.minio_agent = MinIOAgent(cls.MINIO_CONFIG)
|
|
|
|
# 确保测试桶存在
|
|
cls.minio_agent.create_bucket(cls.test_bucket)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
"""清理测试环境"""
|
|
try:
|
|
# 列出并删除桶内所有对象
|
|
objects = cls.minio_agent.list_objects(cls.test_bucket)
|
|
for obj in objects:
|
|
cls.minio_agent.delete_object(cls.test_bucket, obj['object_name'])
|
|
|
|
# 删除测试桶（MinIO要求桶为空才能删除）
|
|
cls.minio_agent._client.remove_bucket(cls.test_bucket)
|
|
print(f"\n测试清理完成，已删除桶: {cls.test_bucket}")
|
|
except Exception as e:
|
|
print(f"清理测试环境失败: {str(e)}")
|
|
|
|
def test_01_create_bucket(self):
|
|
"""测试创建存储桶"""
|
|
new_bucket = f"temp-bucket-{datetime.now().microsecond}"
|
|
result = self.minio_agent.create_bucket(new_bucket)
|
|
self.assertTrue(result, "存储桶创建失败")
|
|
|
|
# 验证桶是否存在
|
|
exists = self.minio_agent._client.bucket_exists(new_bucket)
|
|
self.assertTrue(exists, "存储桶创建后未检测到存在")
|
|
|
|
# 清理临时桶
|
|
self.minio_agent._client.remove_bucket(new_bucket)
|
|
|
|
def test_02_upload_download(self):
|
|
"""测试上传与下载功能"""
|
|
# 上传数据
|
|
upload_meta = self.minio_agent.upload_bytes(
|
|
bucket=self.test_bucket,
|
|
object_name=self.test_object,
|
|
data=self.test_content
|
|
)
|
|
|
|
# 验证上传结果
|
|
self.assertEqual(upload_meta['size'], len(self.test_content), "上传数据大小不匹配")
|
|
self.assertEqual(upload_meta['local_hash'], hashlib.md5(self.test_content).hexdigest(), "本地哈希校验失败")
|
|
|
|
# 下载数据到临时文件
|
|
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
|
|
temp_path = temp_file.name
|
|
|
|
download_meta = self.minio_agent.download_file(
|
|
bucket=self.test_bucket,
|
|
object_name=self.test_object,
|
|
local_path=temp_path
|
|
)
|
|
|
|
# 验证下载内容
|
|
with open(temp_path, 'rb') as f:
|
|
downloaded_content = f.read()
|
|
|
|
self.assertEqual(downloaded_content, self.test_content, "下载数据与原始数据不匹配")
|
|
self.assertEqual(download_meta['size'], len(self.test_content), "下载文件大小不匹配")
|
|
|
|
# 清理临时文件
|
|
os.unlink(temp_path)
|
|
|
|
def test_03_presigned_url(self):
|
|
"""测试生成预签名URL"""
|
|
# 先上传测试文件
|
|
self.minio_agent.upload_bytes(
|
|
self.test_bucket,
|
|
self.test_object,
|
|
self.test_content
|
|
)
|
|
|
|
# 生成URL（有效期30秒）
|
|
url_info = self.minio_agent.get_presigned_url(
|
|
bucket=self.test_bucket,
|
|
object_name=self.test_object,
|
|
expires=30
|
|
)
|
|
|
|
# 验证URL格式
|
|
self.assertIn("http://127.0.0.1:9005", url_info['presigned_url'], "预签名URL格式不正确")
|
|
self.assertEqual(url_info['expires_in'], 30, "过期时间设置不正确")
|
|
|
|
def test_04_list_objects(self):
|
|
"""测试列出对象功能"""
|
|
# 上传多个测试对象
|
|
test_objects = [
|
|
"test-folder/file1.txt",
|
|
"test-folder/file2.csv",
|
|
"another-folder/image.jpg"
|
|
]
|
|
|
|
for obj in test_objects:
|
|
self.minio_agent.upload_bytes(
|
|
self.test_bucket,
|
|
obj,
|
|
b"tese_list_obj"
|
|
)
|
|
|
|
# 列出所有对象
|
|
all_objects = self.minio_agent.list_objects(self.test_bucket)
|
|
self.assertEqual(len(all_objects), len(test_objects) + 1, "列出对象数量不匹配") # +1是之前的test_object
|
|
|
|
# 按前缀筛选
|
|
filtered_objects = self.minio_agent.list_objects(
|
|
self.test_bucket,
|
|
prefix="test-folder/"
|
|
)
|
|
self.assertEqual(len(filtered_objects), 2, "按前缀筛选结果不正确")
|
|
|
|
def test_05_delete_object(self):
|
|
"""测试删除对象功能"""
|
|
# 创建测试对象
|
|
delete_obj = "to-delete/temp.txt"
|
|
self.minio_agent.upload_bytes(
|
|
self.test_bucket,
|
|
delete_obj,
|
|
b"will be delete"
|
|
)
|
|
|
|
# 执行删除
|
|
result = self.minio_agent.delete_object(self.test_bucket, delete_obj)
|
|
self.assertTrue(result, "删除对象失败")
|
|
|
|
# 验证删除
|
|
objects = self.minio_agent.list_objects(self.test_bucket, prefix="to-delete/")
|
|
self.assertEqual(len(objects), 0, "对象删除后仍存在")
|
|
|
|
def test_06_upload_empty_data(self):
|
|
"""测试上传空数据的异常处理"""
|
|
with self.assertRaises(ValueError, msg="未捕获空数据上传异常"):
|
|
self.minio_agent.upload_bytes(
|
|
self.test_bucket,
|
|
"empty.txt",
|
|
b""
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 执行测试并显示详细结果
|
|
unittest.main(verbosity=2) |