44 lines
1.1 KiB
Python
44 lines
1.1 KiB
Python
import asyncio
|
|
import os
|
|
from config import settings
|
|
from adapters.s3_adapter import S3Adapter
|
|
|
|
async def test_s3():
|
|
|
|
endpoint = settings.MINIO_ENDPOINT
|
|
access_key = settings.MINIO_ACCESS_KEY
|
|
secret_key = settings.MINIO_SECRET_KEY
|
|
bucket = settings.MINIO_BUCKET
|
|
|
|
print(f"Connecting to {endpoint}, bucket: {bucket}")
|
|
|
|
s3 = S3Adapter(endpoint, access_key, secret_key, bucket)
|
|
|
|
test_filename = "test_connection.txt"
|
|
test_data = b"Hello MinIO!"
|
|
|
|
print("Uploading...")
|
|
success = await s3.upload_file(test_filename, test_data)
|
|
if success:
|
|
print("Upload successful!")
|
|
else:
|
|
print("Upload failed!")
|
|
return
|
|
|
|
print("Downloading...")
|
|
data = await s3.get_file(test_filename)
|
|
if data == test_data:
|
|
print("Download successful and data matches!")
|
|
else:
|
|
print(f"Download mismatch: {data}")
|
|
|
|
print("Deleting...")
|
|
deleted = await s3.delete_file(test_filename)
|
|
if deleted:
|
|
print("Delete successful!")
|
|
else:
|
|
print("Delete failed!")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_s3())
|