#!/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)