LinkDesk/backend/test_shot_project_id_migration.py

#!/usr/bin/env python3
"""
Test script to verify the shot project_id migration.
Tests all requirements: 2.1, 2.2, 2.3, 2.4, 2.5
"""

import sqlite3
import sys
from pathlib import Path


def test_migration():
    """Test that the migration meets all requirements."""
    
    db_path = "vfx_project_management.db"
    if not Path(db_path).exists():
        print("❌ Database not found")
        return False
    
    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()
        
        print("Testing Shot Project ID Migration Requirements")
        print("=" * 50)
        
        # Requirement 2.1: Non-nullable project_id column
        print("Testing Requirement 2.1: Non-nullable project_id column...")
        cursor.execute("PRAGMA table_info(shots)")
        columns = {col[1]: col for col in cursor.fetchall()}
        
        if 'project_id' not in columns:
            print("❌ project_id column not found")
            return False
        
        project_id_col = columns['project_id']
        if project_id_col[3] != 1:  # NOT NULL constraint
            print("❌ project_id column is nullable")
            return False
        
        print("✅ project_id column exists and is NOT NULL")
        
        # Requirement 2.2: Foreign key constraint to projects table
        print("\nTesting Requirement 2.2: Foreign key constraint...")
        cursor.execute("PRAGMA foreign_key_list(shots)")
        foreign_keys = cursor.fetchall()
        
        project_fk_found = False
        for fk in foreign_keys:
            if fk[2] == 'projects' and fk[3] == 'project_id':
                project_fk_found = True
                break
        
        if not project_fk_found:
            print("❌ Foreign key constraint to projects table not found")
            return False
        
        print("✅ Foreign key constraint to projects table exists")
        
        # Requirement 2.3: Index on project_id column
        print("\nTesting Requirement 2.3: Performance indexes...")
        cursor.execute("PRAGMA index_list(shots)")
        indexes = [index[1] for index in cursor.fetchall()]
        
        required_indexes = ['idx_shots_project_id']
        for index_name in required_indexes:
            if index_name not in indexes:
                print(f"❌ Index {index_name} not found")
                return False
        
        print("✅ Performance indexes exist")
        
        # Requirement 2.4: Data preservation - all existing shots have project_id
        print("\nTesting Requirement 2.4: Data preservation...")
        cursor.execute("SELECT COUNT(*) FROM shots WHERE project_id IS NULL")
        null_count = cursor.fetchone()[0]
        
        if null_count > 0:
            print(f"❌ {null_count} shots have NULL project_id")
            return False
        
        cursor.execute("SELECT COUNT(*) FROM shots")
        total_shots = cursor.fetchone()[0]
        print(f"✅ All {total_shots} shots have project_id populated")
        
        # Requirement 2.5: Data integrity - project_id matches episode's project
        print("\nTesting Requirement 2.5: Data integrity...")
        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]
        if inconsistent_count > 0:
            print(f"❌ {inconsistent_count} shots have inconsistent project_id")
            return False
        
        print("✅ All shots have consistent project_id with their episodes")
        
        # Additional test: Unique constraint for project-scoped names
        print("\nTesting project-scoped name uniqueness...")
        
        # Check that unique index exists
        if 'idx_shots_project_name_unique' not in indexes:
            print("❌ Unique constraint index not found")
            return False
        
        # Test the constraint by trying to insert a duplicate
        try:
            # Get a sample shot to duplicate
            cursor.execute("SELECT project_id, episode_id, name FROM shots LIMIT 1")
            sample = cursor.fetchone()
            if sample:
                cursor.execute("""
                    INSERT INTO shots (project_id, episode_id, name, frame_start, frame_end)
                    VALUES (?, ?, ?, 1001, 1100)
                """, sample)
                print("❌ Duplicate shot name was allowed in same project")
                return False
        except sqlite3.IntegrityError:
            print("✅ Unique constraint prevents duplicate names in same project")
        
        # Performance test: Check query performance with indexes
        print("\nTesting query performance with indexes...")
        
        # Test project filtering query
        cursor.execute("EXPLAIN QUERY PLAN SELECT * FROM shots WHERE project_id = 1")
        query_plan = cursor.fetchall()
        
        # Check if index is being used (should contain "idx_shots_project_id")
        uses_index = any("idx_shots_project_id" in str(step) for step in query_plan)
        if uses_index:
            print("✅ Project filtering queries use index")
        else:
            print("⚠️  Project filtering queries may not be optimized")
        
        print("\n" + "=" * 50)
        print("✅ ALL REQUIREMENTS PASSED")
        print("✅ Migration completed successfully")
        
        # Summary
        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
        """)
        
        print("\nShots per project:")
        for project_name, shot_count in cursor.fetchall():
            print(f"  {project_name}: {shot_count} shots")
        
        return True
        
    except sqlite3.Error as e:
        print(f"❌ Database error: {e}")
        return False
    except Exception as e:
        print(f"❌ Unexpected error: {e}")
        return False
    finally:
        if conn:
            conn.close()


if __name__ == "__main__":
    success = test_migration()
    sys.exit(0 if success else 1)