test_migration_fix.py

#!/usr/bin/env python3
"""
Test script to verify the fixed CouchDB to MySQL migration
This script tests the complete migration pipeline with generalized logic
"""

from couchdb_converter.couchdb2mysql import CouchDBToMySQLConverter
import json

def test_migration():
    """Test the complete migration process"""
    
    print("="*70)
    print("🧪 TESTING FIXED COUCHDB TO MYSQL MIGRATION")
    print("="*70)
    
    # MySQL configuration - UPDATE THESE WITH YOUR CREDENTIALS
    mysql_config = {
        'host': 'localhost',
        'user': 'root',
        'password': 'root',  # Change to your MySQL password
        'database': 'couchdb_migration_test'
    }
    
    # CouchDB configuration
    couchdb_config = {
        'server_url': 'http://admin:root@localhost:5984',
        'db_name': 'nosql2sql_test'
    }
    
    print(f"\n📋 Configuration:")
    print(f"  CouchDB: {couchdb_config['server_url']}/{couchdb_config['db_name']}")
    print(f"  MySQL: {mysql_config['user']}@{mysql_config['host']}/{mysql_config['database']}")
    
    # Create converter
    converter = CouchDBToMySQLConverter(mysql_config, couchdb_config)
    
    try:
        print("\n" + "-"*70)
        print("STEP 1: Connecting to databases")
        print("-"*70)
        
        if not converter.connect_mysql():
            print("❌ Failed to connect to MySQL")
            return False
        print("✓ MySQL connected")
        
        if not converter.connect_couchdb():
            print("❌ Failed to connect to CouchDB")
            return False
        print("✓ CouchDB connected")
        
        print("\n" + "-"*70)
        print("STEP 2: Analyzing CouchDB schema")
        print("-"*70)
        
        if not converter.analyze_couchdb_schema():
            print("❌ Schema analysis failed")
            return False
        
        print("✓ Schema analysis complete")
        print(f"  Document types found: {list(converter.schema_recommendations['statistics']['document_types'].keys())}")
        print(f"  Tables to create: {len(converter.schema_recommendations['tables'])}")
        
        print("\n" + "-"*70)
        print("STEP 3: Creating MySQL schema")
        print("-"*70)
        
        if not converter.create_mysql_schema():
            print("❌ Schema creation failed")
            return False
        
        print("✓ MySQL schema created")
        print(f"  Tables: {', '.join(converter.schema_builder.created_tables)}")
        
        print("\n" + "-"*70)
        print("STEP 4: Migrating data")
        print("-"*70)
        
        if not converter.migrate_data():
            print("❌ Data migration failed")
            return False
        
        print("\n" + "-"*70)
        print("STEP 5: Verifying results")
        print("-"*70)
        
        converter.print_migration_summary()
        
        # Success check
        summary = converter.get_migration_summary()
        if summary['migrated_successfully'] > 0:
            print("\n✅ SUCCESS! Migration completed with data in tables!")
            
            # Show some sample data from first table
            if converter.schema_builder.created_tables:
                first_table = converter.schema_builder.created_tables[0]
                print(f"\n🔍 Sample data from '{first_table}' table:")
                try:
                    converter.mysql_cursor.execute(f"SELECT * FROM `{first_table}` LIMIT 3")
                    rows = converter.mysql_cursor.fetchall()
                    
                    if rows:
                        # Get column names
                        converter.mysql_cursor.execute(f"DESCRIBE `{first_table}`")
                        columns = [col[0] for col in converter.mysql_cursor.fetchall()]
                        
                        for row in rows:
                            print(f"\n  Row:")
                            for col, val in zip(columns, row):
                                val_str = str(val)[:50] + "..." if len(str(val)) > 50 else str(val)
                                print(f"    {col}: {val_str}")
                    else:
                        print("  (Table is empty)")
                        
                except Exception as e:
                    print(f"  Error fetching sample data: {e}")
            
            return True
        else:
            print("\n⚠️  Migration ran but no data was migrated!")
            return False
        
    except Exception as e:
        print(f"\n❌ FATAL ERROR: {e}")
        import traceback
        traceback.print_exc()
        return False
        
    finally:
        converter.close_connections()
        print("\n" + "="*70)
        print("🏁 Test complete")
        print("="*70)

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