""" Test script to verify project_id preservation in soft deletion operations. This script tests that project_id is properly preserved during soft deletion and recovery operations for shots. """ 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.user import User from models.episode import Episode from models.project import Project from services.shot_soft_deletion import ShotSoftDeletionService from services.recovery_service import RecoveryService def test_project_id_preservation(): """Test that project_id is preserved during soft deletion and recovery.""" print("Testing project_id preservation in soft deletion operations...") # Get database session db = next(get_db()) try: # Get a test shot with project_id shot = db.query(Shot).filter( Shot.deleted_at.is_(None), Shot.project_id.isnot(None) ).first() if not shot: print("No active shots with project_id found for testing") return print(f"Found test shot: {shot.name} (ID: {shot.id}, Project ID: {shot.project_id})") # Get an admin user for the operations admin_user = db.query(User).filter(User.role.in_(['admin', 'coordinator'])).first() if not admin_user: print("No admin/coordinator user found for testing") return original_project_id = shot.project_id original_project_name = shot.project.name if shot.project else None # Initialize services deletion_service = ShotSoftDeletionService() recovery_service = RecoveryService() # Test 1: Get deletion info and verify project_id is included print("\n1. Testing deletion info includes project_id...") deletion_info = deletion_service.get_deletion_info(shot.id, db) if deletion_info: print(f" ✓ Deletion info retrieved") print(f" ✓ Project ID: {deletion_info.project_id}") print(f" ✓ Project Name: {deletion_info.project_name}") assert deletion_info.project_id == original_project_id, f"Project ID mismatch: {deletion_info.project_id} != {original_project_id}" assert deletion_info.project_name == original_project_name, f"Project name mismatch: {deletion_info.project_name} != {original_project_name}" print(" ✓ Project information correctly preserved in deletion info") else: print(" ✗ Failed to get deletion info") return # Test 2: Perform soft deletion and verify project_id is preserved print("\n2. Testing soft deletion preserves project_id...") # Commit any pending changes before the deletion operation db.commit() deletion_result = deletion_service.soft_delete_shot_cascade(shot.id, db, admin_user) if deletion_result.success: print(f" ✓ Shot soft deleted successfully") # Verify the shot is marked as deleted but project_id is preserved db.refresh(shot) assert shot.deleted_at is not None, "Shot should be marked as deleted" assert shot.project_id == original_project_id, f"Project ID should be preserved: {shot.project_id} != {original_project_id}" print(f" ✓ Project ID preserved after deletion: {shot.project_id}") else: print(f" ✗ Soft deletion failed: {deletion_result.errors}") return # Test 3: Get recovery preview and verify project_id is included print("\n3. Testing recovery preview includes project_id...") recovery_info = recovery_service.preview_shot_recovery(shot.id, db) if recovery_info: print(f" ✓ Recovery info retrieved") print(f" ✓ Project ID: {recovery_info.project_id}") print(f" ✓ Project Name: {recovery_info.project_name}") assert recovery_info.project_id == original_project_id, f"Project ID mismatch in recovery: {recovery_info.project_id} != {original_project_id}" assert recovery_info.project_name == original_project_name, f"Project name mismatch in recovery: {recovery_info.project_name} != {original_project_name}" print(" ✓ Project information correctly preserved in recovery info") else: print(" ✗ Failed to get recovery info") return # Test 4: Perform recovery and verify project_id consistency print("\n4. Testing recovery maintains project_id consistency...") # Commit any pending changes before the recovery operation db.commit() recovery_result = recovery_service.recover_shot(shot.id, db, admin_user) if recovery_result.success: print(f" ✓ Shot recovered successfully") # Verify the shot is no longer marked as deleted and project_id is still correct db.refresh(shot) assert shot.deleted_at is None, "Shot should no longer be marked as deleted" assert shot.project_id == original_project_id, f"Project ID should remain consistent: {shot.project_id} != {original_project_id}" print(f" ✓ Project ID consistent after recovery: {shot.project_id}") else: print(f" ✗ Recovery failed: {recovery_result.errors}") return # Test 5: Verify deleted shots list includes project_id print("\n5. Testing deleted shots list includes project_id...") # Delete the shot again to test the list db.commit() deletion_result = deletion_service.soft_delete_shot_cascade(shot.id, db, admin_user) if deletion_result.success: deleted_shots = recovery_service.get_deleted_shots(None, db) # Find our test shot in the list test_shot_info = None for deleted_shot in deleted_shots: if deleted_shot.id == shot.id: test_shot_info = deleted_shot break if test_shot_info: print(f" ✓ Found shot in deleted shots list") print(f" ✓ Project ID: {test_shot_info.project_id}") print(f" ✓ Project Name: {test_shot_info.project_name}") assert test_shot_info.project_id == original_project_id, f"Project ID mismatch in list: {test_shot_info.project_id} != {original_project_id}" assert test_shot_info.project_name == original_project_name, f"Project name mismatch in list: {test_shot_info.project_name} != {original_project_name}" print(" ✓ Project information correctly included in deleted shots list") else: print(" ✗ Shot not found in deleted shots list") return # Clean up - recover the shot again db.commit() recovery_service.recover_shot(shot.id, db, admin_user) print("\n✅ All project_id preservation tests passed!") except Exception as e: print(f"\n❌ Error during testing: {str(e)}") import traceback traceback.print_exc() finally: db.close() if __name__ == "__main__": test_project_id_preservation()