181 lines
7.7 KiB
Python
181 lines
7.7 KiB
Python
#!/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()) |