220 lines
8.4 KiB
Python
220 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Teszt script a P0 Garage Selector Fix & Team Management API ellenőrzéséhez.
|
|
|
|
Használat:
|
|
docker compose exec sf_api python3 /app/tests/active/test_garage_selector_and_team_api.py
|
|
|
|
Ez a script:
|
|
1. Bejelentkezik admin@profibot.hu / Admin123! (Vezető Tervező, user_id=2)
|
|
2. Meghívja a GET /api/v1/organizations/my végpontot
|
|
3. Ellenőrzi, hogy a teszt cég (owner_id=2) megjelenik-e
|
|
4. Teszteli az új Team Management API végpontokat
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
import os
|
|
from datetime import datetime, timezone
|
|
|
|
# Add the backend directory to path
|
|
sys.path.insert(0, "/app")
|
|
|
|
import httpx
|
|
|
|
BASE_URL = os.environ.get("API_BASE_URL", "http://localhost:8000")
|
|
API_PREFIX = "/api/v1"
|
|
|
|
|
|
async def login(email: str, password: str) -> str:
|
|
"""Bejelentkezés és token lekérése."""
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
|
resp = await client.post(
|
|
f"{API_PREFIX}/auth/login",
|
|
data={"username": email, "password": password},
|
|
)
|
|
if resp.status_code != 200:
|
|
print(f"❌ Login failed: {resp.status_code} - {resp.text}")
|
|
sys.exit(1)
|
|
data = resp.json()
|
|
token = data.get("access_token") or data.get("token")
|
|
print(f"✅ Login successful (user: {email})")
|
|
return token
|
|
|
|
|
|
async def get_my_organizations(token: str) -> list:
|
|
"""GET /api/v1/organizations/my - Saját szervezetek lekérése."""
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
|
resp = await client.get(
|
|
f"{API_PREFIX}/organizations/my",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
print(f"\n{'='*70}")
|
|
print(f"📋 GET /api/v1/organizations/my")
|
|
print(f"{'='*70}")
|
|
print(f"Status: {resp.status_code}")
|
|
if resp.status_code != 200:
|
|
print(f"❌ Error: {resp.text}")
|
|
return []
|
|
|
|
data = resp.json()
|
|
print(f"Count: {len(data)} organizations found")
|
|
print(f"\nRaw response:")
|
|
print(json.dumps(data, indent=2, default=str))
|
|
|
|
return data
|
|
|
|
|
|
async def list_members(token: str, org_id: int) -> list:
|
|
"""GET /api/v1/organizations/{org_id}/members - Tagok listázása."""
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
|
resp = await client.get(
|
|
f"{API_PREFIX}/organizations/{org_id}/members",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
print(f"\n{'='*70}")
|
|
print(f"📋 GET /api/v1/organizations/{org_id}/members")
|
|
print(f"{'='*70}")
|
|
print(f"Status: {resp.status_code}")
|
|
if resp.status_code != 200:
|
|
print(f"❌ Error: {resp.text}")
|
|
return []
|
|
|
|
data = resp.json()
|
|
print(f"Count: {len(data)} members")
|
|
print(json.dumps(data, indent=2, default=str))
|
|
return data
|
|
|
|
|
|
async def create_invitation(token: str, org_id: int, email: str, role: str = "MANAGER") -> dict:
|
|
"""POST /api/v1/organizations/{org_id}/invitations - Meghívó küldése."""
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
|
resp = await client.post(
|
|
f"{API_PREFIX}/organizations/{org_id}/invitations",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json={"email": email, "role": role},
|
|
)
|
|
print(f"\n{'='*70}")
|
|
print(f"📋 POST /api/v1/organizations/{org_id}/invitations (email={email}, role={role})")
|
|
print(f"{'='*70}")
|
|
print(f"Status: {resp.status_code}")
|
|
print(f"Response: {resp.text}")
|
|
|
|
if resp.status_code in (200, 201):
|
|
return resp.json()
|
|
return None
|
|
|
|
|
|
async def update_member_role(token: str, org_id: int, member_id: int, role: str) -> dict:
|
|
"""PATCH /api/v1/organizations/{org_id}/members/{member_id} - Szerepkör módosítása."""
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
|
resp = await client.patch(
|
|
f"{API_PREFIX}/organizations/{org_id}/members/{member_id}",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
json={"role": role},
|
|
)
|
|
print(f"\n{'='*70}")
|
|
print(f"📋 PATCH /api/v1/organizations/{org_id}/members/{member_id} (role={role})")
|
|
print(f"{'='*70}")
|
|
print(f"Status: {resp.status_code}")
|
|
print(f"Response: {resp.text}")
|
|
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
return None
|
|
|
|
|
|
async def remove_member(token: str, org_id: int, member_id: int) -> dict:
|
|
"""DELETE /api/v1/organizations/{org_id}/members/{member_id} - Tag eltávolítása."""
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
|
resp = await client.delete(
|
|
f"{API_PREFIX}/organizations/{org_id}/members/{member_id}",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
print(f"\n{'='*70}")
|
|
print(f"📋 DELETE /api/v1/organizations/{org_id}/members/{member_id}")
|
|
print(f"{'='*70}")
|
|
print(f"Status: {resp.status_code}")
|
|
print(f"Response: {resp.text}")
|
|
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
return None
|
|
|
|
|
|
async def main():
|
|
print("=" * 70)
|
|
print("🔍 P0 GARAGE SELECTOR FIX & TEAM MANAGEMENT API TESZT")
|
|
print("=" * 70)
|
|
|
|
# 1. Bejelentkezés admin@profibot.hu (Vezető Tervező, user_id=2)
|
|
token = await login("admin@profibot.hu", "Admin123!")
|
|
|
|
# 2. Saját szervezetek lekérése
|
|
orgs = await get_my_organizations(token)
|
|
|
|
if not orgs:
|
|
print("\n❌ CRITICAL: No organizations returned! Garage Selector would be empty!")
|
|
print(" This means the get_my_organizations query is still broken.")
|
|
sys.exit(1)
|
|
|
|
# 3. Ellenőrizzük, hogy van-e olyan org ahol owner_id=2 (admin user)
|
|
admin_owned = [o for o in orgs if o.get("owner_id") == 2]
|
|
if admin_owned:
|
|
print(f"\n✅ Garage Selector FIX VERIFIED: {len(admin_owned)} organization(s) where owner_id=2:")
|
|
for o in admin_owned:
|
|
print(f" - ID: {o['organization_id']}, Name: {o.get('name', 'N/A')}, Role: {o.get('user_role', 'N/A')}")
|
|
else:
|
|
print(f"\n⚠️ No organizations found with owner_id=2. Checking all orgs...")
|
|
for o in orgs:
|
|
print(f" - ID: {o['organization_id']}, Name: {o.get('name', 'N/A')}, Owner: {o.get('owner_id', 'N/A')}, Role: {o.get('user_role', 'N/A')}")
|
|
|
|
# 4. Ha van legalább egy szervezet, teszteljük a Team Management API-t
|
|
if orgs:
|
|
test_org_id = orgs[0]["organization_id"]
|
|
print(f"\n{'='*70}")
|
|
print(f"🧪 TEAM MANAGEMENT API TESZTEK (org_id={test_org_id})")
|
|
print(f"{'='*70}")
|
|
|
|
# 4a. Tagok listázása
|
|
members = await list_members(token, test_org_id)
|
|
|
|
# 4b. Meghívó küldése egy nem létező emailre (invited_email teszt)
|
|
test_email = f"test_invite_{datetime.now(timezone.utc).timestamp()}@example.com"
|
|
invite_result = await create_invitation(token, test_org_id, test_email, "MANAGER")
|
|
|
|
if invite_result:
|
|
print(f"\n✅ Invitation created successfully!")
|
|
|
|
# 4c. Tagok újralistázása (ellenőrizzük, hogy megjelent-e a meghívó)
|
|
members_after = await list_members(token, test_org_id)
|
|
|
|
# Keressük az új meghívót
|
|
pending_invites = [m for m in members_after if m.get("status") == "pending"]
|
|
if pending_invites:
|
|
print(f"\n✅ Pending invitation found in member list!")
|
|
new_member_id = pending_invites[0]["id"]
|
|
|
|
# 4d. Szerepkör módosítása
|
|
role_result = await update_member_role(token, test_org_id, new_member_id, "ADMIN")
|
|
if role_result:
|
|
print(f"\n✅ Role update successful!")
|
|
|
|
# 4e. Tag eltávolítása (meghívó visszavonása)
|
|
remove_result = await remove_member(token, test_org_id, new_member_id)
|
|
if remove_result:
|
|
print(f"\n✅ Member removal successful!")
|
|
else:
|
|
print(f"\n⚠️ No pending invitations found in member list.")
|
|
else:
|
|
print(f"\n⚠️ Could not create test invitation (might already exist).")
|
|
|
|
print(f"\n{'='*70}")
|
|
print("✅ TESZT BEFEJEZŐDÖTT")
|
|
print(f"{'='*70}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|