175 lines
6.7 KiB
Python
175 lines
6.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Backend verification test for scope switcher functionality.
|
|
|
|
Simulates the frontend:
|
|
1. Login as User 28 (tester_pro@profibot.hu) - get JWT token
|
|
2. Call PATCH /active-organization with organization_id = null
|
|
3. Call GET /vehicles - assert it returns exactly 1 car (MNO-345)
|
|
4. Call PATCH /active-organization with organization_id = 15
|
|
5. Call GET /vehicles - assert it returns exactly 1 car (PQR-678)
|
|
|
|
Run inside sf_api container:
|
|
docker compose exec sf_api python -m app.scripts.verify_scope_switcher
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
import httpx
|
|
import json
|
|
from typing import Dict, Any
|
|
|
|
# Add the backend directory to the path
|
|
sys.path.insert(0, '/app')
|
|
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.models.identity import User
|
|
from app.core.security import create_tokens
|
|
from app.core.config import settings
|
|
|
|
async def get_auth_token() -> str:
|
|
"""Get JWT token for test user by simulating login."""
|
|
async with AsyncSessionLocal() as db:
|
|
# Find the test user
|
|
from sqlalchemy import select
|
|
stmt = select(User).where(User.email == "tester_pro@profibot.hu")
|
|
result = await db.execute(stmt)
|
|
user = result.scalar_one_or_none()
|
|
|
|
if not user:
|
|
raise Exception("Test user not found")
|
|
|
|
# Create token directly (simulating login)
|
|
from app.core.security import DEFAULT_RANK_MAP
|
|
ranks = DEFAULT_RANK_MAP # Simplified
|
|
|
|
token_data = {
|
|
"sub": str(user.id),
|
|
"role": user.role.value if hasattr(user.role, 'value') else str(user.role),
|
|
"rank": ranks.get(user.role.value.upper(), 10),
|
|
"scope_level": user.scope_level or "individual",
|
|
"scope_id": str(user.scope_id) if user.scope_id else str(user.id)
|
|
}
|
|
|
|
access_token, _ = create_tokens(data=token_data)
|
|
return access_token
|
|
|
|
async def make_api_request(method: str, endpoint: str, token: str, data: Dict = None) -> Dict[str, Any]:
|
|
"""Make HTTP request to the API."""
|
|
base_url = "http://localhost:8000/api/v1"
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
url = f"{base_url}{endpoint}"
|
|
|
|
if method == "GET":
|
|
response = await client.get(url, headers=headers)
|
|
elif method == "PATCH":
|
|
response = await client.patch(url, headers=headers, json=data)
|
|
else:
|
|
raise ValueError(f"Unsupported method: {method}")
|
|
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def main():
|
|
"""Main verification test."""
|
|
print("=" * 60)
|
|
print("SCOPE SWITCHER VERIFICATION TEST")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
# 1. Get auth token
|
|
print("\n1. AUTHENTICATING AS TEST USER...")
|
|
token = await get_auth_token()
|
|
print(f" ✅ Token obtained (length: {len(token)})")
|
|
|
|
# 2. Test personal mode (organization_id = null)
|
|
print("\n2. TESTING PERSONAL MODE (organization_id = null)...")
|
|
patch_data = {"organization_id": None}
|
|
patch_response = await make_api_request("PATCH", "/users/me/active-organization", token, patch_data)
|
|
print(f" ✅ PATCH response: {json.dumps(patch_response, indent=2)}")
|
|
|
|
# Check if scope_id was updated
|
|
if patch_response.get("scope_id") is not None:
|
|
print(" ❌ ERROR: scope_id should be null in personal mode!")
|
|
return
|
|
|
|
# 3. Get vehicles in personal mode
|
|
print("\n3. GETTING VEHICLES IN PERSONAL MODE...")
|
|
vehicles_response = await make_api_request("GET", "/assets/vehicles", token)
|
|
vehicles_count = len(vehicles_response)
|
|
print(f" ✅ Found {vehicles_count} vehicles")
|
|
|
|
# Check license plates
|
|
license_plates = [v.get("license_plate") for v in vehicles_response if v.get("license_plate")]
|
|
print(f" ✅ License plates: {license_plates}")
|
|
|
|
# Should have exactly 1 car (MNO-345)
|
|
if vehicles_count != 1:
|
|
print(f" ❌ ERROR: Expected 1 vehicle in personal mode, got {vehicles_count}")
|
|
return
|
|
|
|
if "MNO-345" not in license_plates:
|
|
print(f" ❌ ERROR: Expected MNO-345 in personal mode, got {license_plates}")
|
|
return
|
|
|
|
print(" ✅ PERSONAL MODE TEST PASSED: Exactly 1 car (MNO-345)")
|
|
|
|
# 4. Test corporate mode (organization_id = 15)
|
|
print("\n4. TESTING CORPORATE MODE (organization_id = 15)...")
|
|
patch_data = {"organization_id": 15}
|
|
patch_response = await make_api_request("PATCH", "/users/me/active-organization", token, patch_data)
|
|
print(f" ✅ PATCH response: {json.dumps(patch_response, indent=2)}")
|
|
|
|
# Check if scope_id was updated
|
|
if patch_response.get("scope_id") != "15":
|
|
print(f" ❌ ERROR: scope_id should be '15', got {patch_response.get('scope_id')}")
|
|
return
|
|
|
|
# 5. Get vehicles in corporate mode
|
|
print("\n5. GETTING VEHICLES IN CORPORATE MODE...")
|
|
vehicles_response = await make_api_request("GET", "/assets/vehicles", token)
|
|
vehicles_count = len(vehicles_response)
|
|
print(f" ✅ Found {vehicles_count} vehicles")
|
|
|
|
# Check license plates
|
|
license_plates = [v.get("license_plate") for v in vehicles_response if v.get("license_plate")]
|
|
print(f" ✅ License plates: {license_plates}")
|
|
|
|
# Should have exactly 1 car (PQR-678)
|
|
if vehicles_count != 1:
|
|
print(f" ❌ ERROR: Expected 1 vehicle in corporate mode, got {vehicles_count}")
|
|
return
|
|
|
|
if "PQR-678" not in license_plates:
|
|
print(f" ❌ ERROR: Expected PQR-678 in corporate mode, got {license_plates}")
|
|
return
|
|
|
|
print(" ✅ CORPORATE MODE TEST PASSED: Exactly 1 car (PQR-678)")
|
|
|
|
# 6. Final verification
|
|
print("\n" + "=" * 60)
|
|
print("✅ ALL TESTS PASSED!")
|
|
print("=" * 60)
|
|
print("\nSUMMARY:")
|
|
print("- Personal mode (scope_id = null): Shows 1 vehicle (MNO-345)")
|
|
print("- Corporate mode (scope_id = 15): Shows 1 vehicle (PQR-678)")
|
|
print("- Scope switcher endpoint works correctly")
|
|
print("- Vehicle filtering by branch/organization works correctly")
|
|
|
|
except httpx.HTTPStatusError as e:
|
|
print(f"\n❌ HTTP ERROR: {e}")
|
|
print(f"Response: {e.response.text}")
|
|
return
|
|
except Exception as e:
|
|
print(f"\n❌ ERROR: {type(e).__name__}: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |