#!/usr/bin/env python """ Complete Supabase Integration Verification Script Tests all components of the Supabase setup """ import asyncio import sys from pathlib import Path # Add src to path sys.path.insert(0, str(Path(__file__).parent)) def print_header(title): """Print formatted header.""" print(f"\n{'='*70}") print(f" {title}") print(f"{'='*70}") def print_section(title): """Print formatted section.""" print(f"\n{title}") print("-" * 70) def check_mark(condition, message): """Print check or cross mark with message.""" symbol = "✓" if condition else "✗" print(f" {symbol} {message}") async def verify_config(): """Verify configuration system.""" print_section("1. Configuration System") try: from src.core.config.config import get_settings, Settings check_mark(True, "Config module imports successfully") settings = get_settings() check_mark(True, f"Settings instance created: {Settings.__name__}") # Check all required fields required_fields = [ ('APP_NAME', settings.APP_NAME), ('ENVIRONMENT', settings.ENVIRONMENT), ('DATABASE_URL', len(settings.DATABASE_URL) > 0), ('SUPABASE_URL', settings.SUPABASE_URL), ('SUPABASE_API_KEY', len(settings.SUPABASE_API_KEY) > 0), ('SECRET_KEY', len(settings.SECRET_KEY) >= 32), ('REDIS_URL', settings.REDIS_URL), ] for field_name, value in required_fields: if isinstance(value, str) and len(value) > 50: display = value[:47] + "..." else: display = str(value)[:50] check_mark( value if isinstance(value, bool) else bool(value), f"{field_name}: {display}" ) return True except Exception as e: check_mark(False, f"Configuration error: {e}") return False def verify_database_models(): """Verify database models.""" print_section("2. Database Models") try: from src.db.models import User, UserSession, Conversation, Message models = [ ('User', User), ('UserSession', UserSession), ('Conversation', Conversation), ('Message', Message), ] for model_name, model_class in models: check_mark(True, f"{model_name} model defined") return True except Exception as e: check_mark(False, f"Database models error: {e}") return False def verify_database_config(): """Verify database configuration.""" print_section("3. Database Configuration") try: from src.db.database import engine, AsyncSessionLocal, init_db, dispose_engine, get_session check_mark(True, "AsyncEngine created") check_mark(True, "AsyncSessionLocal factory created") check_mark(True, "get_session dependency available") check_mark(True, "init_db function available") check_mark(True, "dispose_engine function available") return True except Exception as e: check_mark(False, f"Database configuration error: {e}") return False def verify_supabase_client(): """Verify Supabase client.""" print_section("4. Supabase Client") try: from src.db.supabase_client import get_supabase_client, get_supabase check_mark(True, "Supabase client module imports successfully") try: client = get_supabase_client() check_mark(True, f"Supabase client created: {type(client).__name__}") # Check client modules has_auth = hasattr(client, 'auth') has_table = hasattr(client, 'table') has_storage = hasattr(client, 'storage') check_mark(has_auth, "Client has auth module") check_mark(has_table, "Client has table module") check_mark(has_storage, "Client has storage module") except ValueError as e: check_mark(False, f"Supabase credentials issue: {e}") except Exception as e: check_mark(False, f"Supabase client error: {e}") return True except Exception as e: check_mark(False, f"Supabase module error: {e}") return False async def verify_database_connection(): """Verify database connection capability.""" print_section("5. Database Connection") try: from src.db.database import get_session try: async for session in get_session(): check_mark(True, f"Database session created: {type(session).__name__}") check_mark(True, "Connection pool functional") break except Exception as e: check_mark(False, f"Database connection error: {e}") return True except Exception as e: check_mark(False, f"Get session error: {e}") return False def verify_pydantic_schemas(): """Verify Pydantic schemas.""" print_section("6. Pydantic Schemas") try: from src.models import ( UserCreate, UserLogin, UserResponse, UserUpdate, Token, TokenData, Message, Conversation ) schemas = [ ('UserCreate', UserCreate), ('UserLogin', UserLogin), ('UserResponse', UserResponse), ('UserUpdate', UserUpdate), ('Token', Token), ('TokenData', TokenData), ('Message', Message), ('Conversation', Conversation), ] for schema_name, schema_class in schemas: check_mark(True, f"{schema_name} schema defined") return True except Exception as e: check_mark(False, f"Schemas error: {e}") return False def verify_api_routes(): """Verify API routes.""" print_section("7. API Routes") try: from src.api.routes import users, auth, login, chat routes = [ ('users', users.router), ('auth', auth.router), ('profile/login', login.router), ('chat', chat.router), ] for route_name, router in routes: check_mark(True, f"Route {route_name}: {router.prefix}") return True except Exception as e: check_mark(False, f"API routes error: {e}") return False def verify_authentication(): """Verify authentication setup.""" print_section("8. Authentication") try: from src.core.dependencies import hash_password, create_access_token from src.auth.dependencies import get_current_user check_mark(True, "hash_password function available") check_mark(True, "create_access_token function available") check_mark(True, "get_current_user dependency available") # Test password hashing with a short password test_password = "TestPwd123" try: hashed = hash_password(test_password) check_mark(len(hashed) > 0, f"Password hashing works: {len(hashed)} chars") except Exception as e: check_mark(False, f"Password hashing failed: {str(e)[:50]}") return True except Exception as e: check_mark(False, f"Authentication error: {str(e)[:50]}") return False def verify_middleware(): """Verify middleware setup.""" print_section("9. Middleware") try: from src.api.middleware.logging import RequestLoggingMiddleware check_mark(True, "RequestLoggingMiddleware available") return True except Exception as e: check_mark(False, f"Middleware error: {e}") return False async def verify_main_app(): """Verify main application setup.""" print_section("10. Main Application") try: from src.api.main import app, create_application check_mark(True, "FastAPI app created") check_mark(True, f"App title: {app.title}") # Check routes are included routes_count = len(app.routes) check_mark(routes_count > 5, f"Routes registered: {routes_count}") return True except Exception as e: check_mark(False, f"Main app error: {e}") return False async def main(): """Run all verification checks.""" print_header("COMPLETE SUPABASE SETUP VERIFICATION") results = [] # Run all checks results.append(("Configuration", await verify_config())) results.append(("Database Models", verify_database_models())) results.append(("Database Config", verify_database_config())) results.append(("Supabase Client", verify_supabase_client())) results.append(("Database Connection", await verify_database_connection())) results.append(("Pydantic Schemas", verify_pydantic_schemas())) results.append(("API Routes", verify_api_routes())) results.append(("Authentication", verify_authentication())) results.append(("Middleware", verify_middleware())) results.append(("Main Application", await verify_main_app())) # Print summary print_header("VERIFICATION SUMMARY") total = len(results) passed = sum(1 for _, result in results if result) for name, result in results: symbol = "✓" if result else "✗" print(f" {symbol} {name}") print(f"\n Total: {passed}/{total} checks passed ({(passed/total)*100:.0f}%)") if passed == total: print("\n 🎉 All checks passed! Project is ready for deployment.") else: print(f"\n ⚠️ {total - passed} checks failed. See details above.") print(f"\n{'='*70}\n") return passed == total if __name__ == "__main__": success = asyncio.run(main()) sys.exit(0 if success else 1)