82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Simple test script to verify token refresh functionality
|
|
"""
|
|
import asyncio
|
|
import httpx
|
|
import json
|
|
|
|
async def test_token_refresh():
|
|
base_url = "http://sf_api:8000"
|
|
|
|
print("1. Logging in as tester_pro@profibot.hu...")
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
# Login
|
|
login_data = {
|
|
"username": "tester_pro@profibot.hu",
|
|
"password": "Password123!"
|
|
}
|
|
|
|
try:
|
|
resp = await client.post(
|
|
f"{base_url}/api/v1/auth/login",
|
|
data=login_data
|
|
)
|
|
resp.raise_for_status()
|
|
login_result = resp.json()
|
|
initial_token = login_result["access_token"]
|
|
print(f"✓ Initial token obtained: {initial_token[:50]}...")
|
|
except Exception as e:
|
|
print(f"Login failed: {e}")
|
|
return
|
|
|
|
# Test switching to personal mode
|
|
print("\n2. Switching to personal mode (organization_id = null)...")
|
|
headers = {"Authorization": f"Bearer {initial_token}", "Content-Type": "application/json"}
|
|
patch_data = {"organization_id": None}
|
|
|
|
try:
|
|
resp = await client.patch(
|
|
f"{base_url}/api/v1/users/me/active-organization",
|
|
json=patch_data,
|
|
headers=headers
|
|
)
|
|
resp.raise_for_status()
|
|
patch_result = 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')}")
|
|
|
|
if new_token != initial_token:
|
|
print("✓ Token refreshed successfully (tokens are different)")
|
|
else:
|
|
print("⚠️ Token not refreshed (tokens are the same)")
|
|
|
|
# Decode token to verify scope_id in payload
|
|
import jwt
|
|
from app.core.config import settings
|
|
try:
|
|
payload = jwt.decode(new_token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
|
|
print(f"✓ Token payload scope_id: {payload.get('scope_id')}")
|
|
print(f"✓ Token payload scope_level: {payload.get('scope_level')}")
|
|
except:
|
|
print("⚠️ Could not decode token")
|
|
|
|
except Exception as e:
|
|
print(f"PATCH failed: {e}")
|
|
if hasattr(e, 'response'):
|
|
try:
|
|
print(f"Response status: {e.response.status_code}")
|
|
print(f"Response text: {e.response.text}")
|
|
except:
|
|
pass
|
|
return
|
|
|
|
print("\n=== TEST COMPLETED SUCCESSFULLY ===")
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
success = asyncio.run(test_token_refresh())
|
|
exit(0 if success else 1) |