kincses/service-finder
1
0
Fork 0
Code Issues 7 Pull Requests Packages Projects 2 Releases Wiki Activity

chore: Backend codebase cleanup and archiving of legacy scripts

Browse Source
This commit is contained in:
Roo
2026-03-22 20:07:37 +00:00
parent 5d96b00f81
commit 309a72cc0b
19 changed files with 530 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

388
backend/archive_v1_scripts/create_sandbox_user.py Normal file
View File

@@ -0,0 +1,388 @@
#!/usr/bin/env python3
"""
Sandbox Seeder Script - Creates a persistent sandbox user in the live/dev database
for manual testing via Swagger.
Steps:
1. Register via POST /api/v1/auth/register
2. Extract verification token from Mailpit API
3. Verify email via POST /api/v1/auth/verify-email
4. Login via POST /api/v1/auth/login to get JWT
5. Complete KYC via POST /api/v1/auth/complete-kyc
6. Create organization via POST /api/v1/organizations/onboard
7. Add a test vehicle/asset via appropriate endpoint
8. Add a fuel expense (15,000 HUF) via POST /api/v1/expenses/add
Prints credentials and IDs for immediate use.
"""
import asyncio
import httpx
import json
import sys
import time
from datetime import date, datetime, timedelta
import uuid
# Configuration
API_BASE = "http://localhost:8000" # FastAPI server (runs inside sf_api container)
MAILPIT_API = "http://sf_mailpit:8025/api/v1/messages"
MAILPIT_DELETE_ALL = "http://sf_mailpit:8025/api/v1/messages"
# Generate unique email each run to avoid duplicate key errors
unique_id = int(time.time())
SANDBOX_EMAIL = f"sandbox_{unique_id}@test.com"
SANDBOX_PASSWORD = "Sandbox123!"
SANDBOX_FIRST_NAME = "Sandbox"
SANDBOX_LAST_NAME = "User"
# Dummy KYC data
DUMMY_KYC = {
"phone_number": "+36123456789",
"birth_place": "Budapest",
"birth_date": "1990-01-01",
"mothers_last_name": "Kovács",
"mothers_first_name": "Éva",
"address_zip": "1051",
"address_city": "Budapest",
"address_street_name": "Váci",
"address_street_type": "utca",
"address_house_number": "1",
"address_stairwell": None,
"address_floor": None,
"address_door": None,
"address_hrsz": None,
"identity_docs": {
"ID_CARD": {
"number": "123456AB",
"expiry_date": "2030-12-31"
}
},
"ice_contact": {
"name": "John Doe",
"phone": "+36198765432",
"relationship": "friend"
},
"preferred_language": "hu",
"preferred_currency": "HUF"
}
# Dummy organization data
DUMMY_ORG = {
"full_name": "Sandbox Test Kft.",
"name": "Sandbox Kft.",
"display_name": "Sandbox Test",
"tax_number": f"{unique_id}"[:8] + "-1-42",
"reg_number": f"01-09-{unique_id}"[:6],
"country_code": "HU",
"language": "hu",
"default_currency": "HUF",
"address_zip": "1051",
"address_city": "Budapest",
"address_street_name": "Váci",
"address_street_type": "utca",
"address_house_number": "2",
"address_stairwell": None,
"address_floor": None,
"address_door": None,
"address_hrsz": None,
"contacts": [
{
"full_name": "Sandbox User",
"email": SANDBOX_EMAIL,
"phone": "+36123456789",
"contact_type": "primary"
}
]
}
# Dummy vehicle data
DUMMY_VEHICLE = {
"catalog_id": 1, # Assuming there's at least one catalog entry
"license_plate": f"SBX-{uuid.uuid4().hex[:4]}".upper(),
"vin": f"VIN{uuid.uuid4().hex[:10]}".upper(),
"nickname": "Sandbox Car",
"purchase_date": "2025-01-01",
"initial_mileage": 5000,
"fuel_type": "petrol",
"transmission": "manual"
}
# Dummy expense data
DUMMY_EXPENSE = {
"asset_id": None, # Will be filled after vehicle creation
"category": "fuel",
"amount": 15000.0,
"date": date.today().isoformat()
}
async def clean_mailpit():
"""Delete all messages in Mailpit before registration to ensure clean state."""
print(" [DEBUG] Entering clean_mailpit()")
async with httpx.AsyncClient() as client:
try:
print(f" [DEBUG] Sending DELETE to {MAILPIT_DELETE_ALL}")
resp = await client.delete(MAILPIT_DELETE_ALL)
print(f" [DEBUG] DELETE response status: {resp.status_code}")
if resp.status_code == 200:
print("🗑️ Mailpit cleaned (all messages deleted).")
else:
print(f"⚠️ Mailpit clean returned {resp.status_code}, continuing anyway.")
except Exception as e:
print(f"⚠️ Mailpit clean failed: {e}, continuing anyway.")
async def fetch_mailpit_token():
"""Fetch the latest verification token from Mailpit with polling."""
import re
import sys
max_attempts = 5
wait_seconds = 3
print(f"[DEBUG] Starting fetch_mailpit_token() with max_attempts={max_attempts}", flush=True)
async with httpx.AsyncClient() as client:
for attempt in range(1, max_attempts + 1):
try:
print(f"[DEBUG] Fetching Mailpit messages (attempt {attempt}/{max_attempts})...", flush=True)
resp = await client.get(MAILPIT_API)
resp.raise_for_status()
messages = resp.json()
# Debug: print raw response summary
total = messages.get("total", 0)
count = messages.get("count", 0)
print(f"[DEBUG] Mailpit response: total={total}, count={count}", flush=True)
if not messages.get("messages"):
print(f"⚠️ No emails in Mailpit (attempt {attempt}/{max_attempts}). Waiting {wait_seconds}s...", flush=True)
await asyncio.sleep(wait_seconds)
continue
# Print each message's subject and recipients for debugging
for idx, msg in enumerate(messages.get("messages", [])):
subject = msg.get("Subject", "No Subject")
to_list = msg.get("To", [])
from_list = msg.get("From", [])
print(f"[DEBUG] Message {idx}: Subject='{subject}', To={to_list}, From={from_list}", flush=True)
print(f"[DEBUG] Looking for email to {SANDBOX_EMAIL}...", flush=True)
# Find the latest email to our sandbox email
for msg in messages.get("messages", []):
# Check if email is in To field (which is a list of dicts)
to_list = msg.get("To", [])
email_found = False
for recipient in to_list:
if isinstance(recipient, dict) and recipient.get("Address") == SANDBOX_EMAIL:
email_found = True
break
elif isinstance(recipient, str) and recipient == SANDBOX_EMAIL:
email_found = True
break
if email_found:
msg_id = msg.get("ID")
print(f"[DEBUG] Found email to {SANDBOX_EMAIL}, message ID: {msg_id}")
# Fetch full message details (Text and HTML are empty in list response)
if msg_id:
try:
# Correct endpoint: /api/v1/message/{id} (singular)
detail_resp = await client.get(f"http://sf_mailpit:8025/api/v1/message/{msg_id}")
detail_resp.raise_for_status()
detail = detail_resp.json()
body = detail.get("Text", "")
html_body = detail.get("HTML", "")
print(f"[DEBUG] Fetched full message details, body length: {len(body)}, HTML length: {len(html_body)}")
except Exception as e:
print(f"[DEBUG] Failed to fetch message details: {e}")
body = msg.get("Text", "")
html_body = msg.get("HTML", "")
else:
body = msg.get("Text", "")
html_body = msg.get("HTML", "")
if body:
print(f"[DEBUG] Body preview (first 500 chars): {body[:500]}...")
# Try to find token using patterns from test suite
patterns = [
r"token=([a-zA-Z0-9\-_]+)",
r"/verify/([a-zA-Z0-9\-_]+)",
r"verification code: ([a-zA-Z0-9\-_]+)",
r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", # UUID pattern
r"[0-9a-f]{32}", # UUID without hyphens
]
for pattern in patterns:
if body:
token_match = re.search(pattern, body, re.I)
if token_match:
token = token_match.group(1) if token_match.groups() else token_match.group(0)
print(f"✅ Token found with pattern '{pattern}' on attempt {attempt}: {token}")
return token
# If not found in text, try HTML body
if html_body:
for pattern in patterns:
html_token_match = re.search(pattern, html_body, re.I)
if html_token_match:
token = html_token_match.group(1) if html_token_match.groups() else html_token_match.group(0)
print(f"✅ Token found in HTML with pattern '{pattern}' on attempt {attempt}: {token}")
return token
print(f"[DEBUG] No token pattern found. Body length: {len(body)}, HTML length: {len(html_body)}")
if body:
print(f"[DEBUG] Full body (first 1000 chars): {body[:1000]}")
if html_body:
print(f"[DEBUG] HTML body snippet (first 500 chars): {html_body[:500]}")
print(f"⚠️ Email found but no token (attempt {attempt}/{max_attempts}). Waiting {wait_seconds}s...")
await asyncio.sleep(wait_seconds)
except Exception as e:
print(f"❌ Mailpit API error on attempt {attempt}: {e}")
await asyncio.sleep(wait_seconds)
print("❌ Could not retrieve token after all attempts.")
return None
async def main():
print("🚀 Starting Sandbox User Creation...")
async with httpx.AsyncClient(base_url=API_BASE, timeout=30.0) as client:
# Step 0: Clean Mailpit to ensure only new emails
print("0. Cleaning Mailpit...")
await clean_mailpit()
# Step 1: Register
print("1. Registering user...")
register_data = {
"email": SANDBOX_EMAIL,
"password": SANDBOX_PASSWORD,
"first_name": SANDBOX_FIRST_NAME,
"last_name": SANDBOX_LAST_NAME,
"region_code": "HU",
"lang": "hu",
"timezone": "Europe/Budapest"
}
resp = await client.post("/api/v1/auth/register", json=register_data)
if resp.status_code not in (200, 201):
print(f"❌ Registration failed: {resp.status_code} {resp.text}")
return
print("✅ Registration successful.")
# Step 2: Get token from Mailpit
print("2. Fetching verification token from Mailpit...")
token = await fetch_mailpit_token()
if not token:
print("❌ Could not retrieve token. Exiting.")
return
print(f"✅ Token found: {token}")
# Step 3: Verify email
print("3. Verifying email...")
resp = await client.post("/api/v1/auth/verify-email", json={"token": token})
if resp.status_code != 200:
print(f"❌ Email verification failed: {resp.status_code} {resp.text}")
return
print("✅ Email verified.")
# Step 4: Login
print("4. Logging in...")
resp = await client.post("/api/v1/auth/login", data={
"username": SANDBOX_EMAIL,
"password": SANDBOX_PASSWORD
})
if resp.status_code != 200:
print(f"❌ Login failed: {resp.status_code} {resp.text}")
return
login_data = resp.json()
access_token = login_data.get("access_token")
if not access_token:
print("❌ No access token in login response.")
return
print("✅ Login successful.")
# Update client headers with JWT
client.headers.update({"Authorization": f"Bearer {access_token}"})
# Step 5: Complete KYC
print("5. Completing KYC...")
resp = await client.post("/api/v1/auth/complete-kyc", json=DUMMY_KYC)
if resp.status_code != 200:
print(f"❌ KYC completion failed: {resp.status_code} {resp.text}")
# Continue anyway (maybe KYC optional)
else:
print("✅ KYC completed.")
# Step 6: Create organization
print("6. Creating organization...")
resp = await client.post("/api/v1/organizations/onboard", json=DUMMY_ORG)
if resp.status_code not in (200, 201):
print(f"❌ Organization creation failed: {resp.status_code} {resp.text}")
# Continue anyway (maybe optional)
org_id = None
else:
org_data = resp.json()
org_id = org_data.get("organization_id")
print(f"✅ Organization created with ID: {org_id}")
# Step 7: Add vehicle/asset
print("7. Adding vehicle/asset...")
asset_id = None
# Try POST /api/v1/assets
resp = await client.post("/api/v1/assets", json=DUMMY_VEHICLE)
if resp.status_code in (200, 201):
asset_data = resp.json()
asset_id = asset_data.get("asset_id") or asset_data.get("id")
print(f"✅ Asset created via /api/v1/assets, ID: {asset_id}")
else:
# Try POST /api/v1/vehicles
resp = await client.post("/api/v1/vehicles", json=DUMMY_VEHICLE)
if resp.status_code in (200, 201):
asset_data = resp.json()
asset_id = asset_data.get("vehicle_id") or asset_data.get("id")
print(f"✅ Vehicle created via /api/v1/vehicles, ID: {asset_id}")
else:
# Try POST /api/v1/catalog/claim
resp = await client.post("/api/v1/catalog/claim", json={
"catalog_id": DUMMY_VEHICLE["catalog_id"],
"license_plate": DUMMY_VEHICLE["license_plate"]
})
if resp.status_code in (200, 201):
asset_data = resp.json()
asset_id = asset_data.get("asset_id") or asset_data.get("id")
print(f"✅ Asset claimed via /api/v1/catalog/claim, ID: {asset_id}")
else:
print(f"⚠️ Could not create vehicle/asset. Skipping. Status: {resp.status_code}, Response: {resp.text}")
# Step 8: Add expense (if asset created)
if asset_id:
print("8. Adding expense (15,000 HUF fuel)...")
expense_data = DUMMY_EXPENSE.copy()
expense_data["asset_id"] = asset_id
resp = await client.post("/api/v1/expenses/add", json=expense_data)
if resp.status_code in (200, 201):
print("✅ Expense added.")
else:
print(f"⚠️ Expense addition failed: {resp.status_code} {resp.text}")
else:
print("⚠️ Skipping expense because no asset ID.")
# Final output
print("\n" + "="*60)
print("🎉 SANDBOX USER CREATION COMPLETE!")
print("="*60)
print(f"Email: {SANDBOX_EMAIL}")
print(f"Password: {SANDBOX_PASSWORD}")
print(f"JWT Access Token: {access_token}")
print(f"Organization ID: {org_id}")
print(f"Asset/Vehicle ID: {asset_id}")
print(f"Login via Swagger: {API_BASE}/docs")
print("="*60)
print("\nYou can now use these credentials for manual testing.")
print("Note: The user is fully verified and has a dummy organization,")
print("a dummy vehicle, and a fuel expense of 15,000 HUF.")
print("="*60)
if __name__ == "__main__":
asyncio.run(main())