82 lines
3.3 KiB
Python
Executable File
82 lines
3.3 KiB
Python
Executable File
import asyncio
|
|
import sys
|
|
import os
|
|
from pathlib import Path
|
|
|
|
# FONTOS: A dotenv betöltése minden app-specifikus import ELŐTT kell megtörténjen!
|
|
from dotenv import load_dotenv
|
|
env_path = Path("/home/coder/project/opt/service_finder/.env")
|
|
load_dotenv(dotenv_path=env_path)
|
|
|
|
# Útvonal beállítása a modulokhoz
|
|
sys.path.append("/home/coder/project/opt/service_finder/backend")
|
|
|
|
# Most már importálhatjuk a session-t, mert a környezeti változók már a memóriában vannak
|
|
from sqlalchemy import select
|
|
from app.db.session import AsyncSessionLocal # Javítva: AsyncSessionLocal-t használunk
|
|
from app.services.social_service import create_service_provider
|
|
from app.models.gamification import UserStats, PointsLedger
|
|
from app.models.user import User
|
|
from app.schemas.social import ServiceProviderCreate
|
|
|
|
async def run_test():
|
|
print("🚀 Gamifikációs integrációs teszt indul...")
|
|
|
|
# Az AsyncSessionLocal() egy context manager, így az 'async with' a helyes használat
|
|
async with AsyncSessionLocal() as db:
|
|
try:
|
|
# 1. Teszt felhasználó lekérése
|
|
result = await db.execute(select(User).limit(1))
|
|
user = result.scalars().first()
|
|
|
|
if not user:
|
|
print("❌ Hiba: Nincs felhasználó az adatbázisban a teszthez!")
|
|
return
|
|
|
|
print(f"👤 Teszt felhasználó: {user.email} (ID: {user.id})")
|
|
|
|
# 2. Új szolgáltató rögzítése (ez váltja ki a pontszerzést)
|
|
unique_suffix = os.urandom(2).hex()
|
|
test_provider = ServiceProviderCreate(
|
|
name=f"Teszt Szerviz {unique_suffix}",
|
|
address="Teszt utca 123.",
|
|
category="Service"
|
|
)
|
|
|
|
print(f"🛠️ Szolgáltató rögzítése: {test_provider.name}...")
|
|
new_provider = await create_service_provider(db, test_provider, user.id)
|
|
print(f"✅ Szolgáltató rögzítve (ID: {new_provider.id})")
|
|
|
|
# 3. Pontszám és napló ellenőrzése
|
|
# Megjegyzés: A social_service commit-ol, így itt újra le kell kérnünk az adatokat
|
|
stats_res = await db.execute(select(UserStats).where(UserStats.user_id == user.id))
|
|
stats = stats_res.scalar_one_or_none()
|
|
|
|
ledger_res = await db.execute(
|
|
select(PointsLedger)
|
|
.where(PointsLedger.user_id == user.id)
|
|
.order_by(PointsLedger.id.desc())
|
|
.limit(1)
|
|
)
|
|
last_entry = ledger_res.scalars().first()
|
|
|
|
print("\n" + "="*30)
|
|
print("📊 TESZT EREDMÉNYEK:")
|
|
if stats:
|
|
print(f"🏆 Összesített pontszám: {stats.total_points}")
|
|
print(f"📈 Aktuális szint: {stats.current_level}")
|
|
else:
|
|
print("⚠️ Figyelem: UserStats nem található (lehet, hogy most készült el?)")
|
|
|
|
if last_entry:
|
|
print(f"📝 Utolsó tranzakció: {last_entry.reason}")
|
|
print(f"💰 Jóváírt pont: {last_entry.points_change}")
|
|
print("="*30)
|
|
|
|
except Exception as e:
|
|
print(f"💥 Kritikus hiba a teszt futtatása közben: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run_test()) |