112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for storage backends
|
|
"""
|
|
import sys
|
|
import os
|
|
from pathlib import Path
|
|
|
|
# Add src directory to Python path
|
|
sys.path.insert(0, str(Path(__file__).parent / "src"))
|
|
|
|
from config import config
|
|
from storage import StorageManager
|
|
|
|
def test_storage():
|
|
"""Test the storage backend"""
|
|
print("🧪 Testing storage backend...")
|
|
|
|
try:
|
|
# Initialize storage manager
|
|
storage = StorageManager(config)
|
|
backend_info = storage.get_backend_info()
|
|
|
|
print(f"✅ Storage backend initialized: {backend_info['type']}")
|
|
|
|
if backend_info['type'] == 'b2':
|
|
print(f" → B2 bucket: {backend_info.get('bucket', 'unknown')}")
|
|
elif backend_info['type'] == 'local':
|
|
print(f" → Local path: {backend_info.get('path', 'unknown')}")
|
|
|
|
# Test file upload
|
|
test_content = b"Hello, this is a test file for storage backend testing!"
|
|
test_file_path = "test/test_file.txt"
|
|
|
|
print(f"\n📤 Testing file upload...")
|
|
success = storage.upload_file(test_content, test_file_path, "text/plain")
|
|
|
|
if success:
|
|
print(f"✅ Upload successful: {test_file_path}")
|
|
|
|
# Test file download
|
|
print(f"📥 Testing file download...")
|
|
downloaded_content = storage.download_file(test_file_path)
|
|
|
|
if downloaded_content == test_content:
|
|
print(f"✅ Download successful and content matches")
|
|
|
|
# Test file exists
|
|
print(f"🔍 Testing file exists...")
|
|
exists = storage.file_exists(test_file_path)
|
|
print(f"✅ File exists: {exists}")
|
|
|
|
# Test file size
|
|
print(f"📏 Testing file size...")
|
|
size = storage.get_file_size(test_file_path)
|
|
print(f"✅ File size: {size} bytes")
|
|
|
|
# Test file listing
|
|
print(f"📋 Testing file listing...")
|
|
files = storage.list_files("test/")
|
|
print(f"✅ Files in test/ folder: {len(files)} files")
|
|
for file in files:
|
|
print(f" → {file}")
|
|
|
|
# Test cleanup
|
|
print(f"🗑️ Testing file deletion...")
|
|
deleted = storage.delete_file(test_file_path)
|
|
print(f"✅ File deleted: {deleted}")
|
|
|
|
else:
|
|
print(f"❌ Downloaded content doesn't match original")
|
|
return False
|
|
else:
|
|
print(f"❌ Upload failed")
|
|
return False
|
|
|
|
print(f"\n🎉 All storage tests passed!")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Storage test failed: {e}")
|
|
return False
|
|
|
|
def main():
|
|
print("🚀 Sharey Storage Backend Test")
|
|
print("=" * 40)
|
|
|
|
# Show current configuration
|
|
print("\n📋 Current Storage Configuration:")
|
|
storage_config = config.get_storage_config()
|
|
print(f" Backend: {storage_config.get('backend', 'unknown')}")
|
|
|
|
if storage_config.get('backend') == 'local':
|
|
print(f" Local Path: {storage_config.get('local_path', 'storage')}")
|
|
elif storage_config.get('backend') == 'b2':
|
|
if config.validate_b2_config():
|
|
b2_config = config.get_b2_config()
|
|
print(f" B2 Bucket: {b2_config.get('bucket_name', 'unknown')}")
|
|
else:
|
|
print(" B2 Configuration: Invalid")
|
|
|
|
# Test the storage backend
|
|
if test_storage():
|
|
print("\n✅ Storage backend is working correctly!")
|
|
sys.exit(0)
|
|
else:
|
|
print("\n❌ Storage backend test failed!")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|