""" End-to-end test for user registration flow with Mailpit email interception. This test validates the complete user journey: 1. Register with a unique email (Lite registration) 2. Intercept activation email via Mailpit API 3. Extract verification token and call verify-email endpoint 4. Login with credentials 5. Complete KYC with dummy data 6. Verify gamification endpoint returns 200 OK """ import asyncio import httpx import pytest import uuid import re import logging from typing import Dict, Optional from datetime import date logger = logging.getLogger(__name__) # Configuration BASE_URL = "http://sf_api:8000" MAILPIT_URL = "http://sf_mailpit:8025" TEST_EMAIL_DOMAIN = "example.com" class MailpitClient: """Client for interacting with Mailpit API.""" def __init__(self, base_url: str = MAILPIT_URL): self.base_url = base_url self.client = httpx.AsyncClient(timeout=30.0) async def get_latest_message(self) -> Optional[Dict]: """Fetch the latest email message from Mailpit.""" try: response = await self.client.get(f"{self.base_url}/api/v1/messages?limit=1") response.raise_for_status() data = response.json() if data.get("messages"): return data["messages"][0] return None except Exception as e: logger.error(f"Failed to fetch latest message: {e}") return None async def get_message_content(self, message_id: str) -> Optional[str]: """Get the full content (HTML and text) of a specific message.""" try: response = await self.client.get(f"{self.base_url}/api/v1/message/{message_id}") response.raise_for_status() data = response.json() # Prefer text over HTML return data.get("Text") or data.get("HTML") or "" except Exception as e: logger.error(f"Failed to fetch message content: {e}") return None async def cleanup(self): """Close the HTTP client.""" await self.client.aclose() class APIClient: """Client for interacting with the Service Finder API.""" def __init__(self, base_url: str = BASE_URL): self.base_url = base_url self.client = httpx.AsyncClient(timeout=30.0) self.token = None async def register(self, email: str, password: str = "TestPassword123!") -> httpx.Response: """Register a new user.""" payload = { "email": email, "password": password, "first_name": "Test", "last_name": "User", "region_code": "HU", "lang": "hu" } response = await self.client.post(f"{self.base_url}/api/v1/auth/register", json=payload) return response async def login(self, email: str, password: str = "TestPassword123!") -> Optional[str]: """Login and return JWT token.""" payload = { "username": email, "password": password } response = await self.client.post(f"{self.base_url}/api/v1/auth/login", data=payload) if response.status_code == 200: data = response.json() self.token = data.get("access_token") return self.token return None async def verify_email(self, token: str) -> httpx.Response: """Call verify-email endpoint with token.""" payload = {"token": token} response = await self.client.post(f"{self.base_url}/api/v1/auth/verify-email", json=payload) return response async def complete_kyc(self, kyc_data: Dict) -> httpx.Response: """Complete KYC for current user.""" headers = {} if self.token: headers["Authorization"] = f"Bearer {self.token}" response = await self.client.post(f"{self.base_url}/api/v1/auth/complete-kyc", json=kyc_data, headers=headers) return response async def get_gamification(self) -> httpx.Response: """Get gamification data for current user.""" headers = {} if self.token: headers["Authorization"] = f"Bearer {self.token}" response = await self.client.get(f"{self.base_url}/api/v1/gamification/me", headers=headers) return response async def cleanup(self): """Close the HTTP client.""" await self.client.aclose() def extract_verification_token(text: str) -> Optional[str]: """ Extract verification token from email text using regex. Looks for UUID patterns in URLs or plain text. """ # Pattern for UUID (version 4) uuid_pattern = r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}' match = re.search(uuid_pattern, text, re.IGNORECASE) if match: return match.group(0) # Fallback: look for token parameter in URL token_pattern = r'(?:token|code)=([0-9a-f\-]+)' match = re.search(token_pattern, text, re.IGNORECASE) if match: return match.group(1) return None def create_dummy_kyc_data() -> Dict: """Create dummy KYC data for testing.""" return { "phone_number": "+36123456789", "birth_place": "Budapest", "birth_date": "1990-01-01", "mothers_last_name": "Kovács", "mothers_first_name": "Éva", "address_zip": "1011", "address_city": "Budapest", "address_street_name": "Kossuth", "address_street_type": "utca", "address_house_number": "1", "address_stairwell": "A", "address_floor": "2", "address_door": "3", "address_hrsz": None, "identity_docs": { "ID_CARD": { "number": "123456AB", "expiry_date": "2030-12-31" } }, "ice_contact": { "name": "Test Contact", "phone": "+36198765432", "relationship": "parent" }, "preferred_language": "hu", "preferred_currency": "HUF" } @pytest.mark.asyncio async def test_user_registration_flow(): """Main E2E test for user registration flow.""" # Generate unique test email test_email = f"test_{uuid.uuid4().hex[:8]}@{TEST_EMAIL_DOMAIN}" logger.info(f"Using test email: {test_email}") # Initialize clients api_client = APIClient() mailpit = MailpitClient() try: # 1. Register new user (Lite registration) logger.info("Step 1: Registering user") reg_response = await api_client.register(test_email) assert reg_response.status_code in (200, 201, 202), f"Registration failed: {reg_response.status_code} - {reg_response.text}" logger.info(f"Registration response: {reg_response.status_code}") # 2. Wait for email (Mailpit may need a moment) await asyncio.sleep(2) # 3. Fetch latest email from Mailpit logger.info("Step 2: Fetching email from Mailpit") message = await mailpit.get_latest_message() assert message is not None, "No email found in Mailpit" logger.info(f"Found email with ID: {message.get('ID')}, Subject: {message.get('Subject')}") # 4. Get email content and extract verification token content = await mailpit.get_message_content(message["ID"]) assert content, "Email content is empty" token = extract_verification_token(content) assert token is not None, f"Could not extract verification token from email content: {content[:500]}" logger.info(f"Extracted verification token: {token}") # 5. Verify email using the token logger.info("Step 3: Verifying email") verify_response = await api_client.verify_email(token) assert verify_response.status_code in (200, 201, 202), f"Email verification failed: {verify_response.status_code} - {verify_response.text}" logger.info(f"Email verification response: {verify_response.status_code}") # 6. Login to get JWT token logger.info("Step 4: Logging in") token = await api_client.login(test_email) assert token is not None, "Login failed - no token received" logger.info("Login successful, token obtained") # 7. Complete KYC with dummy data logger.info("Step 5: Completing KYC") kyc_data = create_dummy_kyc_data() kyc_response = await api_client.complete_kyc(kyc_data) assert kyc_response.status_code in (200, 201, 202), f"KYC completion failed: {kyc_response.status_code} - {kyc_response.text}" logger.info(f"KYC completion response: {kyc_response.status_code}") # 8. Verify gamification endpoint logger.info("Step 6: Checking gamification endpoint") gamification_response = await api_client.get_gamification() assert gamification_response.status_code == 200, f"Gamification endpoint failed: {gamification_response.status_code} - {gamification_response.text}" logger.info(f"Gamification response: {gamification_response.status_code}") # Optional: Validate response structure gamification_data = gamification_response.json() assert "points" in gamification_data or "level" in gamification_data or "achievements" in gamification_data, \ "Gamification response missing expected fields" logger.info("✅ All steps passed! User registration flow works end-to-end.") finally: # Cleanup await api_client.cleanup() await mailpit.cleanup() if __name__ == "__main__": # For manual testing import sys logging.basicConfig(level=logging.INFO) asyncio.run(test_user_registration_flow())