109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import os
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
# Set environment variables for PostgreSQL
|
|
os.environ['DATABASE_TYPE'] = 'postgresql'
|
|
# Use existing environment variables if available, otherwise set defaults
|
|
if not os.environ.get('SBA_DATABASE'):
|
|
os.environ['SBA_DATABASE'] = 'sba_master'
|
|
if not os.environ.get('SBA_DB_USER'):
|
|
os.environ['SBA_DB_USER'] = 'sba_admin'
|
|
if not os.environ.get('SBA_DB_USER_PASSWORD'):
|
|
os.environ['SBA_DB_USER_PASSWORD'] = 'sba_dev_password_2024'
|
|
if not os.environ.get('POSTGRES_HOST'):
|
|
os.environ['POSTGRES_HOST'] = 'localhost'
|
|
if not os.environ.get('POSTGRES_PORT'):
|
|
os.environ['POSTGRES_PORT'] = '5432'
|
|
|
|
# Import after setting environment variables
|
|
from app.db_engine import db, Current, Team, Player, SbaPlayer, Manager, Division
|
|
|
|
logger = logging.getLogger(f'{__name__}.test_postgres')
|
|
|
|
def test_connection():
|
|
"""Test PostgreSQL connection"""
|
|
try:
|
|
logger.info("Testing PostgreSQL connection...")
|
|
db.connect()
|
|
logger.info("✓ Connected to PostgreSQL successfully")
|
|
|
|
# Test basic query
|
|
result = db.execute_sql("SELECT version();").fetchone()
|
|
logger.info(f"✓ PostgreSQL version: {result[0]}")
|
|
|
|
db.close()
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"✗ Connection failed: {e}")
|
|
return False
|
|
|
|
def test_schema_creation():
|
|
"""Test creating tables in PostgreSQL"""
|
|
try:
|
|
logger.info("Testing schema creation...")
|
|
db.connect()
|
|
|
|
# Create tables in dependency order
|
|
# First: tables with no dependencies
|
|
base_tables = [Current, Manager, Division, SbaPlayer]
|
|
db.create_tables(base_tables, safe=True)
|
|
|
|
# Second: tables that depend on base tables
|
|
dependent_tables = [Team, Player]
|
|
db.create_tables(dependent_tables, safe=True)
|
|
|
|
logger.info("✓ Test tables created successfully")
|
|
|
|
# Test table existence
|
|
all_test_tables = base_tables + dependent_tables
|
|
for table in all_test_tables:
|
|
table_name = table._meta.table_name
|
|
result = db.execute_sql("""
|
|
SELECT EXISTS (
|
|
SELECT FROM information_schema.tables
|
|
WHERE table_schema = 'public'
|
|
AND table_name = %s
|
|
);
|
|
""", (table_name,)).fetchone()
|
|
|
|
if result[0]:
|
|
logger.info(f"✓ Table '{table_name}' exists")
|
|
else:
|
|
logger.error(f"✗ Table '{table_name}' missing")
|
|
|
|
db.close()
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"✗ Schema creation failed: {e}")
|
|
return False
|
|
|
|
def main():
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
|
|
logger.info("Starting PostgreSQL migration test...")
|
|
|
|
# Test connection
|
|
if not test_connection():
|
|
logger.error("Connection test failed - stopping")
|
|
return False
|
|
|
|
# Test schema creation
|
|
if not test_schema_creation():
|
|
logger.error("Schema creation test failed - stopping")
|
|
return False
|
|
|
|
logger.info("✓ All PostgreSQL tests passed!")
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
exit(0 if success else 1) |