admin frontend elkezdése, járművek tisztázása, frontend fejleszts
This commit is contained in:
Roo
245
backend/scripts/test_kyc_e2e.py
Normal file
245
backend/scripts/test_kyc_e2e.py
Normal file
@@ -0,0 +1,245 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
E2E Test: Light Registration → Email Verification → KYC Completion
|
||||
Verifies that all necessary database relationships are created after KYC.
|
||||
"""
|
||||
import asyncio
|
||||
import httpx
|
||||
import uuid
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
|
||||
sys.path.insert(0, '/app')
|
||||
|
||||
from sqlalchemy import text
|
||||
from app.db.session import AsyncSessionLocal
|
||||
|
||||
BASE_URL = "http://localhost:8000/api/v1"
|
||||
TEST_EMAIL = f"test_kyc_{uuid.uuid4().hex[:8]}@example.com"
|
||||
TEST_PASSWORD = "TestPass123!"
|
||||
|
||||
async def get_verification_token(email: str):
|
||||
async with AsyncSessionLocal() as session:
|
||||
result = await session.execute(
|
||||
text("SELECT token FROM identity.verification_tokens vt "
|
||||
"JOIN identity.users u ON u.id = vt.user_id "
|
||||
"WHERE u.email = :email ORDER BY vt.created_at DESC LIMIT 1"),
|
||||
{"email": email}
|
||||
)
|
||||
return result.scalar()
|
||||
|
||||
async def check_infrastructure(user_id: int):
|
||||
"""Check all required infrastructure objects exist."""
|
||||
async with AsyncSessionLocal() as session:
|
||||
checks = {}
|
||||
|
||||
# Person
|
||||
r = await session.execute(
|
||||
text("SELECT id, is_active, user_id FROM identity.persons WHERE id = "
|
||||
"(SELECT person_id FROM identity.users WHERE id = :uid)"),
|
||||
{"uid": user_id}
|
||||
)
|
||||
person = r.fetchone()
|
||||
checks["person"] = person is not None
|
||||
if person:
|
||||
checks["person_active"] = person[1] == True
|
||||
checks["person_user_id"] = person[2] == user_id
|
||||
|
||||
# Organization
|
||||
r = await session.execute(
|
||||
text("SELECT id, name, org_type FROM fleet.organizations WHERE owner_id = :uid"),
|
||||
{"uid": user_id}
|
||||
)
|
||||
org = r.fetchone()
|
||||
checks["organization"] = org is not None
|
||||
if org:
|
||||
checks["org_id"] = org[0]
|
||||
|
||||
# Branch
|
||||
if org:
|
||||
r = await session.execute(
|
||||
text("SELECT id, name, is_main FROM fleet.branches WHERE organization_id = :oid"),
|
||||
{"oid": org[0]}
|
||||
)
|
||||
branch = r.fetchone()
|
||||
checks["branch"] = branch is not None
|
||||
if branch:
|
||||
checks["branch_main"] = branch[2] == True
|
||||
|
||||
# OrganizationMember
|
||||
r = await session.execute(
|
||||
text("SELECT id, role FROM fleet.organization_members WHERE user_id = :uid"),
|
||||
{"uid": user_id}
|
||||
)
|
||||
member = r.fetchone()
|
||||
checks["org_member"] = member is not None
|
||||
if member:
|
||||
checks["org_member_role"] = member[1] == "OWNER"
|
||||
|
||||
# Wallet
|
||||
r = await session.execute(
|
||||
text("SELECT id FROM identity.wallets WHERE user_id = :uid"),
|
||||
{"uid": user_id}
|
||||
)
|
||||
checks["wallet"] = r.fetchone() is not None
|
||||
|
||||
# UserStats
|
||||
r = await session.execute(
|
||||
text("SELECT user_id, total_xp, current_level FROM gamification.user_stats WHERE user_id = :uid"),
|
||||
{"uid": user_id}
|
||||
)
|
||||
stats = r.fetchone()
|
||||
checks["user_stats"] = stats is not None
|
||||
if stats:
|
||||
checks["user_stats_xp"] = stats[1] >= 500 # KYC bonus
|
||||
checks["user_stats_level"] = stats[2] >= 1
|
||||
|
||||
# User scope_id
|
||||
r = await session.execute(
|
||||
text("SELECT scope_id FROM identity.users WHERE id = :uid"),
|
||||
{"uid": user_id}
|
||||
)
|
||||
scope = r.fetchone()
|
||||
checks["scope_id_set"] = scope is not None and scope[0] is not None
|
||||
|
||||
return checks
|
||||
|
||||
async def run_test():
|
||||
passed = 0
|
||||
failed = 0
|
||||
|
||||
print("=" * 60)
|
||||
print(f"KYC E2E TEST - Email: {TEST_EMAIL}")
|
||||
print("=" * 60)
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
# --- STEP 1: LITE REGISTRATION ---
|
||||
print("\n--- 1. LITE REGISTRATION ---")
|
||||
reg_payload = {
|
||||
"email": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD,
|
||||
"first_name": "KYC",
|
||||
"last_name": "TestUser",
|
||||
"region_code": "HU"
|
||||
}
|
||||
resp = await client.post(f"{BASE_URL}/auth/register", json=reg_payload)
|
||||
print(f"Status: {resp.status_code}")
|
||||
if resp.status_code == 201:
|
||||
print("✓ Registration successful")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ Registration failed: {resp.text}")
|
||||
failed += 1
|
||||
return passed, failed
|
||||
|
||||
# --- STEP 2: EMAIL VERIFICATION ---
|
||||
print("\n--- 2. EMAIL VERIFICATION ---")
|
||||
token = await get_verification_token(TEST_EMAIL)
|
||||
if not token:
|
||||
print("✗ No verification token found!")
|
||||
failed += 1
|
||||
return passed, failed
|
||||
|
||||
print(f"Token: {token}")
|
||||
verify_resp = await client.post(f"{BASE_URL}/auth/verify-email", json={"token": str(token)})
|
||||
print(f"Status: {verify_resp.status_code}")
|
||||
if verify_resp.status_code == 200:
|
||||
print("✓ Email verification successful")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ Verification failed: {verify_resp.text}")
|
||||
failed += 1
|
||||
return passed, failed
|
||||
|
||||
# --- STEP 3: LOGIN TO GET TOKEN ---
|
||||
print("\n--- 3. LOGIN ---")
|
||||
login_data = {
|
||||
"username": TEST_EMAIL,
|
||||
"password": TEST_PASSWORD,
|
||||
"grant_type": "password"
|
||||
}
|
||||
login_resp = await client.post(f"{BASE_URL}/auth/login", data=login_data)
|
||||
print(f"Status: {login_resp.status_code}")
|
||||
if login_resp.status_code == 200:
|
||||
print("✓ Login successful")
|
||||
passed += 1
|
||||
access_token = login_resp.json().get("access_token")
|
||||
else:
|
||||
print(f"✗ Login failed: {login_resp.text}")
|
||||
failed += 1
|
||||
return passed, failed
|
||||
|
||||
headers = {"Authorization": f"Bearer {access_token}"}
|
||||
|
||||
# --- STEP 4: GET USER INFO ---
|
||||
print("\n--- 4. GET USER INFO ---")
|
||||
me_resp = await client.get(f"{BASE_URL}/auth/me", headers=headers)
|
||||
print(f"Status: {me_resp.status_code}")
|
||||
if me_resp.status_code == 200:
|
||||
user_data = me_resp.json()
|
||||
user_id = user_data.get("id")
|
||||
print(f"User ID: {user_id}")
|
||||
print(f"User data: {user_data}")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ Failed to get user info: {me_resp.text}")
|
||||
failed += 1
|
||||
return passed, failed
|
||||
|
||||
# --- STEP 5: COMPLETE KYC ---
|
||||
print("\n--- 5. COMPLETE KYC ---")
|
||||
kyc_payload = {
|
||||
"first_name": "KYC",
|
||||
"last_name": "TestUser",
|
||||
"birth_date": "1990-01-15",
|
||||
"birth_place": "Budapest",
|
||||
"mothers_last_name": "Anyuka",
|
||||
"mothers_first_name": "Mária",
|
||||
"phone_number": "+36301234567",
|
||||
"address_zip": "1011",
|
||||
"address_city": "Budapest",
|
||||
"address_street_name": "Fő utca",
|
||||
"address_street_type": "utca",
|
||||
"address_house_number": "1",
|
||||
"preferred_currency": "HUF",
|
||||
"region_code": "HU",
|
||||
"identity_docs": {
|
||||
"PASSPORT": {"number": "AB123456", "expiry_date": "2030-01-01"}
|
||||
}
|
||||
}
|
||||
kyc_resp = await client.post(f"{BASE_URL}/auth/complete-kyc", json=kyc_payload, headers=headers)
|
||||
print(f"Status: {kyc_resp.status_code}")
|
||||
print(f"Response: {kyc_resp.text}")
|
||||
if kyc_resp.status_code == 200:
|
||||
print("✓ KYC completion successful")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ KYC failed: {kyc_resp.text}")
|
||||
failed += 1
|
||||
return passed, failed
|
||||
|
||||
# --- STEP 6: VERIFY INFRASTRUCTURE ---
|
||||
print("\n--- 6. VERIFY INFRASTRUCTURE ---")
|
||||
checks = await check_infrastructure(user_id)
|
||||
all_ok = True
|
||||
for check_name, check_result in checks.items():
|
||||
status = "✓" if check_result else "✗"
|
||||
if not check_result:
|
||||
all_ok = False
|
||||
print(f" {status} {check_name}: {check_result}")
|
||||
|
||||
if all_ok:
|
||||
print("✓ All infrastructure checks passed!")
|
||||
passed += 1
|
||||
else:
|
||||
print(f"✗ Some infrastructure checks failed!")
|
||||
failed += 1
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print(f"RESULTS: {passed} passed, {failed} failed")
|
||||
print("=" * 60)
|
||||
return passed, failed
|
||||
|
||||
if __name__ == "__main__":
|
||||
p, f = asyncio.run(run_test())
|
||||
sys.exit(1 if f > 0 else 0)
|
||||
Reference in New Issue
Block a user