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