Epic 3: Economy & Billing Engine (Pénzügyi Motor)
This commit is contained in:
Roo
181
verify_financial_truth_simple.py
Normal file
181
verify_financial_truth_simple.py
Normal file
@@ -0,0 +1,181 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user