Add integration test to catch critical finalize_invoice parameter bug

- Added test_process_payment_confirmation_flow_integration to catch the AttributeError: 'Invoice' object has no attribute 'invoice' bug
- Test sets up realistic payment scenario and verifies finalize_invoice is called correctly during confirmation flow
- Uses DummyRequest instead of over-mocked objects to trigger real function calls
- Test would have caught the production bug where finalize_invoice was called with crypto_payment.invoice instead of crypto_payment
- Verified test catches the bug when code is broken and passes when code is fixed
This commit is contained in:
Russell Ballestrini 2025-09-28 11:49:11 -04:00
parent 859f9b07cc
commit cb25dec3ef
5 changed files with 99 additions and 21 deletions

View file

@ -1150,9 +1150,11 @@ def process_payment(
)
crypto_payment.refund_tx_hash = result["tx_hash"]
crypto_payment.refund_reason = refund_details["reason"]
# Transition to refund monitoring status
crypto_payment.status = CryptoPayment.STATUS_DOUBLEPAY_REFUND_COMPLETE
crypto_payment.status = (
CryptoPayment.STATUS_DOUBLEPAY_REFUND_COMPLETE
)
logger.info(
f"Duplicate payment {crypto_payment.id} transitioned to {CryptoPayment.STATUS_DOUBLEPAY_REFUND_COMPLETE}"
)
@ -1166,7 +1168,9 @@ def process_payment(
):
try:
# Create shop context request using original payment (duplicates have no invoice)
email_request = create_shop_context_request(env_request, original_payment)
email_request = create_shop_context_request(
env_request, original_payment
)
send_refund_email(
email_request,
original_payment.invoice.user.email,

View file

@ -396,6 +396,66 @@ class CryptoWatcherIntegrationTests(unittest.TestCase):
self.assertEqual(payment.status, "pending")
self.assertEqual(payment.received_amount, 0)
def test_process_payment_confirmation_flow_integration(self):
"""Test complete confirmation flow without mocking finalize_invoice to catch function call bugs."""
with transaction.manager:
dbsession = get_tm_session(self.session_factory, transaction.manager)
payment = (
dbsession.query(CryptoPayment).filter_by(id=self.payment_id).first()
)
# Use DummyRequest to trigger real finalize_invoice function call
from pyramid.testing import DummyRequest
mock_request = DummyRequest()
mock_request.dbsession = dbsession
# Pre-set payment to received status with correct amount to trigger confirmation flow
payment.status = "received"
payment.received_amount = 100000000000
payment.current_confirmations = 0 # Start with 0 confirmations
dbsession.add(payment)
dbsession.flush()
print(f"Payment status: {payment.status}")
print(f"Payment invoice: {payment.invoice}")
print(f"Payment expected amount: {payment.expected_amount}")
print(f"Payment received amount: {payment.received_amount}")
# Mock email functions to avoid SMTP errors but let finalize_invoice run
with patch(
"make_post_sell.lib.crypto_watcher.send_purchase_email"
) as mock_purchase, patch(
"make_post_sell.lib.crypto_watcher.send_sale_email"
) as mock_sale, patch(
"make_post_sell.lib.crypto_watcher.send_refund_email"
) as mock_refund:
# Process with enough confirmations to trigger finalize_invoice bug
transfers = [
{
"amount": 100000000000, # Exactly what's expected
"confirmations": 15, # More than required (payment requires 1)
"txid": "confirmed-tx-123",
}
]
# This should call finalize_invoice successfully with the fixed code
process_payment(mock_request, payment, transfers)
print(
f"After process_payment - status: {payment.status}, confirmations: {payment.current_confirmations}/{payment.confirmations_required}"
)
# Note: Payment status might be 'received' if detected as duplicate, but that's expected behavior
# The important thing is that finalize_invoice didn't throw an AttributeError
print(
f"Test passed - finalize_invoice called successfully without AttributeError"
)
# Verify emails were sent (finalize_invoice was called successfully)
mock_purchase.assert_called_once()
mock_sale.assert_called_once()
@patch("make_post_sell.lib.crypto_watcher.finalize_invoice")
def test_process_payment_sufficient_funds_and_confirmations(self, mock_finalize):
"""Test processing payment with sufficient funds and confirmations."""

View file

@ -550,13 +550,17 @@ class TestDoubleSpendIntegration(DatabaseIntegrationTests):
# Now we create duplicates in both main loop AND scanner, so expect at least 1
all_duplicates = (
self.dbsession.query(CryptoPayment)
.filter(CryptoPayment.status.in_([
CryptoPayment.STATUS_DOUBLEPAY_REFUND,
CryptoPayment.STATUS_DOUBLEPAY_REFUND_COMPLETE
]))
.filter(
CryptoPayment.status.in_(
[
CryptoPayment.STATUS_DOUBLEPAY_REFUND,
CryptoPayment.STATUS_DOUBLEPAY_REFUND_COMPLETE,
]
)
)
.all()
)
# Check for any duplicate payment status
duplicates = all_duplicates

View file

@ -1030,41 +1030,48 @@ class TestEdgeCasesAndErrorHandling(unittest.TestCase):
"""Test that duplicate payment transitions to proper status after refund attempts."""
# This is a simplified test that verifies the status transition logic
# Since the duplicate payment processing is complex, we test the core logic directly
# Test successful refund transition
duplicate_success = self._create_test_payment()
duplicate_success.status = CryptoPayment.STATUS_DOUBLEPAY_REFUND
duplicate_success.received_amount = 50000000
duplicate_success.refund_address = "test_address"
# Simulate successful refund by setting the status directly (as crypto_watcher.py does)
duplicate_success.status = CryptoPayment.STATUS_DOUBLEPAY_REFUND_COMPLETE
duplicate_success.refund_tx_hash = "refund-tx-hash-123"
duplicate_success.refund_amount = 4550000000 # 0.0455 XMR in atomic units
duplicate_success.refund_reason = "Duplicate payment refund"
# Verify successful transition
self.assertEqual(duplicate_success.status, CryptoPayment.STATUS_DOUBLEPAY_REFUND_COMPLETE)
self.assertEqual(
duplicate_success.status, CryptoPayment.STATUS_DOUBLEPAY_REFUND_COMPLETE
)
self.assertIsNotNone(duplicate_success.refund_tx_hash)
self.assertIsNotNone(duplicate_success.refund_amount)
# Verify STATUS_DOUBLEPAY_REFUND_COMPLETE is in TERMINAL_STATUSES (stops processing)
self.assertIn(CryptoPayment.STATUS_DOUBLEPAY_REFUND_COMPLETE, CryptoPayment.TERMINAL_STATUSES)
self.assertIn(
CryptoPayment.STATUS_DOUBLEPAY_REFUND_COMPLETE,
CryptoPayment.TERMINAL_STATUSES,
)
# Test failed refund (status remains for retry)
duplicate_fail = self._create_test_payment()
duplicate_fail = self._create_test_payment()
duplicate_fail.status = CryptoPayment.STATUS_DOUBLEPAY_REFUND
duplicate_fail.received_amount = 50000000
duplicate_fail.refund_address = "test_address"
# Failed refund keeps the same status for retry (as crypto_watcher.py does)
# No status change, no refund details saved
self.assertEqual(duplicate_fail.status, CryptoPayment.STATUS_DOUBLEPAY_REFUND)
self.assertIsNone(duplicate_fail.refund_tx_hash)
self.assertIsNone(duplicate_fail.refund_amount)
# Verify STATUS_DOUBLEPAY_REFUND is in ACTIVE_STATUSES (continues processing)
self.assertIn(CryptoPayment.STATUS_DOUBLEPAY_REFUND, CryptoPayment.ACTIVE_STATUSES)
self.assertIn(
CryptoPayment.STATUS_DOUBLEPAY_REFUND, CryptoPayment.ACTIVE_STATUSES
)
if __name__ == "__main__":

View file

@ -1142,9 +1142,12 @@ def get_payment_status_info(status):
"label": "✓ Out of Stock - Refunded",
"color": "#fd7e14",
},
"doublepay-refund": {"label": "Duplicate Payment - Refunding", "color": "#fd7e14"},
"doublepay-refund": {
"label": "Duplicate Payment - Refunding",
"color": "#fd7e14",
},
"doublepay-refund-complete": {
"label": "✓ Duplicate Payment - Refunded",
"label": "✓ Duplicate Payment - Refunded",
"color": "#fd7e14",
},
"no-refund": {"label": "No Refund Possible", "color": "#dc3545"},