Uploaded code
This commit is contained in:
111
test_storage.py
Normal file
111
test_storage.py
Normal file
@@ -0,0 +1,111 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user