131 lines
5.1 KiB
Python
131 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify the token refresh functionality in PATCH /api/v1/users/me/active-organization
|
|
"""
|
|
import asyncio
|
|
import aiohttp
|
|
import json
|
|
|
|
async def test_token_refresh():
|
|
base_url = "http://sf_api:8000"
|
|
|
|
# 1. Login to get initial token
|
|
print("1. Logging in as tester_pro@profibot.hu...")
|
|
async with aiohttp.ClientSession() as session:
|
|
# Login
|
|
login_data = {
|
|
"username": "tester_pro@profibot.hu",
|
|
"password": "TestPassword123!"
|
|
}
|
|
|
|
async with session.post(
|
|
f"{base_url}/api/v1/auth/login",
|
|
data=login_data
|
|
) as resp:
|
|
if resp.status != 200:
|
|
print(f"Login failed: {resp.status}")
|
|
text = await resp.text()
|
|
print(f"Response: {text}")
|
|
return
|
|
|
|
login_result = await resp.json()
|
|
initial_token = login_result["access_token"]
|
|
print(f"✓ Initial token obtained: {initial_token[:50]}...")
|
|
|
|
# 2. Test switching to personal mode (organization_id = null)
|
|
print("\n2. Switching to personal mode (organization_id = null)...")
|
|
headers = {"Authorization": f"Bearer {initial_token}", "Content-Type": "application/json"}
|
|
patch_data = {"organization_id": None}
|
|
|
|
async with session.patch(
|
|
f"{base_url}/api/v1/users/me/active-organization",
|
|
json=patch_data,
|
|
headers=headers
|
|
) as resp:
|
|
if resp.status != 200:
|
|
print(f"PATCH failed: {resp.status}")
|
|
text = await resp.text()
|
|
print(f"Response: {text}")
|
|
return
|
|
|
|
patch_result = await resp.json()
|
|
new_token = patch_result["access_token"]
|
|
user_data = patch_result["user"]
|
|
print(f"✓ New token received: {new_token[:50]}...")
|
|
print(f"✓ User scope_id: {user_data.get('scope_id')}")
|
|
print(f"✓ Token type: {patch_result.get('token_type')}")
|
|
|
|
# Verify tokens are different
|
|
if new_token != initial_token:
|
|
print("✓ Token refreshed successfully (tokens are different)")
|
|
else:
|
|
print("⚠️ Token not refreshed (tokens are the same)")
|
|
|
|
# 3. Test switching to Alpha organization (ID 26)
|
|
print("\n3. Switching to Alpha organization (ID 26)...")
|
|
headers = {"Authorization": f"Bearer {new_token}", "Content-Type": "application/json"}
|
|
patch_data = {"organization_id": "26"}
|
|
|
|
async with session.patch(
|
|
f"{base_url}/api/v1/users/me/active-organization",
|
|
json=patch_data,
|
|
headers=headers
|
|
) as resp:
|
|
if resp.status != 200:
|
|
print(f"PATCH failed: {resp.status}")
|
|
text = await resp.text()
|
|
print(f"Response: {text}")
|
|
return
|
|
|
|
patch_result = await resp.json()
|
|
alpha_token = patch_result["access_token"]
|
|
user_data = patch_result["user"]
|
|
print(f"✓ New token for Alpha: {alpha_token[:50]}...")
|
|
print(f"✓ User scope_id: {user_data.get('scope_id')}")
|
|
|
|
if alpha_token != new_token:
|
|
print("✓ Token refreshed again for Alpha organization")
|
|
else:
|
|
print("⚠️ Token not refreshed for Alpha")
|
|
|
|
# 4. Test switching to Beta organization (ID 27)
|
|
print("\n4. Switching to Beta organization (ID 27)...")
|
|
headers = {"Authorization": f"Bearer {alpha_token}", "Content-Type": "application/json"}
|
|
patch_data = {"organization_id": "27"}
|
|
|
|
async with session.patch(
|
|
f"{base_url}/api/v1/users/me/active-organization",
|
|
json=patch_data,
|
|
headers=headers
|
|
) as resp:
|
|
if resp.status != 200:
|
|
print(f"PATCH failed: {resp.status}")
|
|
text = await resp.text()
|
|
print(f"Response: {text}")
|
|
return
|
|
|
|
patch_result = await resp.json()
|
|
beta_token = patch_result["access_token"]
|
|
user_data = patch_result["user"]
|
|
print(f"✓ New token for Beta: {beta_token[:50]}...")
|
|
print(f"✓ User scope_id: {user_data.get('scope_id')}")
|
|
|
|
if beta_token != alpha_token:
|
|
print("✓ Token refreshed again for Beta organization")
|
|
else:
|
|
print("⚠️ Token not refreshed for Beta")
|
|
|
|
# 5. Verify all tokens are different
|
|
print("\n5. Verifying all tokens are unique...")
|
|
tokens = [initial_token, new_token, alpha_token, beta_token]
|
|
unique_tokens = set(tokens)
|
|
|
|
if len(unique_tokens) == len(tokens):
|
|
print("✓ All tokens are unique (proper refresh on each organization switch)")
|
|
else:
|
|
print(f"⚠️ Only {len(unique_tokens)} unique tokens out of {len(tokens)}")
|
|
|
|
print("\n=== TEST COMPLETED SUCCESSFULLY ===")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_token_refresh()) |