test_finance_service_defaults.py 3.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  1. """Unit tests for finance defaults applied during user creation/update."""
  2. import pytest
  3. from sqlalchemy import select
  4. from backend.app.models.finance import CostCenter, CostCenterMember, UserWallet
  5. from backend.app.models.settings import Settings
  6. from backend.app.models.user import User
  7. from backend.app.services.finance_defaults import ensure_user_finance_defaults
  8. class TestFinanceDefaults:
  9. @pytest.mark.asyncio
  10. async def test_creates_wallet_private_center_and_membership(self, db_session):
  11. db_session.add(Settings(key="currency", value="USD"))
  12. user = User(username="alice", role="user", is_active=True)
  13. db_session.add(user)
  14. await db_session.commit()
  15. await db_session.refresh(user)
  16. changed = await ensure_user_finance_defaults(db_session, user)
  17. await db_session.commit()
  18. assert changed is True
  19. wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
  20. assert wallet is not None
  21. assert wallet.balance == 0.0
  22. assert wallet.currency == "USD"
  23. center = await db_session.scalar(
  24. select(CostCenter).where(CostCenter.owner_user_id == user.id, CostCenter.is_private.is_(True))
  25. )
  26. assert center is not None
  27. assert center.name == "alice"
  28. membership = await db_session.scalar(
  29. select(CostCenterMember).where(
  30. CostCenterMember.cost_center_id == center.id,
  31. CostCenterMember.user_id == user.id,
  32. )
  33. )
  34. assert membership is not None
  35. assert membership.can_print is True
  36. @pytest.mark.asyncio
  37. async def test_updates_private_center_name_and_is_idempotent(self, db_session):
  38. user = User(username="bob", role="user", is_active=True)
  39. db_session.add(user)
  40. await db_session.commit()
  41. await db_session.refresh(user)
  42. initial_changed = await ensure_user_finance_defaults(db_session, user)
  43. await db_session.commit()
  44. assert initial_changed is True
  45. user.username = "bobby"
  46. renamed_changed = await ensure_user_finance_defaults(db_session, user)
  47. await db_session.commit()
  48. assert renamed_changed is True
  49. center = await db_session.scalar(
  50. select(CostCenter).where(CostCenter.owner_user_id == user.id, CostCenter.is_private.is_(True))
  51. )
  52. assert center is not None
  53. assert center.name == "bobby"
  54. idempotent_changed = await ensure_user_finance_defaults(db_session, user)
  55. assert idempotent_changed is False
  56. @pytest.mark.asyncio
  57. async def test_reactivates_existing_private_center(self, db_session):
  58. user = User(username="carol", role="user", is_active=True)
  59. db_session.add(user)
  60. await db_session.flush()
  61. center = CostCenter(
  62. name=user.username,
  63. is_active=False,
  64. is_private=True,
  65. owner_user_id=user.id,
  66. )
  67. db_session.add(center)
  68. await db_session.commit()
  69. changed = await ensure_user_finance_defaults(db_session, user)
  70. await db_session.commit()
  71. assert changed is True
  72. assert center.is_active is True