138 lines
5.0 KiB
Python
138 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for verifying the POST /api/v1/organizations/onboard fix.
|
|
Creates a new organization with unique tax number and verifies that
|
|
a default main branch is automatically created.
|
|
"""
|
|
import asyncio
|
|
import aiohttp
|
|
import json
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
API_BASE = "http://localhost:8000/api/v1"
|
|
TEST_USER_EMAIL = "kincses@gmail.com"
|
|
TEST_USER_PASSWORD = "MiskociA74" # From .env PGADMIN password (likely same)
|
|
|
|
async def get_auth_token(session):
|
|
"""Authenticate and get JWT token using form data."""
|
|
login_url = f"{API_BASE}/auth/login"
|
|
# Use form data as required by OAuth2PasswordRequestForm
|
|
form_data = aiohttp.FormData()
|
|
form_data.add_field('username', TEST_USER_EMAIL)
|
|
form_data.add_field('password', TEST_USER_PASSWORD)
|
|
|
|
try:
|
|
async with session.post(login_url, data=form_data) as resp:
|
|
if resp.status != 200:
|
|
print(f"Login failed: {resp.status}")
|
|
text = await resp.text()
|
|
print(f"Response: {text}")
|
|
return None
|
|
data = await resp.json()
|
|
return data.get("access_token")
|
|
except Exception as e:
|
|
print(f"Login error: {e}")
|
|
return None
|
|
|
|
async def create_test_organization(session, token):
|
|
"""Create a new organization with unique tax number."""
|
|
onboard_url = f"{API_BASE}/organizations/onboard"
|
|
# Generate unique tax number: 98765432-1-11 + timestamp to avoid conflicts
|
|
timestamp = datetime.now().strftime("%H%M%S")
|
|
tax_number = f"98765432-1-{timestamp[-2:]}" # Use last 2 digits of timestamp
|
|
|
|
org_data = {
|
|
"full_name": "Auto-Fixer Pro Kft.",
|
|
"name": "AutoFixerPro",
|
|
"display_name": "Auto-Fixer Pro",
|
|
"tax_number": tax_number,
|
|
"reg_number": "01-09-123456",
|
|
"country_code": "HU",
|
|
"address_zip": "1234",
|
|
"address_city": "Budapest",
|
|
"address_street_name": "Teszt utca",
|
|
"address_street_type": "utca",
|
|
"address_house_number": "42",
|
|
"address_hrsz": ""
|
|
}
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
async with session.post(onboard_url, json=org_data, headers=headers) as resp:
|
|
if resp.status != 201:
|
|
print(f"Organization creation failed: {resp.status}")
|
|
text = await resp.text()
|
|
print(f"Response: {text}")
|
|
return None
|
|
data = await resp.json()
|
|
print(f"Organization created successfully: {data}")
|
|
return data.get("organization_id")
|
|
|
|
async def get_my_organizations(session, token):
|
|
"""Get list of organizations for current user."""
|
|
url = f"{API_BASE}/organizations/my"
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
async with session.get(url, headers=headers) as resp:
|
|
if resp.status != 200:
|
|
print(f"Failed to get organizations: {resp.status}")
|
|
return []
|
|
return await resp.json()
|
|
|
|
async def check_branches_via_sql(org_id):
|
|
"""Check if branch exists via direct SQL query in container."""
|
|
import subprocess
|
|
# Use docker exec to query PostgreSQL
|
|
cmd = [
|
|
"docker", "exec", "shared-postgres",
|
|
"psql", "-U", "kincses", "-d", "service_finder", "-c",
|
|
f"SELECT id, name, is_main FROM fleet.branches WHERE organization_id = {org_id};"
|
|
]
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
print("SQL Query Result:")
|
|
print(result.stdout)
|
|
if result.returncode != 0:
|
|
print(f"SQL Error: {result.stderr}")
|
|
return "Központi Telephely" in result.stdout
|
|
except Exception as e:
|
|
print(f"SQL check error: {e}")
|
|
return False
|
|
|
|
async def main():
|
|
print("=== Testing Organization Onboard Fix ===")
|
|
print("1. Authenticating...")
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
token = await get_auth_token(session)
|
|
if not token:
|
|
print("Authentication failed. Exiting.")
|
|
sys.exit(1)
|
|
print(f"Token obtained: {token[:20]}...")
|
|
|
|
print("\n2. Creating test organization...")
|
|
org_id = await create_test_organization(session, token)
|
|
if not org_id:
|
|
print("Organization creation failed. Exiting.")
|
|
sys.exit(1)
|
|
print(f"Organization ID: {org_id}")
|
|
|
|
print("\n3. Fetching user's organizations...")
|
|
orgs = await get_my_organizations(session, token)
|
|
print(f"Total organizations for user: {len(orgs)}")
|
|
for org in orgs:
|
|
if org.get("organization_id") == org_id:
|
|
print(f" Found created org: {org}")
|
|
break
|
|
|
|
print("\n4. Checking for default branch via SQL...")
|
|
branch_exists = await check_branches_via_sql(org_id)
|
|
if branch_exists:
|
|
print("✅ SUCCESS: Default main branch 'Központi Telephely' was automatically created!")
|
|
else:
|
|
print("❌ FAIL: No default branch found.")
|
|
sys.exit(1)
|
|
|
|
print("\n=== Test completed successfully ===")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |