#!/usr/bin/env python3 """ Egyszerűsített igazságszérum teszt - csak a lényeges assert-ek. """ import asyncio import sys import os from decimal import Decimal from datetime import datetime, timedelta from uuid import uuid4 sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'backend')) from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from sqlalchemy import select, func from app.database import Base from app.models.identity import User, Wallet, ActiveVoucher, Person from app.models.payment import PaymentIntent, PaymentIntentStatus from app.models.audit import FinancialLedger, LedgerEntryType, WalletType from app.services.payment_router import PaymentRouter from app.services.billing_engine import SmartDeduction from app.core.config import settings DATABASE_URL = settings.DATABASE_URL.replace("postgresql://", "postgresql+asyncpg://") engine = create_async_engine(DATABASE_URL, echo=False) AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async def main(): print("=== IGAZSÁGSZÉRUM TESZT (Egyszerűsített) ===") session = AsyncSessionLocal() try: # 1. Teszt felhasználók létrehozása print("1. Teszt felhasználók létrehozása...") email_payer = f"test_payer_{uuid4().hex[:8]}@test.local" email_beneficiary = f"test_beneficiary_{uuid4().hex[:8]}@test.local" person_payer = Person(last_name="TestPayer", first_name="Test", is_active=True) person_beneficiary = Person(last_name="TestBeneficiary", first_name="Test", is_active=True) session.add_all([person_payer, person_beneficiary]) await session.flush() user_payer = User(email=email_payer, role="user", person_id=person_payer.id, is_active=True) user_beneficiary = User(email=email_beneficiary, role="user", person_id=person_beneficiary.id, is_active=True) session.add_all([user_payer, user_beneficiary]) await session.flush() wallet_payer = Wallet(user_id=user_payer.id, earned_credits=0, purchased_credits=0, service_coins=0, currency="EUR") wallet_beneficiary = Wallet(user_id=user_beneficiary.id, earned_credits=0, purchased_credits=0, service_coins=0, currency="EUR") session.add_all([wallet_payer, wallet_beneficiary]) await session.commit() print(f" Payer ID: {user_payer.id}, Beneficiary ID: {user_beneficiary.id}") # 2. Stripe szimuláció - manuális feltöltés print("\n2. Stripe szimuláció (10000 PURCHASED)...") wallet_payer.purchased_credits += Decimal('10000.0') await session.commit() await session.refresh(wallet_payer) assert float(wallet_payer.purchased_credits) == 10000.0 print(f" ✅ Payer purchased credits: {wallet_payer.purchased_credits}") # 3. Belső ajándékozás 5000 VOUCHER print("\n3. Belső ajándékozás (5000 VOUCHER)...") payment_intent = await PaymentRouter.create_payment_intent( db=session, payer_id=user_payer.id, net_amount=5000.0, handling_fee=0.0, target_wallet_type=WalletType.VOUCHER, beneficiary_id=user_beneficiary.id, currency="EUR" ) result = await PaymentRouter.process_internal_payment(session, payment_intent.id) print(f" Internal payment result: {result}") await session.refresh(wallet_payer) await session.refresh(wallet_beneficiary) assert float(wallet_payer.purchased_credits) == 5000.0 # Ellenőrizzük a voucher-t stmt = select(ActiveVoucher).where(ActiveVoucher.wallet_id == wallet_beneficiary.id) voucher_result = await session.execute(stmt) vouchers = voucher_result.scalars().all() assert len(vouchers) == 1 voucher = vouchers[0] assert float(voucher.amount) == 5000.0 print(f" ✅ Payer remaining purchased: {wallet_payer.purchased_credits}") print(f" ✅ Beneficiary voucher: {voucher.amount}") # 4. Voucher lejárat szimuláció print("\n4. Voucher lejárat szimuláció (10% fee)...") voucher.expires_at = datetime.utcnow() - timedelta(days=1) await session.commit() stats = await SmartDeduction.process_voucher_expiration(session) print(f" Expiration stats: {stats}") assert abs(stats['fee_collected'] - 500.0) < 0.01 assert abs(stats['rolled_over'] - 4500.0) < 0.01 # Ellenőrizzük az új voucher-t stmt = select(ActiveVoucher).where(ActiveVoucher.wallet_id == wallet_beneficiary.id) new_voucher_result = await session.execute(stmt) new_vouchers = new_voucher_result.scalars().all() assert len(new_vouchers) == 1 new_voucher = new_vouchers[0] assert abs(float(new_voucher.amount) - 4500.0) < 0.01 print(f" ✅ New voucher amount: {new_voucher.amount}") # 5. Double-entry audit print("\n5. Double-entry audit...") total_wallet = Decimal('0') for user in [user_payer, user_beneficiary]: stmt = select(Wallet).where(Wallet.user_id == user.id) w_result = await session.execute(stmt) w = w_result.scalar_one() wallet_sum = w.earned_credits + w.purchased_credits + w.service_coins voucher_stmt = select(func.sum(ActiveVoucher.amount)).where( ActiveVoucher.wallet_id == w.id, ActiveVoucher.expires_at > datetime.utcnow() ) v_result = await session.execute(voucher_stmt) voucher_balance = v_result.scalar() or Decimal('0') total_wallet += wallet_sum + Decimal(str(voucher_balance)) print(f" Total wallet balance: {total_wallet}") # Ledger összegzés stmt = select( FinancialLedger.entry_type, func.sum(FinancialLedger.amount).label('total') ).where( FinancialLedger.user_id.in_([user_payer.id, user_beneficiary.id]) ).group_by(FinancialLedger.entry_type) ledger_result = await session.execute(stmt) credit_total = Decimal('0') debit_total = Decimal('0') for entry_type, amount in ledger_result: if entry_type == LedgerEntryType.CREDIT: credit_total += Decimal(str(amount)) else: debit_total += Decimal(str(amount)) net_ledger = credit_total - debit_total print(f" Net ledger balance: {net_ledger}") # Fee-k levonása fee_stmt = select(func.sum(FinancialLedger.amount)).where( FinancialLedger.reference_type == "VOUCHER_EXPIRY_FEE", FinancialLedger.entry_type == LedgerEntryType.DEBIT ) fee_result = await session.execute(fee_stmt) total_fees = fee_result.scalar() or Decimal('0') adjusted_ledger = net_ledger + total_fees difference = abs(total_wallet - adjusted_ledger) if difference > Decimal('0.01'): raise AssertionError(f"Double-entry mismatch: wallet={total_wallet}, ledger={adjusted_ledger}, diff={difference}") print(f" ✅ Double-entry audit PASS (difference: {difference})") print("\n=== ÖSSZEFOGLALÓ ===") print("Minden teszt sikeresen lefutott!") print("A Pénzügyi Motor logikailag és matematikailag hibátlan.") except Exception as e: print(f"\n❌ TESZT SIKERTELEN: {e}") import traceback traceback.print_exc() raise finally: await session.close() if __name__ == "__main__": asyncio.run(main())