141 lines
4.6 KiB
Python
141 lines
4.6 KiB
Python
"""
|
|
Test script for soft deletion services.
|
|
|
|
This script tests the basic functionality of the soft deletion services
|
|
to ensure they work correctly with the database.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
sys.path.append('.')
|
|
|
|
from sqlalchemy.orm import sessionmaker
|
|
from database import engine, get_db
|
|
from models.shot import Shot
|
|
from models.asset import Asset
|
|
from models.user import User
|
|
from models.episode import Episode
|
|
from models.project import Project
|
|
from services.shot_soft_deletion import ShotSoftDeletionService
|
|
from services.asset_soft_deletion import AssetSoftDeletionService
|
|
from services.recovery_service import RecoveryService
|
|
|
|
|
|
def test_shot_soft_deletion_service():
|
|
"""Test the ShotSoftDeletionService."""
|
|
print("Testing ShotSoftDeletionService...")
|
|
|
|
# Get database session
|
|
db = next(get_db())
|
|
|
|
try:
|
|
# Get a test shot (first active shot)
|
|
shot = db.query(Shot).filter(Shot.deleted_at.is_(None)).first()
|
|
if not shot:
|
|
print("No active shots found for testing")
|
|
return
|
|
|
|
print(f"Found test shot: {shot.name} (ID: {shot.id})")
|
|
|
|
# Initialize service
|
|
service = ShotSoftDeletionService()
|
|
|
|
# Test getting deletion info
|
|
deletion_info = service.get_deletion_info(shot.id, db)
|
|
if deletion_info:
|
|
print(f"Deletion info retrieved successfully:")
|
|
print(f" - Shot: {deletion_info.shot_name}")
|
|
print(f" - Tasks: {deletion_info.task_count}")
|
|
print(f" - Submissions: {deletion_info.submission_count}")
|
|
print(f" - Attachments: {deletion_info.attachment_count}")
|
|
print(f" - Notes: {deletion_info.note_count}")
|
|
print(f" - Reviews: {deletion_info.review_count}")
|
|
print(f" - Affected users: {len(deletion_info.affected_users)}")
|
|
else:
|
|
print("Failed to get deletion info")
|
|
|
|
print("ShotSoftDeletionService test completed successfully")
|
|
|
|
except Exception as e:
|
|
print(f"Error testing ShotSoftDeletionService: {str(e)}")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_asset_soft_deletion_service():
|
|
"""Test the AssetSoftDeletionService."""
|
|
print("\nTesting AssetSoftDeletionService...")
|
|
|
|
# Get database session
|
|
db = next(get_db())
|
|
|
|
try:
|
|
# Get a test asset (first active asset)
|
|
asset = db.query(Asset).filter(Asset.deleted_at.is_(None)).first()
|
|
if not asset:
|
|
print("No active assets found for testing")
|
|
return
|
|
|
|
print(f"Found test asset: {asset.name} (ID: {asset.id})")
|
|
|
|
# Initialize service
|
|
service = AssetSoftDeletionService()
|
|
|
|
# Test getting deletion info
|
|
deletion_info = service.get_deletion_info(asset.id, db)
|
|
if deletion_info:
|
|
print(f"Deletion info retrieved successfully:")
|
|
print(f" - Asset: {deletion_info.asset_name}")
|
|
print(f" - Category: {deletion_info.asset_category}")
|
|
print(f" - Tasks: {deletion_info.task_count}")
|
|
print(f" - Submissions: {deletion_info.submission_count}")
|
|
print(f" - Attachments: {deletion_info.attachment_count}")
|
|
print(f" - Notes: {deletion_info.note_count}")
|
|
print(f" - Reviews: {deletion_info.review_count}")
|
|
print(f" - Affected users: {len(deletion_info.affected_users)}")
|
|
else:
|
|
print("Failed to get deletion info")
|
|
|
|
print("AssetSoftDeletionService test completed successfully")
|
|
|
|
except Exception as e:
|
|
print(f"Error testing AssetSoftDeletionService: {str(e)}")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_recovery_service():
|
|
"""Test the RecoveryService."""
|
|
print("\nTesting RecoveryService...")
|
|
|
|
# Get database session
|
|
db = next(get_db())
|
|
|
|
try:
|
|
# Initialize service
|
|
service = RecoveryService()
|
|
|
|
# Test getting deleted shots
|
|
deleted_shots = service.get_deleted_shots(None, db)
|
|
print(f"Found {len(deleted_shots)} deleted shots")
|
|
|
|
# Test getting deleted assets
|
|
deleted_assets = service.get_deleted_assets(None, db)
|
|
print(f"Found {len(deleted_assets)} deleted assets")
|
|
|
|
print("RecoveryService test completed successfully")
|
|
|
|
except Exception as e:
|
|
print(f"Error testing RecoveryService: {str(e)}")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("Starting soft deletion services tests...")
|
|
|
|
test_shot_soft_deletion_service()
|
|
test_asset_soft_deletion_service()
|
|
test_recovery_service()
|
|
|
|
print("\nAll tests completed!") |