Spaces:
Paused
Paused
| import socket | |
| import ssl | |
| import sys | |
| import os | |
| import asyncio | |
| import asyncpg | |
| import psycopg2 | |
| from datetime import datetime | |
| # Configuration | |
| HOST = "juycnkjuzylnbruwaqmp.supabase.co" | |
| PORT = 5432 | |
| USER = "postgres" | |
| PASSWORD = "Lovyelias5584." | |
| DATABASE = "postgres" | |
| def log_to_file(message): | |
| timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') | |
| with open('comprehensive_test.log', 'a') as f: | |
| f.write(f"[{timestamp}] {message}\n") | |
| async def test_asyncpg(): | |
| log_to_file("\n=== Testing asyncpg connection ===") | |
| try: | |
| ssl_context = ssl.create_default_context() | |
| ssl_context.check_hostname = False | |
| ssl_context.verify_mode = ssl.CERT_NONE | |
| log_to_file("Attempting asyncpg connection...") | |
| conn = await asyncpg.connect( | |
| user=USER, | |
| password=PASSWORD, | |
| database=DATABASE, | |
| host=HOST, | |
| port=PORT, | |
| ssl=ssl_context | |
| ) | |
| log_to_file("asyncpg connection successful!") | |
| await conn.close() | |
| return True | |
| except Exception as e: | |
| log_to_file(f"asyncpg connection failed: {str(e)}") | |
| return False | |
| def test_psycopg2(): | |
| log_to_file("\n=== Testing psycopg2 connection ===") | |
| try: | |
| log_to_file("Attempting psycopg2 connection...") | |
| conn = psycopg2.connect( | |
| dbname=DATABASE, | |
| user=USER, | |
| password=PASSWORD, | |
| host=HOST, | |
| port=PORT, | |
| sslmode='require' | |
| ) | |
| log_to_file("psycopg2 connection successful!") | |
| conn.close() | |
| return True | |
| except Exception as e: | |
| log_to_file(f"psycopg2 connection failed: {str(e)}") | |
| return False | |
| def test_socket(): | |
| log_to_file("\n=== Testing raw socket connection ===") | |
| try: | |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
| sock.settimeout(10) | |
| # Try DNS resolution | |
| try: | |
| ip = socket.gethostbyname(HOST) | |
| log_to_file(f"DNS resolution successful: {HOST} -> {ip}") | |
| except socket.gaierror as e: | |
| log_to_file(f"DNS resolution failed: {str(e)}") | |
| return False | |
| # Try SSL connection | |
| context = ssl.create_default_context() | |
| context.check_hostname = False | |
| context.verify_mode = ssl.CERT_NONE | |
| wrapped_socket = context.wrap_socket(sock) | |
| log_to_file(f"Attempting connection to {HOST}:{PORT}...") | |
| result = wrapped_socket.connect_ex((HOST, PORT)) | |
| if result == 0: | |
| log_to_file(f"Socket connection successful!") | |
| return True | |
| else: | |
| log_to_file(f"Socket connection failed with error code: {result}") | |
| return False | |
| except Exception as e: | |
| log_to_file(f"Socket connection failed with error: {str(e)}") | |
| return False | |
| finally: | |
| try: | |
| sock.close() | |
| except: | |
| pass | |
| async def main(): | |
| # Clear previous log | |
| with open('comprehensive_test.log', 'w') as f: | |
| f.write(f"=== Comprehensive Database Connection Test ===\n") | |
| f.write(f"Started at: {datetime.now()}\n") | |
| f.write(f"Python version: {sys.version}\n") | |
| f.write(f"Platform: {sys.platform}\n\n") | |
| # Run tests | |
| socket_result = test_socket() | |
| psycopg2_result = test_psycopg2() | |
| asyncpg_result = await test_asyncpg() | |
| # Log summary | |
| log_to_file("\n=== Test Summary ===") | |
| log_to_file(f"Socket test: {'PASSED' if socket_result else 'FAILED'}") | |
| log_to_file(f"psycopg2 test: {'PASSED' if psycopg2_result else 'FAILED'}") | |
| log_to_file(f"asyncpg test: {'PASSED' if asyncpg_result else 'FAILED'}") | |
| # Exit with status code | |
| return 0 if (socket_result and psycopg2_result and asyncpg_result) else 1 | |
| if __name__ == "__main__": | |
| sys.exit(asyncio.run(main())) |