168 lines
7.4 KiB
Python
168 lines
7.4 KiB
Python
"""
|
|
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() |