Olivier Michel

#!/usr/bin/env python3
"""
Simple script to test S3 connectivity and download.
Usage: python s3_test.py
"""

import os
import time
import tempfile
from minio import Minio
from minio.error import S3Error
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

def get_minio_client():
    """Create and return a Minio client using credentials from .env file"""
    endpoint = os.getenv("S3_ENDPOINT")
    access_key = os.getenv("S3_ACCESS_KEY")
    secret_key = os.getenv("S3_SECRET_KEY")
    secure = True 
    
    print(f"Connecting to S3 endpoint: {endpoint} (secure={secure})")
    
    return Minio(
        endpoint=endpoint,
        access_key=access_key,
        secret_key=secret_key,
        secure=secure
    )

def test_connection():
    """Test basic connectivity to S3"""
    try:
        client = get_minio_client()
        
        # Try to list buckets as a basic connectivity test
        buckets = client.list_buckets()
        print(f"Successfully connected to S3. Found {len(buckets)} buckets:")
        for bucket in buckets:
            print(f"  - {bucket.name}")
        
        return client
    except Exception as e:
        print(f"Failed to connect to S3: {e}")
        return None

def test_bucket_access(client, bucket_name):
    """Test access to a specific bucket"""
    try:
        exists = client.bucket_exists(bucket_name)
        if exists:
            print(f"Bucket '{bucket_name}' exists and is accessible")
            
            # List a few objects to verify access
            count = 0
            objects = []
            for obj in client.list_objects(bucket_name, recursive=True):
                print(f"  - {obj.object_name} ({obj.size} bytes)")
                objects.append(obj)
                count += 1
                if count >= 5:
                    break
            
            print(f"Found at least {len(objects)} objects in bucket")
            return objects
        else:
            print(f"Bucket '{bucket_name}' does not exist or is not accessible")
            return []
    except Exception as e:
        print(f"Error accessing bucket '{bucket_name}': {e}")
        return []

def test_download(client, bucket_name, object_name):
    """Test downloading a specific object"""
    try:
        print(f"Attempting to download: {object_name}")
        start_time = time.time()
        
        # Create a temporary file to store the downloaded object
        with tempfile.NamedTemporaryFile(delete=False) as temp_file:
            temp_path = temp_file.name
        
        # Download the object
        client.fget_object(bucket_name, object_name, temp_path)
        
        # Check if the file was downloaded successfully
        file_size = os.path.getsize(temp_path)
        elapsed_time = time.time() - start_time
        
        print(f"Download successful: {file_size} bytes in {elapsed_time:.2f} seconds")
        print(f"Average download speed: {file_size/elapsed_time/1024:.2f} KB/s")
        
        # Clean up
        os.unlink(temp_path)
        return True
    except S3Error as e:
        print(f"S3 Error during download: {e}")
        return False
    except Exception as e:
        print(f"Error downloading object: {e}")
        return False

def main():
    """Main function to run all tests"""
    print("=== S3 Connectivity Test ===")
    
    # Test basic connectivity
    client = test_connection()
    if not client:
        return
    
    # Get bucket name from environment variable
    bucket_name = "redb"
    print(f"\n=== Testing Access to Bucket: {bucket_name} ===")
    objects = test_bucket_access(client, bucket_name)
    
    if objects:
        # Test downloading the first object
        first_object = objects[0].object_name
        print(f"\n=== Testing Download of Object: {first_object} ===")
        test_download(client, bucket_name, first_object)
        
        # If the first one worked, let's try a specific file from the error log
        print(f"\n=== Testing Download of Specific Object ===")
        specific_object = "00/06/00062073114defac39cb7ab9cb99a0b77c535a09a241c60d47ddd0b6431629f1.zip"
        test_download(client, bucket_name, specific_object)

if __name__ == "__main__":
    main()