126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Ground Zero Test Script - Step 1: Registration
|
|
|
|
This script tests the registration endpoint to ensure it returns 201 Created
|
|
even when email sending fails (due to SMTP timeout or Brevo API issues).
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
import uuid
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
# Add the app to path
|
|
sys.path.insert(0, '/app')
|
|
from app.core.config import settings
|
|
from app.models.identity import User
|
|
|
|
async def test_registration():
|
|
"""Test registration endpoint"""
|
|
print("🚀 Starting Ground Zero Test - Step 1: Registration")
|
|
print("=" * 60)
|
|
|
|
# Create async engine
|
|
engine = create_async_engine(
|
|
str(settings.SQLALCHEMY_DATABASE_URI),
|
|
echo=False,
|
|
pool_size=5,
|
|
)
|
|
|
|
async_session = sessionmaker(
|
|
engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
|
|
async with async_session() as session:
|
|
try:
|
|
# Generate unique test email
|
|
test_email = f"groundzero_test_{uuid.uuid4().hex[:8]}@example.com"
|
|
test_password = "TestPassword123!"
|
|
test_first_name = "Ground"
|
|
test_last_name = "Zero"
|
|
|
|
print(f"📝 Test data:")
|
|
print(f" Email: {test_email}")
|
|
print(f" Password: {test_password}")
|
|
print(f" Name: {test_first_name} {test_last_name}")
|
|
|
|
# Import FastAPI test client
|
|
from fastapi.testclient import TestClient
|
|
from app.main import app
|
|
|
|
client = TestClient(app)
|
|
|
|
# Prepare registration payload
|
|
payload = {
|
|
"email": test_email,
|
|
"password": test_password,
|
|
"first_name": test_first_name,
|
|
"last_name": test_last_name,
|
|
"region_code": "HU",
|
|
"lang": "hu"
|
|
}
|
|
|
|
print("\n📤 Sending POST request to /api/v1/auth/register...")
|
|
|
|
# Make the request
|
|
response = client.post("/api/v1/auth/register", json=payload)
|
|
|
|
print(f"📥 Response status: {response.status_code}")
|
|
print(f"📥 Response body: {response.text}")
|
|
|
|
# Assert response is 201 Created (not 500)
|
|
assert response.status_code == 201, f"Expected 201, got {response.status_code}"
|
|
|
|
# Parse response
|
|
response_data = response.json()
|
|
assert response_data["status"] == "success"
|
|
assert "user_id" in response_data
|
|
assert response_data["email"] == test_email
|
|
|
|
print(f"✅ Registration endpoint returned 201 Created")
|
|
print(f"✅ User ID: {response_data['user_id']}")
|
|
|
|
# Query database to verify user was inserted
|
|
print("\n🔍 Verifying database insertion...")
|
|
stmt = select(User).where(User.email == test_email)
|
|
result = await session.execute(stmt)
|
|
user = result.scalar_one_or_none()
|
|
|
|
assert user is not None, "User not found in database"
|
|
assert user.email == test_email
|
|
assert user.is_active == False # Should be pending until email verification
|
|
assert user.person.first_name == test_first_name
|
|
assert user.person.last_name == test_last_name
|
|
|
|
print(f"✅ User found in database with ID: {user.id}")
|
|
print(f"✅ User status: {'ACTIVE' if user.is_active else 'PENDING (correct)'}")
|
|
|
|
# Clean up test user
|
|
print("\n🧹 Cleaning up test user...")
|
|
await session.delete(user.person) # This will cascade delete user
|
|
await session.commit()
|
|
|
|
print("✅ Test user cleaned up")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("🎉 GROUND ZERO TEST PASSED!")
|
|
print("✅ Registration endpoint returns 201 Created")
|
|
print("✅ User is inserted into database with pending status")
|
|
print("✅ Email failure does not crash registration")
|
|
print("=" * 60)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ TEST FAILED: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
# Run the test
|
|
success = asyncio.run(test_registration())
|
|
sys.exit(0 if success else 1) |