#!/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: Get list of users to find a test user print("\n" + "=" * 60) print("STEP 2: Get users list") print("=" * 60) r = await client.get("/admin/users/", headers=headers) print(f"Status: {r.status_code}") if r.status_code != 200: print(f"Failed to get users: {r.text}") return False users_data = r.json() users = users_data if isinstance(users_data, list) else users_data.get("items", users_data.get("data", [])) print(f"Found {len(users)} users") # Find a non-superadmin user to test with test_user = None for u in users: if u.get("email") != "admin@profibot.hu": test_user = u break if not test_user: print("No test user found (only admin exists)") # Try to get user by ID 1 or 2 for uid in [2, 3, 1]: r = await client.get(f"/admin/users/{uid}", headers=headers) if r.status_code == 200: test_user = r.json() print(f"Found user via direct lookup: {test_user.get('email')}") break if not test_user: print("Could not find any test user") return False user_id = test_user.get("id") user_email = test_user.get("email") print(f"Test user: ID={user_id}, Email={user_email}") # Step 3: Get user details before update print("\n" + "=" * 60) print(f"STEP 3: Get user {user_id} details (before update)") print("=" * 60) r = await client.get(f"/admin/users/{user_id}", headers=headers) print(f"Status: {r.status_code}") if r.status_code != 200: print(f"Failed to get user details: {r.text}") return False user_before = r.json() print(f"User before update:") print(f" is_vip: {user_before.get('is_vip')}") print(f" preferred_language: {user_before.get('preferred_language')}") print(f" is_active: {user_before.get('is_active')}") # 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", "max_vehicles": 15, "max_garages": 5, } 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')}") print(f" max_vehicles: {updated_user.get('max_vehicles')}") print(f" max_garages: {updated_user.get('max_garages')}") # 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" assert updated_user.get("max_vehicles") == 15, "max_vehicles not updated" assert updated_user.get("max_garages") == 5, "max_garages 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 status: {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 status: {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 status: {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)