#!/usr/bin/env python3 """ Test file for direct AWS S3 access File test truy cập trực tiếp AWS S3 """ import os import sys import boto3 from botocore.exceptions import ClientError, NoCredentialsError def load_credentials(credential_file='train_files/crediential.txt'): """ Load AWS credentials from file Đọc thông tin xác thực AWS từ file """ credentials = {} try: with open(credential_file, 'r') as f: for line in f: line = line.strip() if line.startswith('export '): # Remove 'export ' prefix line = line[7:] if '=' in line: key, value = line.split('=', 1) # Remove quotes if present value = value.strip('"') credentials[key] = value print("✓ Credentials loaded successfully") print(f" - AWS_ACCESS_KEY_ID: {credentials.get('AWS_ACCESS_KEY_ID', 'N/A')[:20]}...") return credentials except FileNotFoundError: print(f"✗ Error: Credential file not found: {credential_file}") return None except Exception as e: print(f"✗ Error loading credentials: {e}") return None def set_aws_credentials(credentials): """ Set AWS credentials as environment variables Thiết lập thông tin xác thực AWS vào biến môi trường """ if not credentials: return False try: os.environ['AWS_ACCESS_KEY_ID'] = credentials.get('AWS_ACCESS_KEY_ID', '') os.environ['AWS_SECRET_ACCESS_KEY'] = credentials.get('AWS_SECRET_ACCESS_KEY', '') os.environ['AWS_SESSION_TOKEN'] = credentials.get('AWS_SESSION_TOKEN', '') print("✓ AWS credentials set in environment") return True except Exception as e: print(f"✗ Error setting credentials: {e}") return False def test_s3_connection(region='ap-southeast-1'): """ Test S3 connection and list buckets Kiểm tra kết nối S3 và liệt kê các bucket """ try: # Create S3 client s3_client = boto3.client('s3', region_name=region) print("\n=== Testing S3 Connection ===") print("Đang kiểm tra kết nối S3...") # List buckets response = s3_client.list_buckets() print(f"\n✓ Successfully connected to S3!") print(f"✓ Total buckets found: {len(response['Buckets'])}") print("\nAvailable S3 Buckets:") print("Các S3 Bucket có sẵn:") for i, bucket in enumerate(response['Buckets'], 1): print(f" {i}. {bucket['Name']} (Created: {bucket['CreationDate']})") return s3_client, response['Buckets'] except NoCredentialsError: print("\n✗ Error: No AWS credentials found") print("✗ Lỗi: Không tìm thấy thông tin xác thực AWS") return None, None except ClientError as e: print(f"\n✗ AWS Client Error: {e}") error_code = e.response.get('Error', {}).get('Code', 'Unknown') print(f"✗ Error Code: {error_code}") if error_code == 'ExpiredToken': print("✗ AWS session token has expired. Please refresh your credentials.") print("✗ Token AWS đã hết hạn. Vui lòng làm mới thông tin xác thực.") return None, None except Exception as e: print(f"\n✗ Unexpected error: {e}") return None, None def test_bucket_access(s3_client, bucket_name, max_objects=10): """ Test access to a specific bucket and list objects Kiểm tra truy cập bucket cụ thể và liệt kê các object """ try: print(f"\n=== Testing Bucket Access: {bucket_name} ===") print(f"Đang kiểm tra truy cập bucket: {bucket_name}") # List objects in bucket response = s3_client.list_objects_v2( Bucket=bucket_name, MaxKeys=max_objects ) if 'Contents' in response: print(f"\n✓ Successfully accessed bucket: {bucket_name}") print(f"✓ Found {len(response['Contents'])} objects (showing max {max_objects}):") print("\nObjects in bucket:") for i, obj in enumerate(response['Contents'], 1): size_mb = obj['Size'] / (1024 * 1024) print(f" {i}. {obj['Key']}") print(f" Size: {size_mb:.2f} MB | Modified: {obj['LastModified']}") else: print(f"\n✓ Bucket {bucket_name} is accessible but empty") return True except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') print(f"\n✗ Error accessing bucket {bucket_name}") print(f"✗ Error Code: {error_code}") if error_code == 'NoSuchBucket': print("✗ Bucket does not exist") elif error_code == 'AccessDenied': print("✗ Access denied to this bucket") return False except Exception as e: print(f"\n✗ Unexpected error: {e}") return False def test_download_object(s3_client, bucket_name, object_key, local_path='test_download'): """ Test downloading an object from S3 Kiểm tra tải xuống object từ S3 """ try: print(f"\n=== Testing Object Download ===") print(f"Bucket: {bucket_name}") print(f"Object: {object_key}") print(f"Local path: {local_path}") # Create local directory if not exists os.makedirs(os.path.dirname(local_path) if os.path.dirname(local_path) else '.', exist_ok=True) # Download object s3_client.download_file(bucket_name, object_key, local_path) file_size = os.path.getsize(local_path) print(f"\n✓ Successfully downloaded: {object_key}") print(f"✓ File size: {file_size / (1024*1024):.2f} MB") print(f"✓ Saved to: {local_path}") return True except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') print(f"\n✗ Error downloading object") print(f"✗ Error Code: {error_code}") return False except Exception as e: print(f"\n✗ Unexpected error: {e}") return False def main(): """ Main test function Hàm test chính """ print("=" * 60) print("AWS S3 Direct Access Test") print("Test Truy Cập Trực Tiếp AWS S3") print("=" * 60) # Step 1: Load credentials print("\n[Step 1] Loading credentials from file...") credentials = load_credentials() if not credentials: print("\n✗ Test failed: Could not load credentials") return False # Step 2: Set credentials in environment print("\n[Step 2] Setting AWS credentials...") if not set_aws_credentials(credentials): print("\n✗ Test failed: Could not set credentials") return False # Step 3: Test S3 connection print("\n[Step 3] Testing S3 connection...") s3_client, buckets = test_s3_connection() if not s3_client: print("\n✗ Test failed: Could not connect to S3") return False # Step 4: Test specific bucket access (if buckets exist) if buckets and len(buckets) > 0: print("\n[Step 4] Testing bucket access...") first_bucket = buckets[0]['Name'] test_bucket_access(s3_client, first_bucket, max_objects=5) print("\n" + "=" * 60) print("✓ Test completed!") print("✓ Test hoàn thành!") print("=" * 60) return True if __name__ == "__main__": try: success = main() sys.exit(0 if success else 1) except KeyboardInterrupt: print("\n\n✗ Test interrupted by user") sys.exit(1) except Exception as e: print(f"\n✗ Fatal error: {e}") import traceback traceback.print_exc() sys.exit(1)