140 lines
5.7 KiB
Python
140 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
API teszt script User 55 számára - hivatalos API végpontok használatával
|
|
"""
|
|
import sys
|
|
import asyncio
|
|
import httpx
|
|
import json
|
|
|
|
sys.path.append('/app/backend')
|
|
|
|
from app.db.session import AsyncSessionLocal
|
|
from sqlalchemy import select
|
|
from app.models.identity import User
|
|
from app.core.security import create_tokens
|
|
|
|
async def get_auth_token():
|
|
"""Hitelesítési token lekérése User 55 számára"""
|
|
async with AsyncSessionLocal() as db:
|
|
result = await db.execute(select(User).where(User.id == 55))
|
|
user = result.scalar_one_or_none()
|
|
if user:
|
|
access_token, _ = create_tokens(data={'sub': str(user.id)})
|
|
return access_token
|
|
else:
|
|
raise Exception("User 55 nem található")
|
|
|
|
async def test_api_endpoints():
|
|
"""API végpontok tesztelése"""
|
|
# Token lekérése
|
|
token = await get_auth_token()
|
|
print("✅ Hitelesítési token megkapva User 55 számára")
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
async with httpx.AsyncClient(base_url="http://localhost:8000", timeout=30.0) as client:
|
|
print("\n=== API TESZTEK ===")
|
|
|
|
# 1. Szervezetek listázása
|
|
print("\n1. GET /api/v1/organizations/my - Szervezetek listázása")
|
|
try:
|
|
response = await client.get("/api/v1/organizations/my", headers=headers)
|
|
print(f" Status: {response.status_code}")
|
|
|
|
if response.status_code == 200:
|
|
orgs = response.json()
|
|
print(f" Talált szervezetek: {len(orgs)} db")
|
|
|
|
for i, org in enumerate(orgs, 1):
|
|
print(f" {i}. {org.get('full_name')} (ID: {org.get('organization_id')})")
|
|
print(f" Típus: {org.get('name')}, Adószám: {org.get('tax_number')}")
|
|
print(f" Státusz: {org.get('status')}, Aktív: {org.get('is_active')}")
|
|
else:
|
|
print(f" Hiba: {response.text}")
|
|
except Exception as e:
|
|
print(f" Hiba a kérés során: {e}")
|
|
|
|
# 2. Gamification ellenőrzés (ha van API)
|
|
print("\n2. Gamification státusz (adatbázis ellenőrzés)")
|
|
async with AsyncSessionLocal() as db:
|
|
from sqlalchemy import text
|
|
result = await db.execute(text('''
|
|
SELECT total_xp, current_level
|
|
FROM gamification.user_stats
|
|
WHERE user_id = 55
|
|
'''))
|
|
stats = result.fetchone()
|
|
if stats:
|
|
stats_dict = dict(stats._mapping)
|
|
print(f" Összes XP: {stats_dict['total_xp']}")
|
|
print(f" Jelenlegi szint: {stats_dict['current_level']}")
|
|
|
|
# Points ledger ellenőrzés
|
|
result = await db.execute(text('''
|
|
SELECT reason, points, created_at::date as date
|
|
FROM gamification.points_ledger
|
|
WHERE user_id = 55
|
|
ORDER BY created_at DESC
|
|
'''))
|
|
ledger = result.fetchall()
|
|
print(f" Ponttörténet: {len(ledger)} bejegyzés")
|
|
for entry in ledger:
|
|
entry_dict = dict(entry._mapping)
|
|
print(f" - {entry_dict['reason']}: {entry_dict['points']} pont ({entry_dict['date']})")
|
|
|
|
# 3. User scope ellenőrzés
|
|
print("\n3. User scope információ")
|
|
async with AsyncSessionLocal() as db:
|
|
result = await db.execute(text('''
|
|
SELECT scope_level, scope_id
|
|
FROM identity.users
|
|
WHERE id = 55
|
|
'''))
|
|
user_scope = result.fetchone()
|
|
if user_scope:
|
|
scope_dict = dict(user_scope._mapping)
|
|
print(f" Scope level: {scope_dict['scope_level']}")
|
|
print(f" Scope ID: {scope_dict['scope_id']}")
|
|
|
|
# Scope organization ellenőrzés
|
|
if scope_dict['scope_id']:
|
|
result = await db.execute(text(f'''
|
|
SELECT full_name, org_type
|
|
FROM fleet.organizations
|
|
WHERE id = {scope_dict['scope_id']}
|
|
'''))
|
|
scope_org = result.fetchone()
|
|
if scope_org:
|
|
scope_org_dict = dict(scope_org._mapping)
|
|
print(f" Scope szervezet: {scope_org_dict['full_name']} ({scope_org_dict['org_type']})")
|
|
|
|
# 4. Telephelyek ellenőrzés
|
|
print("\n4. Telephelyek (adatbázis)")
|
|
async with AsyncSessionLocal() as db:
|
|
result = await db.execute(text('''
|
|
SELECT b.name, b.is_main, o.full_name as org_name
|
|
FROM fleet.branches b
|
|
JOIN fleet.organizations o ON b.organization_id = o.id
|
|
WHERE b.organization_id IN (
|
|
SELECT organization_id FROM fleet.organization_members WHERE user_id = 55
|
|
)
|
|
ORDER BY b.is_main DESC
|
|
'''))
|
|
branches = result.fetchall()
|
|
print(f" Talált telephelyek: {len(branches)} db")
|
|
for branch in branches:
|
|
branch_dict = dict(branch._mapping)
|
|
main_str = "FŐTELEPHELY" if branch_dict['is_main'] else "melléktelephely"
|
|
print(f" - {branch_dict['org_name']}: {branch_dict['name']} ({main_str})")
|
|
|
|
async def main():
|
|
print("=== USER 55 API TESZTELÉSE ===")
|
|
print("Felhasználó: test_fallback@profibot.hu (ID: 55)")
|
|
await test_api_endpoints()
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |