152 lines
4.9 KiB
Python
152 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify database configuration for both SQLite and PostgreSQL
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import json
|
|
|
|
# Add the app directory to the Python path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'app'))
|
|
|
|
def test_sqlite():
|
|
"""Test SQLite database functionality"""
|
|
print("=== Testing SQLite Database ===")
|
|
|
|
# Clear any existing PostgreSQL environment variables
|
|
for var in ['POSTGRES_HOST', 'POSTGRES_DB', 'POSTGRES_USER', 'POSTGRES_PASSWORD']:
|
|
if var in os.environ:
|
|
del os.environ[var]
|
|
|
|
try:
|
|
from app.database import db_manager, init_db, get_cached_data, save_to_cache, cleanup_old_cache_entries
|
|
|
|
print(f"Database type: {db_manager.db_type}")
|
|
|
|
# Initialize database
|
|
init_db()
|
|
print("✓ Database initialized successfully")
|
|
|
|
# Test cache operations
|
|
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("✓ SQLite database test passed!")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"✗ SQLite test failed: {e}")
|
|
return False
|
|
|
|
def test_postgres():
|
|
"""Test PostgreSQL database functionality"""
|
|
print("\n=== Testing PostgreSQL Database ===")
|
|
|
|
# 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 True # Not a failure, just skipped
|
|
|
|
try:
|
|
from app.database import db_manager, init_db, get_cached_data, save_to_cache
|
|
|
|
print(f"Database type: {db_manager.db_type}")
|
|
|
|
# Initialize database
|
|
init_db()
|
|
print("✓ Database initialized successfully")
|
|
|
|
# Test cache operations
|
|
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 test passed!")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"✗ PostgreSQL test failed: {e}")
|
|
return False
|
|
|
|
def main():
|
|
"""Run all database tests"""
|
|
print("Database Configuration Test")
|
|
print("=" * 50)
|
|
|
|
# Test SQLite
|
|
sqlite_success = test_sqlite()
|
|
|
|
# Test PostgreSQL
|
|
postgres_success = test_postgres()
|
|
|
|
print("\n" + "=" * 50)
|
|
print("Test Results:")
|
|
print(f"SQLite: {'✓ PASSED' if sqlite_success else '✗ FAILED'}")
|
|
print(f"PostgreSQL: {'✓ PASSED' if postgres_success else '✗ FAILED'}")
|
|
|
|
if sqlite_success and postgres_success:
|
|
print("\n🎉 All tests passed! Database configuration is working correctly.")
|
|
return 0
|
|
else:
|
|
print("\n❌ Some tests failed. Please check the configuration.")
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |