WTOAAM/repositoryZN/utils/minioUtil/unitTest.py

from minioClientPool import MinioClientPool
from threadSafeMinioClient import ThreadSafeMinioClient

# Example Usage
if __name__ == "__main__":
    # Upload a file
    bucket_name = 'test'
    test_bucket_name="test-xzy"

    pool = MinioClientPool('192.168.50.234:6900', '6VkF2ul6X7udr7RLsG2W', 'jtBuqZ80biRWQf6sbwzDQJwHtEBicPjkZBtvjTrA')
    ts_minio_client = ThreadSafeMinioClient(pool)
    
    # Check if a bucket exists
    exists = ts_minio_client.bucket_exists(test_bucket_name)
    print("Bucket exists:", exists)

    # Create a new bucket if it doesn't exist
    if not exists:
        ts_minio_client.create_bucket(test_bucket_name)
        print("Bucket create successfully:",test_bucket_name)
    else :
        ts_minio_client.delete_bucket(test_bucket_name)
        print("Bucket delete successfully:",test_bucket_name)

    # List all buckets
    buckets = ts_minio_client.get_buckets()
    print("Buckets:", [bucket.name for bucket in buckets])

    # Delete a bucket
    # Make sure to handle this carefully, as it requires the bucket to be empty
    ts_minio_client.delete_bucket('your-old-bucket')

    object_name = 'new-example-object.txt'
    file_path = 'data/upload/path-to-your-local-file.txt'  # Path to the local file to upload
    upload_result = ts_minio_client.upload_file(test_bucket_name, object_name, file_path)
    print(f"File uploaded successfully: {upload_result}")

    # List of files to upload (object_name, file_path)
    files_to_upload = [
        ('new-example-object1.txt', r'data/upload/path-to-your-local-file1.txt'),
        ('new-example-object2.txt', r'data/upload/path-to-your-local-file2.txt')
    ]

    # Upload files
    upload_results = ts_minio_client.upload_files(test_bucket_name, files_to_upload)
    print("Files uploaded successfully:")
    for result in upload_results:
        print(result)

     # Download a file
    save_path = r'data/download/local-example-object.txt'  # The path where you want to save the file
    ts_minio_client.download_file(test_bucket_name, object_name, save_path)
    print(f"File downloaded successfully and saved to {save_path}")
    
    # List of files to download (object_name, local_save_path)
    files_to_download = [
        ('new-example-object1.txt', r'data/download/local-example-object1.txt'),
        ('new-example-object1.txt', r'data/download/local-example-object2.txt')
    ]

    # Download files
    ts_minio_client.download_files(test_bucket_name, files_to_download)
    print("Files downloaded successfully.")