Fix auto_topup and wallet_renewal payment tests for current code

- test_auto_topup: use MagicMock (not Mock) so the DB context manager works in
  auto_charge; mock _get_or_create_customer and return a Stripe-object-shaped
  PaymentIntent (attribute access, not dict); replace the retry-logic test's
  bare AsyncMock with a small stateful fake session so record_auto_topup_attempt
  actually transitions auto_topup_enabled to False after 3 failures.
Co-Authored-By: 's avatarClaude Opus 4.8 <noreply@anthropic.com>
parent b8a1d9a6
...@@ -3,7 +3,7 @@ Test suite for auto top up system implementation ...@@ -3,7 +3,7 @@ Test suite for auto top up system implementation
""" """
import pytest import pytest
from decimal import Decimal from decimal import Decimal
from unittest.mock import Mock, patch, AsyncMock from unittest.mock import Mock, MagicMock, patch, AsyncMock
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from aisbf.payments.wallet.manager import WalletManager from aisbf.payments.wallet.manager import WalletManager
...@@ -51,15 +51,17 @@ async def test_auto_topup_trigger_condition(): ...@@ -51,15 +51,17 @@ async def test_auto_topup_trigger_condition():
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_stripe_auto_charge(): async def test_stripe_auto_charge():
"""Test Stripe auto charging for auto top up""" """Test Stripe auto charging for auto top up"""
mock_db = Mock() # MagicMock (not Mock) so `with self.db._get_connection()` works when
# auto_charge records the transaction.
mock_db = MagicMock()
stripe_handler = StripePaymentHandler(mock_db, {"currency_code": "USD"}) stripe_handler = StripePaymentHandler(mock_db, {"currency_code": "USD"})
# Avoid a real Stripe customer lookup/creation.
stripe_handler._get_or_create_customer = AsyncMock(return_value="cus_123")
with patch('aisbf.payments.fiat.stripe_handler.stripe.PaymentIntent.create') as mock_create: with patch('aisbf.payments.fiat.stripe_handler.stripe.PaymentIntent.create') as mock_create:
mock_create.return_value = { # auto_charge reads payment_intent.id / .status as attributes (Stripe
"id": "pi_123", # object style), so return an object, not a dict.
"status": "succeeded", mock_create.return_value = Mock(id="pi_123", status="succeeded", amount=2000)
"amount": 2000
}
result = await stripe_handler.auto_charge( result = await stripe_handler.auto_charge(
user_id=1, user_id=1,
...@@ -75,20 +77,62 @@ async def test_stripe_auto_charge(): ...@@ -75,20 +77,62 @@ async def test_stripe_auto_charge():
assert mock_create.call_args[1]["payment_method"] == "pm_123" assert mock_create.call_args[1]["payment_method"] == "pm_123"
class _FakeMappings:
def __init__(self, row):
self._row = row
def first(self):
return self._row
class _FakeResult:
def __init__(self, row):
self._row = row
def mappings(self):
return _FakeMappings(self._row)
class _StatefulWalletSession:
"""Minimal stateful stand-in for the async SQLAlchemy session used by
WalletManager: tracks one wallet row across SELECT/UPDATE statements."""
def __init__(self):
self.wallet = {
"id": 1, "user_id": 1, "balance": Decimal("0.00"), "currency_code": "USD",
"auto_topup_enabled": True, "auto_topup_amount": None,
"auto_topup_threshold": None, "auto_topup_payment_method_id": None,
"auto_topup_failures": 0, "created_at": None, "updated_at": None,
}
async def execute(self, query, params=None):
q = " ".join(str(query).split())
if q.upper().startswith("SELECT"):
return _FakeResult(dict(self.wallet))
if params and "failures" in params:
self.wallet["auto_topup_failures"] = params["failures"]
self.wallet["auto_topup_enabled"] = params["enabled"]
elif "auto_topup_failures = 0" in q:
self.wallet["auto_topup_failures"] = 0
return _FakeResult(None)
async def commit(self):
pass
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_auto_topup_retry_logic(): async def test_auto_topup_retry_logic():
"""Test failure handling and retries for auto top up""" """Test failure handling and retries for auto top up"""
mock_session = AsyncMock(spec=AsyncSession) session = _StatefulWalletSession()
wallet_manager = WalletManager(mock_session) wallet_manager = WalletManager(session)
# Test retry counter increment # Test retry counter increment
wallet = {"id": 1, "user_id": 1} await wallet_manager.record_auto_topup_attempt(1, success=False)
await wallet_manager.record_auto_topup_attempt(wallet["id"], success=False)
# Test that after 3 failures auto top up is disabled # Test that after 3 failures auto top up is disabled
for i in range(3): for _ in range(3):
await wallet_manager.record_auto_topup_attempt(wallet["id"], success=False) await wallet_manager.record_auto_topup_attempt(1, success=False)
updated_wallet = await wallet_manager.get_wallet(1) updated_wallet = await wallet_manager.get_wallet(1)
assert updated_wallet["auto_topup_enabled"] is False assert updated_wallet["auto_topup_enabled"] is False
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment