101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify PostgreSQL migration is working properly
|
|
"""
|
|
|
|
import psycopg2
|
|
import sys
|
|
|
|
def test_database_migration():
|
|
"""Test if the database migration was successful"""
|
|
|
|
try:
|
|
# Connect to PostgreSQL
|
|
conn = psycopg2.connect(
|
|
host='localhost',
|
|
port=5432,
|
|
user='pipeline_admin',
|
|
password='secure_pipeline_2024',
|
|
database='dev_pipeline'
|
|
)
|
|
cursor = conn.cursor()
|
|
|
|
print("🧪 Testing PostgreSQL Migration")
|
|
print("=" * 40)
|
|
|
|
# Test tables exist
|
|
tables_to_check = [
|
|
'price_tiers',
|
|
'frontend_technologies',
|
|
'backend_technologies',
|
|
'database_technologies',
|
|
'cloud_technologies',
|
|
'testing_technologies',
|
|
'mobile_technologies',
|
|
'devops_technologies',
|
|
'ai_ml_technologies',
|
|
'tools',
|
|
'price_based_stacks',
|
|
'stack_recommendations'
|
|
]
|
|
|
|
print("📋 Checking table existence:")
|
|
for table in tables_to_check:
|
|
cursor.execute(f"""
|
|
SELECT EXISTS (
|
|
SELECT FROM information_schema.tables
|
|
WHERE table_schema = 'public'
|
|
AND table_name = '{table}'
|
|
);
|
|
""")
|
|
exists = cursor.fetchone()[0]
|
|
status = "✅" if exists else "❌"
|
|
print(f" {status} {table}")
|
|
|
|
print("\n📊 Checking data counts:")
|
|
for table in tables_to_check:
|
|
try:
|
|
cursor.execute(f'SELECT COUNT(*) FROM {table};')
|
|
count = cursor.fetchone()[0]
|
|
print(f" {table}: {count} records")
|
|
except Exception as e:
|
|
print(f" {table}: Error - {e}")
|
|
|
|
# Test specific data
|
|
print("\n🔍 Testing specific data:")
|
|
|
|
# Test price tiers
|
|
cursor.execute("SELECT tier_name, min_price_usd, max_price_usd FROM price_tiers ORDER BY min_price_usd;")
|
|
price_tiers = cursor.fetchall()
|
|
print(f" Price tiers: {len(price_tiers)}")
|
|
for tier in price_tiers:
|
|
print(f" - {tier[0]}: ${tier[1]} - ${tier[2]}")
|
|
|
|
# Test stack recommendations
|
|
cursor.execute("SELECT business_domain, COUNT(*) FROM stack_recommendations GROUP BY business_domain;")
|
|
domains = cursor.fetchall()
|
|
print(f" Domain recommendations: {len(domains)}")
|
|
for domain in domains:
|
|
print(f" - {domain[0]}: {domain[1]} recommendations")
|
|
|
|
# Test tools
|
|
cursor.execute("SELECT category, COUNT(*) FROM tools GROUP BY category;")
|
|
tool_categories = cursor.fetchall()
|
|
print(f" Tool categories: {len(tool_categories)}")
|
|
for category in tool_categories:
|
|
print(f" - {category[0]}: {category[1]} tools")
|
|
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
print("\n✅ Database migration test completed successfully!")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ Database migration test failed: {e}")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
success = test_database_migration()
|
|
sys.exit(0 if success else 1)
|