LinkDesk/backend/test_migration_integrity.py

106 lines
4.2 KiB
Python

#!/usr/bin/env python3
"""
Test script to verify migration integrity and shot operations.
"""
import sqlite3
import sys
def test_migration_integrity():
"""Test the integrity of the migration and shot operations."""
print("Testing migration integrity and shot operations...")
conn = sqlite3.connect('vfx_project_management.db')
cursor = conn.cursor()
try:
# Test 1: Verify all shots have project_id
cursor.execute('SELECT COUNT(*) FROM shots WHERE project_id IS NULL')
null_count = cursor.fetchone()[0]
print(f'✅ Shots with NULL project_id: {null_count} (should be 0)')
# Test 2: Verify project-episode consistency
cursor.execute('''
SELECT COUNT(*) FROM shots s
JOIN episodes e ON s.episode_id = e.id
WHERE s.project_id != e.project_id
''')
inconsistent_count = cursor.fetchone()[0]
print(f'✅ Shots with inconsistent project_id: {inconsistent_count} (should be 0)')
# Test 3: Check foreign key constraints work
cursor.execute('PRAGMA foreign_keys = ON')
try:
cursor.execute('INSERT INTO shots (project_id, episode_id, name) VALUES (99999, 1, "test_shot")')
cursor.execute('DELETE FROM shots WHERE name = "test_shot"')
print('❌ ERROR: Foreign key constraint not working!')
return False
except sqlite3.IntegrityError as e:
print(f'✅ Foreign key constraint working: {str(e)[:50]}...')
# Test 4: Check unique constraint works for project-scoped names
cursor.execute('SELECT project_id, name FROM shots WHERE deleted_at IS NULL LIMIT 1')
result = cursor.fetchone()
if result:
project_id, name = result
try:
# Try to insert a shot with same name in same project
cursor.execute('INSERT INTO shots (project_id, episode_id, name) VALUES (?, 1, ?)', (project_id, name))
cursor.execute('DELETE FROM shots WHERE name = ?', (name,))
print('❌ ERROR: Unique constraint not working!')
return False
except sqlite3.IntegrityError as e:
print(f'✅ Unique constraint working: {str(e)[:50]}...')
# Test 5: Verify indexes exist
cursor.execute("PRAGMA index_list(shots)")
indexes = [index[1] for index in cursor.fetchall()]
required_indexes = ['idx_shots_project_id', 'idx_shots_name', 'idx_shots_project_name_unique']
for index_name in required_indexes:
if index_name in indexes:
print(f'✅ Index {index_name} exists')
else:
print(f'❌ Index {index_name} missing')
return False
# Test 6: Test project filtering
cursor.execute('SELECT DISTINCT project_id FROM shots LIMIT 2')
project_ids = [row[0] for row in cursor.fetchall()]
if len(project_ids) >= 1:
test_project_id = project_ids[0]
cursor.execute('SELECT COUNT(*) FROM shots WHERE project_id = ?', (test_project_id,))
project_shot_count = cursor.fetchone()[0]
print(f'✅ Project filtering works: Project {test_project_id} has {project_shot_count} shots')
# Test 7: Show summary statistics
cursor.execute('''
SELECT p.name, COUNT(s.id) as shot_count
FROM projects p
LEFT JOIN shots s ON p.id = s.project_id AND s.deleted_at IS NULL
GROUP BY p.id, p.name
ORDER BY shot_count DESC
LIMIT 5
''')
print("\n📊 Shot distribution by project (top 5):")
for project_name, shot_count in cursor.fetchall():
print(f" {project_name}: {shot_count} shots")
return True
except Exception as e:
print(f"❌ Test failed with error: {e}")
return False
finally:
conn.close()
if __name__ == "__main__":
success = test_migration_integrity()
if success:
print("\n✅ All migration integrity tests passed!")
sys.exit(0)
else:
print("\n❌ Some tests failed!")
sys.exit(1)