minimal_test.py

import asyncio
import asyncpg
import ssl
import platform
import sys
import logging
from typing import Optional

# Configure logging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

async def get_ssl_context() -> ssl.SSLContext:
    """Create an SSL context for database connection."""
    ssl_context = ssl.create_default_context()
    ssl_context.check_hostname = False
    ssl_context.verify_mode = ssl.CERT_NONE
    return ssl_context

async def test_connection() -> None:
    """Test the database connection with detailed logging."""
    logger.info("Starting database connection test...")
    
    if platform.system() == 'Windows':
        logger.info("Configuring Windows-specific event loop policy...")
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    
    try:
        ssl_context = await get_ssl_context()
        logger.info("Attempting to establish database connection...")
        
        conn: Optional[asyncpg.Connection] = None
        try:
            conn = await asyncpg.connect(
                user='postgres',
                password='Lovyelias5584.',
                database='postgres',
                host='juycnkjuzylnbruwaqmp.supabase.co',
                port=5432,
                ssl=ssl_context,
                command_timeout=30.0,
                server_settings={'application_name': 'pos_backend_test'}
            )
            
            logger.info("Connection established successfully")
            
            # Test the connection with a simple query
            result = await conn.fetchval('SELECT version()')
            logger.info(f"Database version: {result}")
            
            # Test table access
            tables = await conn.fetch("""
                SELECT table_name 
                FROM information_schema.tables 
                WHERE table_schema = 'public'
            """)
            logger.info("Available tables:")
            for table in tables:
                logger.info(f"- {table['table_name']}")
                
        except Exception as e:
            logger.error(f"Database connection failed: {str(e)}", exc_info=True)
            raise
        finally:
            if conn:
                logger.info("Closing database connection...")
                await conn.close()
                logger.info("Database connection closed")
    
    except Exception as e:
        logger.error(f"Test failed: {str(e)}", exc_info=True)
        raise
    else:
        logger.info("Database connection test completed successfully")

if __name__ == "__main__":
    try:
        asyncio.run(test_connection())
    except KeyboardInterrupt:
        logger.info("Test interrupted by user")
    except Exception as e:
        logger.error("Fatal error occurred", exc_info=True)
        sys.exit(1)