54 lines
2.4 KiB
Python
54 lines
2.4 KiB
Python
import asyncio, uuid
|
|
from app.database import AsyncSessionLocal
|
|
from app.models.identity.identity import User, UserRole, Wallet
|
|
from app.models.system.audit import SecurityAuditLog, FinancialLedger
|
|
from sqlalchemy import select, text
|
|
|
|
async def test():
|
|
uid = str(uuid.uuid4())[:8]
|
|
email = f'hard_delete_test_{uid}@example.com'
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
# Insert test user
|
|
result = await db.execute(
|
|
text("""
|
|
INSERT INTO identity.users
|
|
(email, hashed_password, role, subscription_plan, preferred_language, region_code, preferred_currency, ui_mode, is_vip, is_active, is_deleted, created_at, scope_level, custom_permissions, alternative_emails, email_history, visual_settings)
|
|
VALUES
|
|
(:email, 'fakehash', 'USER', 'FREE', 'en', 'HU', 'HUF', 'personal', false, false, false, NOW(), 'individual', '{}'::jsonb, '[]'::jsonb, '[]'::jsonb, '{"theme": "default", "primary_color": null, "wall_logo_url": null}'::jsonb)
|
|
RETURNING id
|
|
"""),
|
|
{"email": email}
|
|
)
|
|
await db.commit()
|
|
|
|
result = await db.execute(select(User).where(User.email == email))
|
|
u = result.scalar_one_or_none()
|
|
test_id = u.id
|
|
print(f'Created test user #{test_id}: {u.email}')
|
|
|
|
# Verify no wallet/ledger
|
|
wr = await db.execute(select(Wallet).where(Wallet.user_id == test_id).limit(1))
|
|
lr = await db.execute(select(FinancialLedger).where(FinancialLedger.user_id == test_id).limit(1))
|
|
print(f' Has Wallet: {wr.scalar_one_or_none() is not None}')
|
|
print(f' Has Ledger: {lr.scalar_one_or_none() is not None}')
|
|
|
|
# Simulate the FIXED hard delete: delete first, then audit
|
|
await db.delete(u)
|
|
await db.flush()
|
|
|
|
audit = SecurityAuditLog(
|
|
actor_id=1, target_id=test_id, action='hard_delete_user',
|
|
is_critical=True,
|
|
payload_before={'user_id': test_id, 'email': email, 'role': 'USER'},
|
|
payload_after={'details': f'User #{test_id} permanently deleted by admin #1'},
|
|
)
|
|
db.add(audit)
|
|
await db.commit()
|
|
|
|
check = await db.get(User, test_id)
|
|
print(f' User after delete: {check}')
|
|
print('✅ Hard delete simulation SUCCESSFUL')
|
|
|
|
asyncio.run(test())
|