Files
balanceboard/test_db_connection.py
chelsea e821a26b48 Initial commit: BalanceBoard - Reddit-style content aggregator
- Flask-based web application with PostgreSQL
- User authentication and session management
- Content moderation and filtering
- Docker deployment with docker-compose
- Admin interface for content management

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-10-11 16:11:13 -05:00

86 lines
2.3 KiB
Python

#!/usr/bin/env python3
"""
Test script to verify PostgreSQL connection for the app.
"""
import os
import psycopg2
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
def test_connection():
"""Test database connection using app's configuration"""
# Get database configuration from environment
db_user = os.getenv('POSTGRES_USER', 'balanceboard')
db_password = os.getenv('POSTGRES_PASSWORD', 'balanceboard123')
db_host = os.getenv('POSTGRES_HOST', 'localhost')
db_port = os.getenv('POSTGRES_PORT', '5432')
db_name = os.getenv('POSTGRES_DB', 'balanceboard')
print(f"Testing connection to PostgreSQL:")
print(f" Host: {db_host}")
print(f" Port: {db_port}")
print(f" Database: {db_name}")
print(f" User: {db_user}")
try:
# Test connection
conn = psycopg2.connect(
host=db_host,
port=db_port,
database=db_name,
user=db_user,
password=db_password
)
# Create a cursor
cur = conn.cursor()
# Test query
cur.execute("SELECT version();")
version = cur.fetchone()
print(f"\n✓ Connection successful!")
print(f" PostgreSQL version: {version[0]}")
# Test if we can create a simple table
cur.execute("""
CREATE TABLE IF NOT EXISTS test_table (
id SERIAL PRIMARY KEY,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
# Insert test data
cur.execute("INSERT INTO test_table DEFAULT VALUES;")
conn.commit()
# Query test data
cur.execute("SELECT COUNT(*) FROM test_table;")
count = cur.fetchone()[0]
print(f"✓ Database operations successful!")
print(f" Test table has {count} rows")
# Clean up
cur.execute("DROP TABLE IF EXISTS test_table;")
conn.commit()
# Close connections
cur.close()
conn.close()
print("✓ Connection test completed successfully!")
return True
except Exception as e:
print(f"\n✗ Connection failed:")
print(f" Error: {e}")
return False
if __name__ == "__main__":
test_connection()