198 lines
7.5 KiB
Python
198 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for PATCH /admin/users/{user_id} endpoint.
|
|
Uses form-data for OAuth2 login, then tests the admin user update endpoint.
|
|
"""
|
|
import httpx
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
|
|
BASE_URL = "http://sf_api:8000/api/v1"
|
|
|
|
async def test_admin_user_patch():
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
|
# Step 1: Login as admin using form data (OAuth2PasswordRequestForm)
|
|
print("=" * 60)
|
|
print("STEP 1: Login as admin")
|
|
print("=" * 60)
|
|
|
|
login_data = {
|
|
"username": "admin@profibot.hu",
|
|
"password": "Admin123!",
|
|
"remember_me": "true"
|
|
}
|
|
|
|
r = await client.post("/auth/login", data=login_data)
|
|
print(f"Status: {r.status_code}")
|
|
|
|
if r.status_code != 200:
|
|
print(f"Login failed: {r.text}")
|
|
return False
|
|
|
|
token_data = r.json()
|
|
access_token = token_data.get("access_token")
|
|
print(f"Token received: {access_token[:50]}...")
|
|
|
|
headers = {"Authorization": f"Bearer {access_token}"}
|
|
|
|
# Step 2: Try to find a non-superadmin user
|
|
print("\n" + "=" * 60)
|
|
print("STEP 2: Find a non-superadmin user")
|
|
print("=" * 60)
|
|
|
|
# Try user IDs 2, 3, 4, etc. until we find a regular user
|
|
test_user_id = None
|
|
for uid in [2, 3, 4, 5]:
|
|
r = await client.get(f"/admin/users/{uid}", headers=headers)
|
|
if r.status_code == 200:
|
|
user_data = r.json()
|
|
role = user_data.get("role", "")
|
|
print(f" User ID={uid}: email={user_data.get('email')}, role={role}")
|
|
if role != "superadmin":
|
|
test_user_id = uid
|
|
break
|
|
else:
|
|
print(f" User ID={uid}: Status={r.status_code}")
|
|
|
|
if test_user_id is None:
|
|
print("No non-superadmin user found. Using user ID 2 as fallback.")
|
|
test_user_id = 2
|
|
|
|
# Get the user details
|
|
r = await client.get(f"/admin/users/{test_user_id}", headers=headers)
|
|
if r.status_code != 200:
|
|
print(f"Failed to get user {test_user_id}: {r.text}")
|
|
return False
|
|
|
|
user_data = r.json()
|
|
user_id = user_data.get("id")
|
|
user_email = user_data.get("email")
|
|
user_role = user_data.get("role")
|
|
print(f"\nUsing test user: ID={user_id}, Email={user_email}, Role={user_role}")
|
|
|
|
# Step 3: Get user details before update
|
|
print("\n" + "=" * 60)
|
|
print(f"STEP 3: User details (before update)")
|
|
print("=" * 60)
|
|
|
|
print(f" is_vip: {user_data.get('is_vip')}")
|
|
print(f" preferred_language: {user_data.get('preferred_language')}")
|
|
print(f" is_active: {user_data.get('is_active')}")
|
|
print(f" region_code: {user_data.get('region_code')}")
|
|
print(f" preferred_currency: {user_data.get('preferred_currency')}")
|
|
print(f" subscription_plan: {user_data.get('subscription_plan')}")
|
|
|
|
# Step 4: PATCH the user
|
|
print("\n" + "=" * 60)
|
|
print(f"STEP 4: PATCH user {user_id} - update fields")
|
|
print("=" * 60)
|
|
|
|
patch_data = {
|
|
"is_vip": True,
|
|
"preferred_language": "en",
|
|
"region_code": "HU",
|
|
"preferred_currency": "HUF",
|
|
}
|
|
|
|
r = await client.patch(f"/admin/users/{user_id}", json=patch_data, headers=headers)
|
|
print(f"Status: {r.status_code}")
|
|
|
|
if r.status_code != 200:
|
|
print(f"PATCH failed: {r.text}")
|
|
return False
|
|
|
|
updated_user = r.json()
|
|
print(f"User after update:")
|
|
print(f" is_vip: {updated_user.get('is_vip')}")
|
|
print(f" preferred_language: {updated_user.get('preferred_language')}")
|
|
print(f" region_code: {updated_user.get('region_code')}")
|
|
print(f" preferred_currency: {updated_user.get('preferred_currency')}")
|
|
|
|
# Verify the changes
|
|
assert updated_user.get("is_vip") == True, "is_vip not updated"
|
|
assert updated_user.get("preferred_language") == "en", "preferred_language not updated"
|
|
assert updated_user.get("region_code") == "HU", "region_code not updated"
|
|
assert updated_user.get("preferred_currency") == "HUF", "preferred_currency not updated"
|
|
|
|
print("\n✅ All basic field updates verified!")
|
|
|
|
# Step 5: Test person sub-object update
|
|
print("\n" + "=" * 60)
|
|
print(f"STEP 5: PATCH user {user_id} - update person fields")
|
|
print("=" * 60)
|
|
|
|
patch_person_data = {
|
|
"person": {
|
|
"last_name": "Updated",
|
|
"first_name": "User",
|
|
"phone": "+36123456789"
|
|
}
|
|
}
|
|
|
|
r = await client.patch(f"/admin/users/{user_id}", json=patch_person_data, headers=headers)
|
|
print(f"Status: {r.status_code}")
|
|
|
|
if r.status_code != 200:
|
|
print(f"PATCH person failed: {r.text}")
|
|
else:
|
|
updated_user = r.json()
|
|
person = updated_user.get("person", {})
|
|
print(f"Person after update:")
|
|
print(f" last_name: {person.get('last_name')}")
|
|
print(f" first_name: {person.get('first_name')}")
|
|
print(f" phone: {person.get('phone')}")
|
|
|
|
if person.get("last_name") == "Updated":
|
|
print("✅ Person fields updated successfully!")
|
|
else:
|
|
print("⚠️ Person fields may not have been updated (might not have a person record)")
|
|
|
|
# Step 6: Test error cases
|
|
print("\n" + "=" * 60)
|
|
print("STEP 6: Test error cases")
|
|
print("=" * 60)
|
|
|
|
# 6a: Non-existent user
|
|
r = await client.patch("/admin/users/99999", json={"is_vip": True}, headers=headers)
|
|
print(f"6a - Non-existent user (99999): Status={r.status_code}")
|
|
if r.status_code == 404:
|
|
print("✅ Correctly returned 404")
|
|
else:
|
|
print(f"⚠️ Unexpected: {r.text}")
|
|
|
|
# 6b: Empty update
|
|
r = await client.patch(f"/admin/users/{user_id}", json={}, headers=headers)
|
|
print(f"6b - Empty update: Status={r.status_code}")
|
|
if r.status_code == 400:
|
|
print("✅ Correctly returned 400 for empty update")
|
|
elif r.status_code == 200:
|
|
print("⚠️ Empty update returned 200 (no changes made)")
|
|
else:
|
|
print(f"⚠️ Unexpected: {r.text}")
|
|
|
|
# 6c: Invalid field
|
|
r = await client.patch(f"/admin/users/{user_id}", json={"nonexistent_field": "test"}, headers=headers)
|
|
print(f"6c - Invalid field: Status={r.status_code}")
|
|
if r.status_code == 422:
|
|
print("✅ Correctly returned 422 for invalid field")
|
|
else:
|
|
print(f"⚠️ Unexpected: {r.text}")
|
|
|
|
# 6d: Try to modify superadmin (should be blocked)
|
|
r = await client.patch("/admin/users/1", json={"is_vip": True}, headers=headers)
|
|
print(f"6d - Modify superadmin: Status={r.status_code}")
|
|
if r.status_code == 403:
|
|
print("✅ Correctly blocked superadmin modification")
|
|
else:
|
|
print(f"⚠️ Unexpected: {r.text}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("ALL TESTS COMPLETED")
|
|
print("=" * 60)
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
success = asyncio.run(test_admin_user_patch())
|
|
sys.exit(0 if success else 1)
|