Victor Phan
237 lines
7.9 KiB
Python
237 lines
7.9 KiB
Python
#!/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)
|