238 lines
9.4 KiB
Python
238 lines
9.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for batch operations functionality.
|
|
Tests batch deletion and recovery operations for shots and assets.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
from datetime import datetime
|
|
from sqlalchemy.orm import Session
|
|
|
|
# Add the backend directory to the path
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from database import SessionLocal
|
|
from models.user import User
|
|
from models.shot import Shot
|
|
from models.asset import Asset
|
|
from models.task import Task
|
|
from services.batch_operations import BatchOperationsService
|
|
|
|
|
|
def test_batch_operations():
|
|
"""Test batch operations functionality."""
|
|
|
|
print("=== Testing Batch Operations ===\n")
|
|
|
|
# Create database session
|
|
db = SessionLocal()
|
|
|
|
try:
|
|
# Get a test user (admin)
|
|
admin_user = db.query(User).filter(User.is_admin == True).first()
|
|
if not admin_user:
|
|
print("❌ No admin user found. Please create an admin user first.")
|
|
return False
|
|
|
|
print(f"Using admin user: {admin_user.first_name} {admin_user.last_name}")
|
|
|
|
# Get some test shots and assets
|
|
test_shots = db.query(Shot).filter(Shot.deleted_at.is_(None)).limit(3).all()
|
|
test_assets = db.query(Asset).filter(Asset.deleted_at.is_(None)).limit(2).all()
|
|
|
|
if not test_shots and not test_assets:
|
|
print("❌ No active shots or assets found for testing.")
|
|
return False
|
|
|
|
print(f"Found {len(test_shots)} shots and {len(test_assets)} assets for testing")
|
|
|
|
# Initialize batch operations service
|
|
batch_service = BatchOperationsService()
|
|
|
|
# Test 1: Batch deletion preview
|
|
print("\n--- Test 1: Batch Deletion Preview ---")
|
|
shot_ids = [shot.id for shot in test_shots]
|
|
asset_ids = [asset.id for asset in test_assets]
|
|
|
|
preview = batch_service.get_batch_deletion_preview(shot_ids, asset_ids, db)
|
|
|
|
print(f"Preview Results:")
|
|
print(f" Total items: {preview['total_items']}")
|
|
print(f" Estimated tasks: {preview['estimated_tasks']}")
|
|
print(f" Estimated submissions: {preview['estimated_submissions']}")
|
|
print(f" Estimated attachments: {preview['estimated_attachments']}")
|
|
print(f" Estimated notes: {preview['estimated_notes']}")
|
|
print(f" Estimated reviews: {preview['estimated_reviews']}")
|
|
print(f" Affected users: {preview['affected_users_count']}")
|
|
print(f" Projects affected: {preview['projects_affected_count']}")
|
|
|
|
# Test 2: Batch shot deletion
|
|
if test_shots:
|
|
print("\n--- Test 2: Batch Shot Deletion ---")
|
|
shot_ids_to_delete = [shot.id for shot in test_shots[:2]] # Delete first 2 shots
|
|
|
|
result = batch_service.batch_delete_shots(shot_ids_to_delete, db, admin_user, batch_size=10)
|
|
|
|
print(f"Batch Deletion Results:")
|
|
print(f" Total items: {result.total_items}")
|
|
print(f" Successful deletions: {result.successful_deletions}")
|
|
print(f" Failed deletions: {result.failed_deletions}")
|
|
print(f" Operation duration: {result.operation_duration:.2f}s")
|
|
print(f" Total deleted tasks: {result.total_deleted_tasks}")
|
|
print(f" Total deleted submissions: {result.total_deleted_submissions}")
|
|
|
|
for item in result.items:
|
|
status = "✓" if item.success else "✗"
|
|
print(f" {status} Shot {item.id} ({item.name}): {item.error or 'Success'}")
|
|
|
|
# Test 3: Batch asset deletion
|
|
if test_assets:
|
|
print("\n--- Test 3: Batch Asset Deletion ---")
|
|
asset_ids_to_delete = [asset.id for asset in test_assets[:1]] # Delete first asset
|
|
|
|
result = batch_service.batch_delete_assets(asset_ids_to_delete, db, admin_user, batch_size=10)
|
|
|
|
print(f"Batch Deletion Results:")
|
|
print(f" Total items: {result.total_items}")
|
|
print(f" Successful deletions: {result.successful_deletions}")
|
|
print(f" Failed deletions: {result.failed_deletions}")
|
|
print(f" Operation duration: {result.operation_duration:.2f}s")
|
|
print(f" Total deleted tasks: {result.total_deleted_tasks}")
|
|
|
|
for item in result.items:
|
|
status = "✓" if item.success else "✗"
|
|
print(f" {status} Asset {item.id} ({item.name}): {item.error or 'Success'}")
|
|
|
|
# Test 4: Batch shot recovery
|
|
if test_shots:
|
|
print("\n--- Test 4: Batch Shot Recovery ---")
|
|
shot_ids_to_recover = [shot.id for shot in test_shots[:2]]
|
|
|
|
result = batch_service.batch_recover_shots(shot_ids_to_recover, db, admin_user, batch_size=10)
|
|
|
|
print(f"Batch Recovery Results:")
|
|
print(f" Total items: {result.total_items}")
|
|
print(f" Successful recoveries: {result.successful_recoveries}")
|
|
print(f" Failed recoveries: {result.failed_recoveries}")
|
|
print(f" Operation duration: {result.operation_duration:.2f}s")
|
|
print(f" Total recovered tasks: {result.total_recovered_tasks}")
|
|
|
|
for item in result.items:
|
|
status = "✓" if item.success else "✗"
|
|
print(f" {status} Shot {item.id} ({item.name}): {item.error or 'Success'}")
|
|
|
|
# Test 5: Batch asset recovery
|
|
if test_assets:
|
|
print("\n--- Test 5: Batch Asset Recovery ---")
|
|
asset_ids_to_recover = [asset.id for asset in test_assets[:1]]
|
|
|
|
result = batch_service.batch_recover_assets(asset_ids_to_recover, db, admin_user, batch_size=10)
|
|
|
|
print(f"Batch Recovery Results:")
|
|
print(f" Total items: {result.total_items}")
|
|
print(f" Successful recoveries: {result.successful_recoveries}")
|
|
print(f" Failed recoveries: {result.failed_recoveries}")
|
|
print(f" Operation duration: {result.operation_duration:.2f}s")
|
|
print(f" Total recovered tasks: {result.total_recovered_tasks}")
|
|
|
|
for item in result.items:
|
|
status = "✓" if item.success else "✗"
|
|
print(f" {status} Asset {item.id} ({item.name}): {item.error or 'Success'}")
|
|
|
|
# Test 6: Batch size optimization
|
|
print("\n--- Test 6: Batch Size Optimization ---")
|
|
|
|
# Test with different scenarios
|
|
scenarios = [
|
|
(100, 0, "Simple items with no related data"),
|
|
(50, 500, "Moderate complexity"),
|
|
(20, 2000, "High complexity"),
|
|
(10, 5000, "Very high complexity")
|
|
]
|
|
|
|
for total_items, related_items, description in scenarios:
|
|
optimal_size = batch_service.optimize_batch_size(total_items, related_items)
|
|
print(f" {description}: {total_items} items, {related_items} related → batch size: {optimal_size}")
|
|
|
|
print("\n✓ All batch operations tests completed successfully!")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Test failed with error: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_performance_monitoring():
|
|
"""Test performance monitoring for batch operations."""
|
|
|
|
print("\n=== Testing Performance Monitoring ===\n")
|
|
|
|
db = SessionLocal()
|
|
|
|
try:
|
|
# Count current database records
|
|
shot_count = db.query(Shot).count()
|
|
asset_count = db.query(Asset).count()
|
|
task_count = db.query(Task).count()
|
|
|
|
print(f"Current Database State:")
|
|
print(f" Shots: {shot_count}")
|
|
print(f" Assets: {asset_count}")
|
|
print(f" Tasks: {task_count}")
|
|
|
|
# Count deleted records
|
|
deleted_shots = db.query(Shot).filter(Shot.deleted_at.isnot(None)).count()
|
|
deleted_assets = db.query(Asset).filter(Asset.deleted_at.isnot(None)).count()
|
|
deleted_tasks = db.query(Task).filter(Task.deleted_at.isnot(None)).count()
|
|
|
|
print(f"\nDeleted Records:")
|
|
print(f" Deleted shots: {deleted_shots}")
|
|
print(f" Deleted assets: {deleted_assets}")
|
|
print(f" Deleted tasks: {deleted_tasks}")
|
|
|
|
# Performance metrics
|
|
batch_service = BatchOperationsService()
|
|
|
|
# Test optimal batch sizes for different scenarios
|
|
print(f"\nOptimal Batch Sizes:")
|
|
test_cases = [
|
|
(10, 50),
|
|
(100, 500),
|
|
(500, 2500),
|
|
(1000, 10000)
|
|
]
|
|
|
|
for items, related in test_cases:
|
|
optimal = batch_service.optimize_batch_size(items, related)
|
|
complexity = related / items if items > 0 else 0
|
|
print(f" {items} items, {related} related (complexity: {complexity:.1f}) → batch size: {optimal}")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ Performance monitoring test failed: {str(e)}")
|
|
return False
|
|
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("Batch Operations Test Suite")
|
|
print("=" * 50)
|
|
|
|
success1 = test_batch_operations()
|
|
success2 = test_performance_monitoring()
|
|
|
|
if success1 and success2:
|
|
print("\n🎉 All tests passed!")
|
|
sys.exit(0)
|
|
else:
|
|
print("\n❌ Some tests failed!")
|
|
sys.exit(1) |