96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify PostgreSQL database creation and setup
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
|
|
# Add the app directory to the Python path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'app'))
|
|
|
|
def test_postgres_setup():
|
|
"""Test PostgreSQL database creation and setup"""
|
|
print("=== Testing PostgreSQL Database Setup ===")
|
|
|
|
# Check if PostgreSQL credentials are available
|
|
required_vars = ['POSTGRES_HOST', 'POSTGRES_DB', 'POSTGRES_USER', 'POSTGRES_PASSWORD']
|
|
missing_vars = [var for var in required_vars if not os.getenv(var)]
|
|
|
|
if missing_vars:
|
|
print(f"⚠ PostgreSQL test skipped - missing environment variables: {', '.join(missing_vars)}")
|
|
print("Set these variables to test PostgreSQL:")
|
|
print(" POSTGRES_HOST=your-host")
|
|
print(" POSTGRES_DB=your-database")
|
|
print(" POSTGRES_USER=your-username")
|
|
print(" POSTGRES_PASSWORD=your-password")
|
|
return False
|
|
|
|
try:
|
|
from app.database import db_manager, init_db
|
|
|
|
print(f"Database type: {db_manager.db_type}")
|
|
print(f"PostgreSQL host: {os.getenv('POSTGRES_HOST')}")
|
|
print(f"PostgreSQL database: {os.getenv('POSTGRES_DB')}")
|
|
print(f"PostgreSQL user: {os.getenv('POSTGRES_USER')}")
|
|
|
|
# Initialize database (this will create the database if it doesn't exist)
|
|
init_db()
|
|
print("✓ Database initialized successfully")
|
|
|
|
# Test a simple cache operation
|
|
from app.database import save_to_cache, get_cached_data
|
|
|
|
test_url = "https://example.com"
|
|
test_route = "test"
|
|
test_data = {"title": "Test Page", "content": "Test content"}
|
|
|
|
# Save to cache
|
|
save_to_cache(test_url, test_route, test_data)
|
|
print("✓ Data saved to cache")
|
|
|
|
# Retrieve from cache
|
|
cached_data = get_cached_data(test_url, test_route)
|
|
if cached_data and cached_data == test_data:
|
|
print("✓ Data retrieved from cache successfully")
|
|
else:
|
|
print("✗ Failed to retrieve data from cache")
|
|
return False
|
|
|
|
# Test cache stats
|
|
from app.services.cache import get_cache_stats
|
|
stats = get_cache_stats()
|
|
if stats['status'] == 'success':
|
|
print("✓ Cache stats retrieved successfully")
|
|
print(f" Database type: {stats['stats']['database_type']}")
|
|
print(f" Total entries: {stats['stats']['total_entries']}")
|
|
else:
|
|
print("✗ Failed to get cache stats")
|
|
return False
|
|
|
|
print("✓ PostgreSQL database setup test passed!")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"✗ PostgreSQL setup test failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
def main():
|
|
"""Run the PostgreSQL setup test"""
|
|
print("PostgreSQL Database Setup Test")
|
|
print("=" * 50)
|
|
|
|
success = test_postgres_setup()
|
|
|
|
print("\n" + "=" * 50)
|
|
if success:
|
|
print("🎉 PostgreSQL setup test passed! Database creation and connection working correctly.")
|
|
return 0
|
|
else:
|
|
print("❌ PostgreSQL setup test failed. Please check the configuration and PostgreSQL server.")
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |