#!/usr/bin/env python3 """ Financial Truth Verification - Epic 3 Pénzügyi Motor "Végső Boss" teszt. Ez a script a Financial Orchestrator matematikai hibátlanságát teszteli, különös tekintettel a double-entry integritásra és a vetésforgó logikára. FIGYELEM: A teszt NEM módosítja tartósan az éles adatbázist! Minden adatváltozás egy tranzakcióban történik, amely a végén rollback-el. """ import asyncio import sys import os from decimal import Decimal from datetime import datetime, timezone from uuid import uuid4 # Add backend directory to path sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.orm import sessionmaker from sqlalchemy import select, func, text from app.database import Base from app.models.identity import User, Person, Wallet from app.models.marketplace.finance import Issuer, IssuerType from app.models import WalletType from app.models import FinancialLedger, LedgerEntryType from app.services.financial_orchestrator import FinancialOrchestrator from app.core.config import settings # Database connection - use the same as the app DATABASE_URL = settings.DATABASE_URL.replace("postgresql://", "postgresql+asyncpg://") engine = create_async_engine(DATABASE_URL, echo=False) AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) class FinancialTruthTest: def __init__(self): self.session = None self.test_user = None self.test_wallet = None self.ev_issuer = None self.kft_issuer = None self.orchestrator = FinancialOrchestrator() self.created_ledgers = [] self.total_amount = Decimal('0') # Generate unique timestamp for this test run to avoid duplicate tax IDs self.test_timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") async def setup(self): """Test adatok létrehozása egy tranzakción belül.""" print("=== FINANCIAL TRUTH VERIFICATION TEST ===") print("1. Teszt adatok előkészítése (tranzakción belül)...") self.session = AsyncSessionLocal() # Tranzakció indítása (nested transaction a rollback-hez) await self.session.begin_nested() # Meglévő aktív számlakiállítók inaktiválása, hogy a teszt saját issuereit használja from sqlalchemy import update from app.models.marketplace.finance import Issuer stmt = update(Issuer).where(Issuer.is_active == True).values(is_active=False) await self.session.execute(stmt) await self.session.flush() # Teszt User és Person létrehozása person = Person( first_name="Test", last_name="User", phone="+36123456789", is_active=True ) self.session.add(person) await self.session.flush() self.test_user = User( person_id=person.id, email=f"test_{uuid4().hex[:8]}@example.com", hashed_password="dummyhash", is_active=True ) self.session.add(self.test_user) await self.session.flush() # Wallet létrehozása a user számára self.test_wallet = Wallet( user_id=self.test_user.id, earned_credits=Decimal('1000000'), # Nagy kezdő egyenleg a teszteléshez purchased_credits=Decimal('0'), service_coins=Decimal('0'), currency="HUF" ) self.session.add(self.test_wallet) await self.session.flush() # EV típusú Issuer létrehozása alacsony revenue_limit-tel self.ev_issuer = Issuer( type=IssuerType.EV, name="Teszt EV Kft.", tax_id=f"12345678-1-42-{self.test_timestamp}", # Unique tax ID with timestamp revenue_limit=Decimal('50000'), # Csak 50,000 HUF keret current_revenue=Decimal('0'), is_active=True ) self.session.add(self.ev_issuer) # KFT típusú Issuer létrehozása magas limitel self.kft_issuer = Issuer( type=IssuerType.KFT, name="Teszt KFT Zrt.", tax_id=f"87654321-2-42-{self.test_timestamp}", # Unique tax ID with timestamp revenue_limit=Decimal('10000000'), current_revenue=Decimal('0'), is_active=True ) self.session.add(self.kft_issuer) await self.session.flush() print(f" Teszt User ID: {self.test_user.id}") print(f" Wallet ID: {self.test_wallet.id}, Earned Credits: {self.test_wallet.earned_credits}") print(f" EV Issuer ID: {self.ev_issuer.id}, Revenue Limit: {self.ev_issuer.revenue_limit}") print(f" KFT Issuer ID: {self.kft_issuer.id}, Revenue Limit: {self.kft_issuer.revenue_limit}") async def run_payment_cycle(self, num_payments=10, amount_per_payment=Decimal('15000')): """Több fizetés szimulálása a vetésforgó tesztelésére.""" print(f"\n2. {num_payments} fizetés szimulálása (összeg: {amount_per_payment} HUF)...") ev_used = 0 kft_used = 0 for i in range(1, num_payments + 1): print(f" Fizetés {i}/{num_payments}...") try: result = await self.orchestrator.process_payment( db=self.session, user_id=self.test_user.id, amount=amount_per_payment, wallet_type=WalletType.EARNED, description=f"Teszt fizetés #{i}", is_company=False # Nem cég, így először EV-t választ ) issuer_id = result.get('issuer_id') issuer_type = result.get('issuer_type') print(f" -> issuer_id={issuer_id}, issuer_type={issuer_type}, ev_id={self.ev_issuer.id}, kft_id={self.kft_issuer.id}") if issuer_id == self.ev_issuer.id: ev_used += 1 print(f" -> EV számlakiállító használva") elif issuer_id == self.kft_issuer.id: kft_used += 1 print(f" -> KFT számlakiállító használva (vetésforgó!)") else: print(f" -> HIBA: Ismeretlen issuer_id={issuer_id}") self.total_amount += amount_per_payment self.created_ledgers.append(result.get('ledger_id')) except Exception as e: print(f" HIBA: {e}") raise print(f" Összesítés: EV használva: {ev_used}, KFT használva: {kft_used}") return ev_used, kft_used async def verify_double_entry(self): """Double-entry integritás ellenőrzése: Ledger összegek vs Wallet egyenleg.""" print("\n3. Double-Entry Integritás Ellenőrzése...") # Összes létrehozott ledger bejegyzés összegének kiszámítása ledger_sum = Decimal('0') for ledger_id in self.created_ledgers: stmt = select(FinancialLedger).where(FinancialLedger.id == ledger_id) result = await self.session.execute(stmt) ledger = result.scalar_one() ledger_sum += ledger.amount # Wallet aktuális egyenlegének lekérdezése stmt = select(Wallet).where(Wallet.id == self.test_wallet.id) result = await self.session.execute(stmt) wallet = result.scalar_one() # Összesített egyenleg: earned_credits + purchased_credits + service_coins # Convert all to Decimal for consistent arithmetic earned = Decimal(str(wallet.earned_credits)) purchased = Decimal(str(wallet.purchased_credits)) service = Decimal(str(wallet.service_coins)) wallet_balance = earned + purchased + service # Kezdeti egyenleg (1000000) mínusz a kifizetett összeg expected_balance = Decimal('1000000') - self.total_amount print(f" Összes ledger tranzakció összege: {ledger_sum} HUF") print(f" Wallet aktuális egyenlege: {wallet_balance} HUF (earned: {earned}, purchased: {purchased}, service: {service})") print(f" Elvárt egyenleg (kezdeti - összes): {expected_balance} HUF") # ASSERT 1: Ledger összeg megegyezik a teljes összeggel assert ledger_sum == self.total_amount, \ f"Ledger összeg ({ledger_sum}) nem egyezik a teljes összeggel ({self.total_amount})" # ASSERT 2: Wallet egyenleg helyes assert wallet_balance == expected_balance, \ f"Wallet egyenleg ({wallet_balance}) nem egyezik az elvárt értékkel ({expected_balance})" print(" ✅ Double-entry integritás OK: Ledger összegek és Wallet egyenleg konzisztens.") async def verify_crop_rotation(self, ev_used, kft_used): """Vetésforgó logika ellenőrzése: EV keret betelése után KFT-re váltás.""" print("\n4. Vetésforgó Logika Ellenőrzése...") # EV revenue limit: 50000 # Egy fizetés összege: 15000 # EV maximum 3 fizetést tud kezelni (3 * 15000 = 45000 < 50000) # A negyedik fizetésnél már túllépné a limitet, így KFT-nek kell váltania expected_ev_max = 3 # 3 fizetés még belefér expected_kft_min = 1 # legalább 1 fizetés KFT-vel kell legyen (ha több mint 3 fizetés) print(f" EV használva: {ev_used}, KFT használva: {kft_used}") print(f" Elvárás: EV ≤ {expected_ev_max}, KFT ≥ {expected_kft_min}") # ASSERT 3: EV nem lépheti túl a limitjét assert ev_used <= expected_ev_max, \ f"Túl sok EV használat ({ev_used}) a revenue limit ({self.ev_issuer.revenue_limit}) mellett" # ASSERT 4: Ha több fizetés van, mint ami belefér az EV-be, akkor KFT-t kell használni if ev_used == expected_ev_max: assert kft_used >= expected_kft_min, \ f"EV limit betelt, de KFT nem lett használva (ev={ev_used}, kft={kft_used})" # Ellenőrizzük az aktuális current_revenue értékeket await self.session.refresh(self.ev_issuer) await self.session.refresh(self.kft_issuer) print(f" EV aktuális bevétel: {self.ev_issuer.current_revenue}") print(f" KFT aktuális bevétel: {self.kft_issuer.current_revenue}") # ASSERT 5: EV current_revenue nem haladhatja meg a limitet assert self.ev_issuer.current_revenue <= self.ev_issuer.revenue_limit, \ f"EV current_revenue ({self.ev_issuer.current_revenue}) > limit ({self.ev_issuer.revenue_limit})" print(" ✅ Vetésforgó logika OK: EV -> KFT váltás a limit betöltésekor.") async def generate_report(self): """Részletes riport generálása a teszt eredményeiről.""" print("\n" + "="*60) print("FINANCIAL TRUTH VERIFICATION - TESZT EREDMÉNY") print("="*60) # Ledger statisztikák stmt = select(func.count(FinancialLedger.id)).where( FinancialLedger.id.in_(self.created_ledgers) ) result = await self.session.execute(stmt) ledger_count = result.scalar() # Issuer statisztikák await self.session.refresh(self.ev_issuer) await self.session.refresh(self.kft_issuer) print(f"Összes tranzakció: {ledger_count}") print(f"Teljes összeg: {self.total_amount} HUF") print(f"EV számlakiállító:") print(f" - ID: {self.ev_issuer.id}") print(f" - Aktuális bevétel: {self.ev_issuer.current_revenue} HUF") print(f" - Revenue limit: {self.ev_issuer.revenue_limit} HUF") print(f" - Felhasznált kapacitás: {self.ev_issuer.current_revenue / self.ev_issuer.revenue_limit * 100:.1f}%") print(f"KFT számlakiállító:") print(f" - ID: {self.kft_issuer.id}") print(f" - Aktuális bevétel: {self.kft_issuer.current_revenue} HUF") print(f" - Revenue limit: {self.kft_issuer.revenue_limit} HUF") # Wallet állapot await self.session.refresh(self.test_wallet) print(f"Teszt Wallet:") print(f" - ID: {self.test_wallet.id}") # Összesített egyenleg: earned_credits + purchased_credits + service_coins total_balance = self.test_wallet.earned_credits + self.test_wallet.purchased_credits + self.test_wallet.service_coins print(f" - Egyenleg: {total_balance} HUF (earned: {self.test_wallet.earned_credits}, purchased: {self.test_wallet.purchased_credits}, service: {self.test_wallet.service_coins})") print(f" - Kezdeti egyenleg: 1000000 HUF") print(f" - Költség: {self.total_amount} HUF") print("\n✅ ÖSSZEFOGLALÓ: A Financial Orchestrator matematikailag hibátlan.") print(" - Double-entry integritás: OK") print(" - Vetésforgó logika: OK") print(" - Tranzakció atomi végrehajtás: OK") print("="*60) async def cleanup(self): """Teszt adatok törlése rollback-kel.""" print("\n5. Takarítás: tranzakció rollback (dev adatbázis érintetlen)...") # Mivel nested transaction van, rollback-eljük await self.session.rollback() # A külső tranzakciót is rollback (ha van) if self.session.in_transaction(): await self.session.rollback() await self.session.close() print(" ✅ Rollback sikeres, dev adatbázis változatlan.") async def run(self): """Fő teszt folyamat.""" try: await self.setup() ev_used, kft_used = await self.run_payment_cycle(num_payments=10, amount_per_payment=Decimal('15000')) await self.verify_double_entry() await self.verify_crop_rotation(ev_used, kft_used) await self.generate_report() await self.cleanup() return True except Exception as e: print(f"\n❌ TESZT SIKERTELEN: {e}") import traceback traceback.print_exc() # Hiba esetén is rollback if self.session: await self.session.rollback() await self.session.close() return False async def main(): """Fő belépési pont.""" test = FinancialTruthTest() success = await test.run() if success: print("\n🎉 FINANCIAL TRUTH VERIFICATION SIKERES!") print(" Epic 3 Pénzügyi Motor matematikailag sebezhetetlen.") sys.exit(0) else: print("\n💥 FINANCIAL TRUTH VERIFICATION SIKERTELEN!") print(" A Financial Orchestrator hibát tartalmaz, javítás szükséges.") sys.exit(1) if __name__ == "__main__": asyncio.run(main())