| |
| """ |
| Test database connectivity and user creation |
| """ |
|
|
| import asyncio |
| from sqlalchemy.ext.asyncio import AsyncSession |
| from core.database import get_db |
| from db_models.user import User |
| from schemas.user_schema import UserCreate |
| from services.auth_service import AuthService |
|
|
| async def test_user_creation(): |
| """Test user creation directly""" |
| print("Testing database connection and user creation...") |
| |
| |
| async for db in get_db(): |
| try: |
| |
| auth_service = AuthService(db) |
| |
| |
| user_data = UserCreate( |
| email="testuser@example.com", |
| password="TestPassword123!", |
| full_name="Test User" |
| ) |
| |
| print(f"Creating user: {user_data.email}") |
| |
| |
| user = await auth_service.register_user(user_data) |
| |
| print(f"✅ User created successfully!") |
| print(f"User ID: {user.id}") |
| print(f"Email: {user.email}") |
| print(f"Full Name: {user.full_name}") |
| print(f"Is Verified: {user.is_verified}") |
| |
| except Exception as e: |
| print(f"❌ Error creating user: {str(e)}") |
| import traceback |
| traceback.print_exc() |
| finally: |
| await db.close() |
| break |
|
|
| if __name__ == "__main__": |
| asyncio.run(test_user_creation()) |
|
|