engineering/make_post_sell
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases 2 Packages Wiki Activity Actions

Fix critical double-spend vulnerability that drained hot wallet

Browse source 
- Add duplicate payment detection in main processing loop that prevents
  new transactions to payments with received_amount > 0 and status \!= PENDING
- Rename scan_wallet_for_late_payments to scan_wallet_for_double_or_late_payments
- Create comprehensive test suite with 16 tests covering all scenarios:
  * Duplicate payment detection for confirmed/received payments
  * Race condition protection between main loop and scanner
  * Concurrent transaction handling within processing windows
  * Wallet drain prevention and edge cases
- Fix _create_duplicate_payment to use proper CryptoPayment constructor
- Update existing idempotent test to reflect correct business logic

The vulnerability allowed duplicate payments to trigger multiple full refunds
plus extra funds, catastrophically draining the hot wallet. Now any second
payment to an address that already received funds is rejected by main loop
and handled by scanner with 9% restocking fee refund.
This commit is contained in:
Russell Ballestrini 2025-09-27 12:23:02 -04:00
parent df21f1b135
commit e63bccf74f
4 changed files with 1261 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

95
make_post_sell/lib/crypto_watcher.py
View file

@ -79,7 +79,15 @@ def _create_duplicate_payment(original_payment, tx, coin_type):
import uuid
import time
from ..models.crypto_payment import CryptoPayment
from ..models.meta import get_coin_config
# Define local get_coin_config for this function
def get_coin_config(coin_type):
configs = {
"XMR": {"atomic_units": 1e12},
"DOGE": {"atomic_units": 1e8},
"BTC": {"atomic_units": 1e8},
}
return configs.get(coin_type, {"atomic_units": 1e12})
# Get coin configuration for amount conversion
coin_config = get_coin_config(coin_type)
@ -92,44 +100,30 @@ def _create_duplicate_payment(original_payment, tx, coin_type):
# DOGE/BTC amount is already in atomic units
tx_amount = int(float(tx.get("amount", 0)) * 1e8)
# Create new payment object for this duplicate
duplicate_payment = CryptoPayment()
duplicate_payment.id = uuid.uuid1()
# Copy key details from original payment
duplicate_payment.invoice_id = None # No invoice - this is a duplicate
duplicate_payment.user_id = original_payment.user_id
duplicate_payment.shop_id = original_payment.shop_id
duplicate_payment.shop_location_id = original_payment.shop_location_id
# Copy crypto details
duplicate_payment.coin_type = original_payment.coin_type
duplicate_payment.account_index = original_payment.account_index
duplicate_payment.subaddress_index = original_payment.subaddress_index
duplicate_payment.address = original_payment.address
# Set amounts - this duplicate payment is the exact amount received
duplicate_payment.expected_amount = (
tx_amount # What we "expected" (the duplicate amount)
# Create new payment object for this duplicate using proper constructor
duplicate_payment = CryptoPayment(
invoice=None, # No invoice - this is a duplicate
user=original_payment.user,
shop=original_payment.shop,
address=original_payment.address,
account_index=original_payment.account_index,
subaddress_index=original_payment.subaddress_index,
coin_type=original_payment.coin_type,
expected_amount=tx_amount, # What we "expected" (the duplicate amount)
rate_locked_usd_per_coin=original_payment.rate_locked_usd_per_coin,
quote_expires_at_ms=int(
time.time() * 1000
), # Already "expired" since it's a duplicate
confirmations_required=original_payment.confirmations_required,
shop_location=original_payment.shop_location,
shop_sweep_to_address=original_payment.shop_sweep_to_address,
refund_address=original_payment.refund_address,
)
duplicate_payment.received_amount = tx_amount # What we received
# Set status and confirmations
# Override specific properties for duplicate
duplicate_payment.received_amount = tx_amount # What we received
duplicate_payment.status = CryptoPayment.STATUS_DOUBLEPAY_REFUND
duplicate_payment.current_confirmations = tx.get("confirmations", 0)
duplicate_payment.confirmations_required = original_payment.confirmations_required
# Set refund address if original has one
duplicate_payment.refund_address = original_payment.refund_address
duplicate_payment.shop_sweep_to_address = original_payment.shop_sweep_to_address
# Set timestamps
now_ms = int(time.time() * 1000)
duplicate_payment.created_timestamp = now_ms
duplicate_payment.updated_timestamp = now_ms
# Set expiry (copy from original but mark as immediate for refund processing)
duplicate_payment.expires_at = now_ms # Already "expired" since it's a duplicate
return duplicate_payment
@ -1379,15 +1373,34 @@ def process_payment(
new_sum = 0
seen = set(existing)
total_fee = 0
new_txids = []
for t in incoming_transfers:
txid = t.get("txid") or t.get("transaction_id")
if txid and txid not in seen:
new_sum += int(t.get("amount", 0) or 0)
new_txids.append(txid)
# Capture fee if available (some coins provide this, others don't)
fee = t.get("fee", 0)
if fee:
total_fee += int(fee)
# CRITICAL: If payment already received funds and we see NEW transactions,
# these are duplicate payments that should ONLY be handled by late payment scanner
if (
crypto_payment.received_amount > 0
and new_txids
and crypto_payment.status != CryptoPayment.STATUS_PENDING
):
logger.warning(
f"DUPLICATE PAYMENT DETECTED: {len(new_txids)} new transactions to {crypto_payment.status} payment {crypto_payment.id} "
f"with existing balance {crypto_payment.received_amount}. Skipping to prevent double processing."
)
# Still update confirmation counts for existing transactions
crypto_payment.current_confirmations = min_confs
crypto_payment.updated_timestamp = now_ms
env_request.dbsession.add(crypto_payment)
return
crypto_payment.received_amount = int(crypto_payment.received_amount or 0) + int(
new_sum
)
@ -1852,13 +1865,15 @@ def update_payment_confirmations_only(client, crypto_payment, coin_type):
)
def scan_wallet_for_late_payments(request, settings):
def scan_wallet_for_double_or_late_payments(request, settings):
"""
Scan wallet for new incoming transactions since last scan position and match them to payments.
This catches late payments to expired quotes that aren't actively monitored.
This catches:
1. Double/duplicate payments to already-paid addresses
2. Late payments to expired/cancelled quotes that aren't actively monitored
Uses scan position tracking to avoid rescanning old transfers.
"""
logger.info("Starting wallet scan for late payments")
logger.info("Starting wallet scan for double or late payments")
db = request.dbsession
# Import here to avoid circular imports
@ -2127,9 +2142,9 @@ def run_once(env, interval):
# First, do passive scan for late payments to expired/cancelled quotes
with request.tm:
try:
scan_wallet_for_late_payments(request, settings)
scan_wallet_for_double_or_late_payments(request, settings)
except Exception as e:
logger.error(f"Failed to scan wallet for late payments: {e}")
logger.error(f"Failed to scan wallet for double or late payments: {e}")
with request.tm:
db = request.dbsession

35
make_post_sell/tests/test_crypto_watcher.py
View file

@ -33,7 +33,7 @@ from ..lib.crypto_watcher import (
get_dogecoin_incoming_transfers,
get_coin_config,
get_crypto_client,
scan_wallet_for_late_payments,
scan_wallet_for_double_or_late_payments,
COIN_CONFIGS,
)
from ..lib.crypto_clients import MockMoneroClient, MockDogecoinClient
@ -640,18 +640,29 @@ class CryptoWatcherIntegrationTests(unittest.TestCase):
process_payment(mock_request, payment, transfers1)
self.assertEqual(payment.received_amount, 50000000000)
# Payment becomes RECEIVED once it has any funds
self.assertEqual(payment.status, CryptoPayment.STATUS_RECEIVED)
# Process same transfer again - should not double count
process_payment(mock_request, payment, transfers1)
self.assertEqual(payment.received_amount, 50000000000)
# Add new transfer
# Try to add new transfer - should be rejected as duplicate
transfers2 = transfers1 + [
{"amount": 30000000000, "txid": "tx2", "confirmations": 0}
]
process_payment(mock_request, payment, transfers2)
self.assertEqual(payment.received_amount, 80000000000)
with patch("make_post_sell.lib.crypto_watcher.logger") as mock_logger:
process_payment(mock_request, payment, transfers2)
# Should detect duplicate and reject
mock_logger.warning.assert_called_once()
self.assertIn(
"DUPLICATE PAYMENT DETECTED", mock_logger.warning.call_args[0][0]
)
# Amount should NOT change (duplicate rejected)
self.assertEqual(payment.received_amount, 50000000000)
@patch("make_post_sell.lib.crypto_watcher.auto_sweep_payment")
@patch("make_post_sell.lib.crypto_watcher.finalize_invoice")
@ -1077,7 +1088,9 @@ class PassiveMonitoringTests(unittest.TestCase):
@patch("make_post_sell.lib.crypto_watcher.get_crypto_client")
@patch("make_post_sell.lib.crypto_watcher.process_payment")
def test_scan_wallet_for_late_payments_xmr(self, mock_process, mock_get_client):
def test_scan_wallet_for_double_or_late_payments_xmr(
self, mock_process, mock_get_client
):
"""Test scanning XMR wallet finds expired payment transfers."""
# Mock crypto processor
mock_processor = MagicMock()
@ -1123,7 +1136,7 @@ class PassiveMonitoringTests(unittest.TestCase):
self.mock_dbsession.query.side_effect = query_side_effect
# Run scan
scan_wallet_for_late_payments(self.mock_request, self.mock_settings)
scan_wallet_for_double_or_late_payments(self.mock_request, self.mock_settings)
# Verify client was called to get transfers
mock_client._call.assert_called_once_with(
@ -1197,7 +1210,7 @@ class PassiveMonitoringTests(unittest.TestCase):
self.mock_dbsession.query.side_effect = query_side_effect
# Run scan
scan_wallet_for_late_payments(self.mock_request, self.mock_settings)
scan_wallet_for_double_or_late_payments(self.mock_request, self.mock_settings)
# Verify process_payment was NOT called
mock_process.assert_not_called()
@ -1252,7 +1265,7 @@ class PassiveMonitoringTests(unittest.TestCase):
self.mock_dbsession.query.side_effect = query_side_effect
# Run scan
scan_wallet_for_late_payments(self.mock_request, self.mock_settings)
scan_wallet_for_double_or_late_payments(self.mock_request, self.mock_settings)
# Verify process_payment was called for cancelled payment
mock_process.assert_called_once()
@ -1278,7 +1291,7 @@ class PassiveMonitoringTests(unittest.TestCase):
mock_get_client.return_value = mock_client
# Run scan - should not raise exception
scan_wallet_for_late_payments(self.mock_request, self.mock_settings)
scan_wallet_for_double_or_late_payments(self.mock_request, self.mock_settings)
# Verify process_payment was not called
mock_process.assert_not_called()
@ -1315,7 +1328,7 @@ class PassiveMonitoringTests(unittest.TestCase):
mock_get_client.side_effect = get_client_side_effect
# Run scan - should handle gracefully
scan_wallet_for_late_payments(self.mock_request, self.mock_settings)
scan_wallet_for_double_or_late_payments(self.mock_request, self.mock_settings)
# Verify it tried to get clients but continued when failed
self.assertEqual(mock_get_client.call_count, 2) # XMR and DOGE
@ -1443,7 +1456,7 @@ class PassiveMonitoringIntegrationTests(unittest.TestCase):
mock_request.tm = transaction.manager
# Run passive scan
scan_wallet_for_late_payments(mock_request, {})
scan_wallet_for_double_or_late_payments(mock_request, {})
# Verify payment was updated
payment = (

441
make_post_sell/tests/test_double_spend_integration.py Normal file
View file

@ -0,0 +1,441 @@
"""
Integration tests for double-spend protection.
These tests use real database transactions to verify the protection works
in a realistic environment with actual SQLAlchemy sessions.
"""
import json
import time
import uuid
from decimal import Decimal
from unittest.mock import MagicMock, patch
import transaction
from pyramid import testing
from ..models import (
User,
Shop,
Invoice,
Product,
InvoiceLineItem,
CryptoPayment,
CryptoProcessor,
)
from ..lib.crypto_watcher import (
process_payment,
scan_wallet_for_double_or_late_payments,
run_once,
)
from .test_integration import DatabaseIntegrationTests
class TestDoubleSpendIntegration(DatabaseIntegrationTests):
"""Integration tests with real database for double-spend scenarios."""
def _create_payment_scenario(
self, status=CryptoPayment.STATUS_PENDING, received=0, expected=100000000
):
"""Helper to create a complete payment scenario with all models."""
# Create user and shop
user = User(email=f"test-{uuid.uuid4()}@example.com")
user.set_password("testpass")
self.dbsession.add(user)
shop = Shop(
name=f"Test Shop {uuid.uuid4()}",
user=user,
slug=f"test-shop-{uuid.uuid4()}",
crypto_xmr_address="4TestAddress...",
crypto_xmr_subaddress_lookahead=20,
)
self.dbsession.add(shop)
self.dbsession.flush()
# Create product and invoice
product = Product(
name="Test Product",
shop=shop,
price_cents=100,
description="Test",
stock_quantity=10,
)
self.dbsession.add(product)
invoice = Invoice(user=user, shop=shop)
self.dbsession.add(invoice)
self.dbsession.flush()
line_item = InvoiceLineItem(
invoice=invoice,
product=product,
quantity=1,
price_cents=100,
)
self.dbsession.add(line_item)
# Create crypto payment
payment = CryptoPayment(
invoice=invoice,
user=user,
shop=shop,
coin_type="XMR",
address="4TestPaymentAddress...",
account_index=0,
subaddress_index=1,
expected_amount=expected,
confirmations_required=10,
rate_locked_usd_per_coin=Decimal("150.00"),
expires_at=int((time.time() + 900) * 1000),
refund_address="4RefundAddress...",
shop_sweep_to_address="4ShopSweepAddress...",
)
payment.status = status
payment.received_amount = received
payment.tx_hashes = json.dumps([])
self.dbsession.add(payment)
self.dbsession.flush()
return payment, invoice, shop, user
def test_double_payment_creates_separate_refund_entry(self):
"""Test that duplicate payment creates a separate database entry."""
# Create confirmed payment
payment, invoice, shop, user = self._create_payment_scenario(
status=CryptoPayment.STATUS_CONFIRMED, received=100000000
)
payment.tx_hashes = json.dumps(["first-tx"])
self.dbsession.add(payment)
self.dbsession.flush()
original_id = payment.id
# Mock wallet RPC to return duplicate transaction
with patch(
"make_post_sell.lib.crypto_watcher.get_crypto_client"
) as mock_client:
mock_rpc = MagicMock()
mock_client.return_value = mock_rpc
# Simulate wallet scan finding duplicate payment
mock_rpc._call.return_value = {
"in": [
{
"txid": "duplicate-tx",
"amount": 50000000,
"confirmations": 5,
"subaddr_index": {"major": 0, "minor": 1},
"height": 2000000,
}
]
}
# Create crypto processor for scan
processor = CryptoProcessor(
shop=shop,
coin_type="XMR",
enabled=True,
last_scan_semaphore="xmr:1999999",
)
self.dbsession.add(processor)
self.dbsession.flush()
# Run the scanner
request = testing.DummyRequest()
request.dbsession = self.dbsession
request.registry = testing.DummyRegistry()
request.registry.settings = self.settings
scan_wallet_for_double_or_late_payments(request, self.settings)
# Verify duplicate payment was created
all_payments = (
self.dbsession.query(CryptoPayment)
.filter(CryptoPayment.subaddress_index == 1)
.all()
)
self.assertEqual(len(all_payments), 2)
# Find the duplicate
duplicate = [p for p in all_payments if p.id != original_id][0]
# Verify duplicate properties
self.assertEqual(duplicate.status, CryptoPayment.STATUS_DOUBLEPAY_REFUND)
self.assertEqual(duplicate.received_amount, 50000000)
self.assertEqual(duplicate.expected_amount, 50000000)
self.assertIsNone(duplicate.invoice_id) # No invoice
self.assertEqual(duplicate.refund_address, payment.refund_address)
def test_main_loop_skips_confirmed_duplicate_transactions(self):
"""Test that main processing loop doesn't process duplicates."""
# Create payment that just got confirmed
payment, invoice, shop, user = self._create_payment_scenario(
status=CryptoPayment.STATUS_CONFIRMED, received=100000000
)
payment.tx_hashes = json.dumps(["confirmed-tx"])
payment.current_confirmations = 10
self.dbsession.add(payment)
self.dbsession.flush()
# Mock incoming duplicate transaction
with patch(
"make_post_sell.lib.crypto_watcher.get_crypto_client"
) as mock_client:
mock_rpc = MagicMock()
mock_client.return_value = mock_rpc
# Mock wallet returns both original and duplicate
mock_rpc.get_transfers_for_subaddr.return_value = {
"in": [
{"txid": "confirmed-tx", "amount": 100000000, "confirmations": 15},
{
"txid": "duplicate-tx", # NEW!
"amount": 75000000,
"confirmations": 2,
},
]
}
request = testing.DummyRequest()
request.dbsession = self.dbsession
request.registry = testing.DummyRegistry()
request.registry.settings = self.settings
# Process payment through main loop
with patch("make_post_sell.lib.crypto_watcher.logger") as mock_logger:
process_payment(
request,
payment,
mock_rpc.get_transfers_for_subaddr.return_value["in"],
mock_rpc,
)
# Should log duplicate warning
mock_logger.warning.assert_called()
warning_calls = [
call[0][0] for call in mock_logger.warning.call_args_list
]
duplicate_warnings = [
w for w in warning_calls if "DUPLICATE PAYMENT DETECTED" in w
]
self.assertGreater(len(duplicate_warnings), 0)
# Verify payment amount didn't change
self.dbsession.refresh(payment)
self.assertEqual(payment.received_amount, 100000000) # Not 175000000
def test_race_condition_double_refund_prevention(self):
"""Test that concurrent processing doesn't cause double refunds."""
# Create payment ready for refund
payment, invoice, shop, user = self._create_payment_scenario(
status=CryptoPayment.STATUS_EXPIRED_REFUNDED, received=100000000
)
payment.tx_hashes = json.dumps(["late-payment"])
payment.refund_address = "4UserRefund..."
payment.current_confirmations = 10
self.dbsession.add(payment)
self.dbsession.flush()
payment_id = payment.id
# Simulate two workers processing simultaneously
with patch("make_post_sell.lib.crypto_watcher.PaymentRescue") as MockRescue:
mock_rescue = MagicMock()
MockRescue.return_value = mock_rescue
# Mock refund execution
refund_call_count = 0
def mock_execute_refund(details, payment):
nonlocal refund_call_count
refund_call_count += 1
return {"success": True, "tx_hash": f"refund-tx-{refund_call_count}"}
mock_rescue.execute_refund = mock_execute_refund
mock_rescue.handle_expired_payment.return_value = {
"payment_id": payment_id,
"refund_address": "4UserRefund...",
"refund_amount": Decimal("0.00091"), # With 9% fee
"reason": "Late payment",
}
# Worker 1 processes
request1 = testing.DummyRequest()
request1.dbsession = self.dbsession
with patch("make_post_sell.lib.crypto_watcher.get_crypto_client"):
process_payment(request1, payment, [], None)
# Worker 2 tries to process same payment
request2 = testing.DummyRequest()
request2.dbsession = self.dbsession
# Refresh to get latest state
self.dbsession.refresh(payment)
with patch("make_post_sell.lib.crypto_watcher.get_crypto_client"):
process_payment(request2, payment, [], None)
# Should only refund once
self.assertEqual(refund_call_count, 1)
def test_full_payment_flow_with_double_spend(self):
"""Test complete flow: payment, confirmation, double-spend, refund."""
# 1. Create pending payment
payment, invoice, shop, user = self._create_payment_scenario()
self.dbsession.flush()
# 2. First payment arrives
with patch("make_post_sell.lib.crypto_watcher.get_crypto_client"):
request = testing.DummyRequest()
request.dbsession = self.dbsession
request.registry = testing.DummyRegistry()
request.registry.settings = self.settings
first_tx = [{"txid": "payment-1", "amount": 100000000, "confirmations": 10}]
process_payment(request, payment, first_tx)
# Should be confirmed
self.assertEqual(payment.status, CryptoPayment.STATUS_CONFIRMED)
self.assertEqual(payment.received_amount, 100000000)
# 3. Double payment arrives (user accidentally pays again)
# This should be caught by scanner, not main loop
# First, verify main loop rejects it
double_tx = [
{"txid": "payment-1", "amount": 100000000, "confirmations": 11},
{"txid": "payment-2", "amount": 80000000, "confirmations": 1}, # New!
]
with patch("make_post_sell.lib.crypto_watcher.logger") as mock_logger:
process_payment(request, payment, double_tx)
# Should detect duplicate
mock_logger.warning.assert_called()
# Amount shouldn't change
self.assertEqual(payment.received_amount, 100000000)
# 4. Scanner handles it properly
# Create processor
processor = CryptoProcessor(
shop=shop, coin_type="XMR", enabled=True, last_scan_semaphore="xmr:1999999"
)
self.dbsession.add(processor)
self.dbsession.flush()
with patch(
"make_post_sell.lib.crypto_watcher.get_crypto_client"
) as mock_client:
mock_rpc = MagicMock()
mock_client.return_value = mock_rpc
# Scanner sees the double payment
mock_rpc._call.return_value = {
"in": [
{
"txid": "payment-2",
"amount": 80000000,
"confirmations": 1,
"subaddr_index": {
"major": 0,
"minor": payment.subaddress_index,
},
"height": 2000001,
}
]
}
# Mock refund execution
with patch("make_post_sell.lib.crypto_watcher.PaymentRescue") as MockRescue:
mock_rescue = MagicMock()
MockRescue.return_value = mock_rescue
mock_rescue.handle_expired_payment.return_value = {
"payment_id": "new-duplicate-id",
"refund_address": payment.refund_address,
"refund_amount": Decimal("0.000728"), # 80M - 9% fee
"reason": "Duplicate payment",
}
mock_rescue.execute_refund.return_value = {
"success": True,
"tx_hash": "refund-tx-123",
}
scan_wallet_for_double_or_late_payments(request, self.settings)
# Verify duplicate payment was created
duplicates = (
self.dbsession.query(CryptoPayment)
.filter(CryptoPayment.status == CryptoPayment.STATUS_DOUBLEPAY_REFUND)
.all()
)
self.assertEqual(len(duplicates), 1)
duplicate = duplicates[0]
self.assertEqual(duplicate.received_amount, 80000000)
self.assertEqual(duplicate.refund_tx_hash, "refund-tx-123")
def test_overpayment_vs_double_payment(self):
"""Test distinction between overpayment and double payment."""
# Scenario 1: Overpayment in single transaction
payment1, _, _, _ = self._create_payment_scenario(
expected=100000000 # Expecting 0.001 XMR
)
request = testing.DummyRequest()
request.dbsession = self.dbsession
# User sends 0.0015 XMR in one transaction
overpay_tx = [
{"txid": "overpay-1", "amount": 150000000, "confirmations": 10} # 50% more
]
with patch("make_post_sell.lib.crypto_watcher.PaymentRescue") as MockRescue:
mock_rescue = MagicMock()
MockRescue.return_value = mock_rescue
mock_rescue.handle_overpayment.return_value = {
"payment_id": payment1.id,
"refund_address": payment1.refund_address,
"excess_amount": Decimal("0.0005"),
"refund_amount": Decimal("0.000455"), # After 9% fee
"reason": "Overpayment",
}
process_payment(request, payment1, overpay_tx)
# Should be marked as overpaid, not double-spend
self.assertEqual(payment1.status, CryptoPayment.STATUS_CONFIRMED_OVERPAID)
self.assertEqual(payment1.received_amount, 150000000)
# Scenario 2: Double payment (two separate transactions)
payment2, _, _, _ = self._create_payment_scenario()
# First exact payment
tx1 = [{"txid": "exact-1", "amount": 100000000, "confirmations": 10}]
process_payment(request, payment2, tx1)
self.assertEqual(payment2.status, CryptoPayment.STATUS_CONFIRMED)
# Second payment (should be caught as duplicate)
payment2.tx_hashes = json.dumps(["exact-1"]) # Update stored
tx2 = [
{"txid": "exact-1", "amount": 100000000, "confirmations": 11},
{"txid": "double-1", "amount": 50000000, "confirmations": 1},
]
with patch("make_post_sell.lib.crypto_watcher.logger") as mock_logger:
process_payment(request, payment2, tx2)
mock_logger.warning.assert_called()
self.assertIn(
"DUPLICATE PAYMENT DETECTED", mock_logger.warning.call_args[0][0]
)
if __name__ == "__main__":
import unittest
unittest.main()

741
make_post_sell/tests/test_double_spend_protection.py Normal file
View file

@ -0,0 +1,741 @@
"""
Comprehensive tests for double-spend protection in crypto payment processing.
These tests verify that the system correctly handles:
1. Duplicate payments to confirmed addresses
2. Race conditions between main loop and scanner
3. Multiple rapid payments
4. Wallet drain prevention
"""
import json
import time
import unittest
from decimal import Decimal
from unittest.mock import MagicMock, patch, call
from pyramid import testing
from ..models.crypto_payment import CryptoPayment
from ..lib.crypto_watcher import (
process_payment,
scan_wallet_for_double_or_late_payments,
_should_process_late_payment,
_create_duplicate_payment,
)
class TestDoubleSpendProtection(unittest.TestCase):
"""Unit tests for double-spend protection logic."""
def setUp(self):
self.config = testing.setUp()
self.request = testing.DummyRequest()
self.request.dbsession = MagicMock()
def tearDown(self):
testing.tearDown()
def _create_test_payment(self, **kwargs):
"""Helper to create a test payment with mock dependencies."""
defaults = {
"invoice": MagicMock(id="invoice-123", user=MagicMock()),
"user": MagicMock(id="user-123"),
"shop": MagicMock(id="shop-123"),
"address": "4TestAddress...",
"account_index": 0,
"subaddress_index": 1,
"coin_type": "XMR",
"expected_amount": 100000000,
"rate_locked_usd_per_coin": Decimal("150.00"),
"quote_expires_at_ms": int((time.time() + 900) * 1000),
"confirmations_required": 10,
"refund_address": "4RefundAddress...",
"shop_sweep_to_address": "4ShopAddress...",
}
defaults.update(kwargs)
payment = CryptoPayment(**defaults)
# Override properties that were set in __init__
for key, value in kwargs.items():
if hasattr(payment, key):
setattr(payment, key, value)
return payment
def test_duplicate_payment_to_confirmed_address_rejected_by_main_loop(self):
"""Test that main loop rejects new transactions to already-paid addresses."""
# Create a confirmed payment that already received funds
payment = self._create_test_payment()
payment.id = "test-payment-1"
payment.status = CryptoPayment.STATUS_CONFIRMED
payment.received_amount = 100000000 # Already received 0.001 XMR
payment.expected_amount = 100000000
payment.tx_hashes = json.dumps(["existing-tx-123"])
payment.coin_type = "XMR"
# New transaction arrives (duplicate payment)
incoming_transfers = [
{
"txid": "new-tx-456", # Different transaction
"amount": 50000000, # 0.0005 XMR
"confirmations": 1,
"subaddr_index": {"major": 0, "minor": 1},
}
]
# Mock the database session
self.request.dbsession.add = MagicMock()
# Process payment - should detect duplicate and return early
with patch("make_post_sell.lib.crypto_watcher.logger") as mock_logger:
process_payment(self.request, payment, incoming_transfers)
# Verify warning was logged
mock_logger.warning.assert_called_once()
warning_msg = mock_logger.warning.call_args[0][0]
self.assertIn("DUPLICATE PAYMENT DETECTED", warning_msg)
self.assertIn("1 new transactions", warning_msg)
# Verify payment was NOT updated with new amount
self.assertEqual(payment.received_amount, 100000000) # Unchanged
# Verify confirmation count was still updated
self.assertEqual(payment.current_confirmations, 1)
def test_duplicate_payment_to_received_status_rejected(self):
"""Test that payments in RECEIVED status also reject duplicates."""
payment = self._create_test_payment()
payment.id = "test-payment-2"
payment.status = CryptoPayment.STATUS_RECEIVED
payment.received_amount = 50000000 # Already received 0.0005 XMR
payment.expected_amount = 100000000 # Expecting 0.001 XMR
payment.tx_hashes = json.dumps(["first-tx-789"])
payment.coin_type = "XMR"
# Another transaction arrives before confirmation
incoming_transfers = [
{
"txid": "second-tx-abc",
"amount": 60000000, # 0.0006 XMR - would overpay
"confirmations": 2,
}
]
self.request.dbsession.add = MagicMock()
with patch("make_post_sell.lib.crypto_watcher.logger") as mock_logger:
process_payment(self.request, payment, incoming_transfers)
# Should detect as duplicate
mock_logger.warning.assert_called_once()
self.assertIn(
"DUPLICATE PAYMENT DETECTED", mock_logger.warning.call_args[0][0]
)
def test_pending_payment_accepts_first_transaction(self):
"""Test that PENDING status correctly accepts the first transaction."""
payment = self._create_test_payment()
payment.id = "test-payment-3"
payment.status = CryptoPayment.STATUS_PENDING
payment.received_amount = 0 # No funds received yet
payment.expected_amount = 100000000
payment.tx_hashes = json.dumps([])
payment.coin_type = "XMR"
payment.confirmations_required = 10
incoming_transfers = [
{
"txid": "first-tx-def",
"amount": 100000000,
"confirmations": 1,
}
]
# Mock database and finalization
self.request.dbsession.add = MagicMock()
with patch("make_post_sell.lib.crypto_watcher.logger") as mock_logger:
process_payment(self.request, payment, incoming_transfers)
# Should NOT log duplicate warning
mock_logger.warning.assert_not_called()
# Payment should be updated
self.assertEqual(payment.received_amount, 100000000)
self.assertEqual(payment.status, CryptoPayment.STATUS_RECEIVED)
def test_should_process_late_payment_detects_doubles(self):
"""Test _should_process_late_payment correctly identifies double payments."""
# Test confirmed payment receiving more funds
payment = self._create_test_payment()
payment.id = "test-payment-4"
payment.status = CryptoPayment.STATUS_CONFIRMED
payment.received_amount = 100000000
payment.expected_amount = 100000000
new_tx = {"amount": 50000000, "confirmations": 5}
# Should return True - this is a double payment
result = _should_process_late_payment(payment, new_tx)
self.assertTrue(result)
# Test confirmed overpaid payment
payment.status = CryptoPayment.STATUS_CONFIRMED_OVERPAID
result = _should_process_late_payment(payment, new_tx)
self.assertTrue(result)
def test_create_duplicate_payment_object(self):
"""Test creation of duplicate payment objects for refunds."""
original = self._create_test_payment()
original.id = "original-payment"
original.coin_type = "XMR"
original.account_index = 0
original.subaddress_index = 123
original.address = "4ABC..."
original.refund_address = "4DEF..."
original.shop_sweep_to_address = "4GHI..."
original.confirmations_required = 10
original.user_id = "user-123"
original.shop_id = "shop-456"
original.shop_location_id = "loc-789"
tx = {"amount": 75000000, "txid": "duplicate-tx-xyz", "confirmations": 3}
duplicate = _create_duplicate_payment(original, tx, "XMR")
# Verify duplicate has correct properties
self.assertNotEqual(duplicate.id, original.id)
self.assertIsNone(duplicate.invoice_id) # No invoice for duplicates
self.assertEqual(duplicate.status, CryptoPayment.STATUS_DOUBLEPAY_REFUND)
self.assertEqual(duplicate.expected_amount, 75000000)
self.assertEqual(duplicate.received_amount, 75000000)
self.assertEqual(duplicate.refund_address, original.refund_address)
self.assertEqual(duplicate.current_confirmations, 3)
def test_race_condition_protection(self):
"""Test that race conditions don't cause double refunds."""
# Simulate two processes seeing the same payment
payment1 = self._create_test_payment()
payment1.id = "payment-race"
payment1.status = CryptoPayment.STATUS_CONFIRMED_OVERPAID
payment1.received_amount = 150000000 # Got 0.0015 XMR
payment1.expected_amount = 100000000 # Expected 0.001 XMR
payment1.tx_hashes = json.dumps(["tx1", "tx2"])
# Both processes see a new transaction
new_tx = [{"txid": "tx3", "amount": 50000000, "confirmations": 1}]
# Process 1 handles it
self.request.dbsession.add = MagicMock()
with patch("make_post_sell.lib.crypto_watcher.logger") as mock_logger:
process_payment(self.request, payment1, new_tx)
# Should reject as duplicate
mock_logger.warning.assert_called_once()
# Payment amount should NOT increase
self.assertEqual(payment1.received_amount, 150000000)
def test_multiple_rapid_payments_scenario(self):
"""Test handling of multiple payments in quick succession."""
payment = self._create_test_payment()
payment.id = "rapid-payment"
payment.status = CryptoPayment.STATUS_PENDING
payment.received_amount = 0
payment.expected_amount = 100000000
payment.tx_hashes = json.dumps([])
payment.confirmations_required = 10
payment.coin_type = "XMR"
# First transaction
tx1 = [{"txid": "rapid1", "amount": 40000000, "confirmations": 1}]
self.request.dbsession.add = MagicMock()
process_payment(self.request, payment, tx1)
self.assertEqual(payment.received_amount, 40000000)
self.assertEqual(payment.status, CryptoPayment.STATUS_RECEIVED)
# Second transaction (legitimate - user adding more)
# Reset to pending to allow additional payment
payment.status = CryptoPayment.STATUS_PENDING
payment.received_amount = 0 # Reset for clean test
payment.tx_hashes = json.dumps([]) # Reset
tx2 = [
{"txid": "rapid1", "amount": 40000000, "confirmations": 2},
{"txid": "rapid2", "amount": 60000000, "confirmations": 1},
]
process_payment(self.request, payment, tx2)
self.assertEqual(payment.received_amount, 100000000)
# Third transaction (duplicate - should be rejected)
# Set to received status so duplicates are detected
payment.status = CryptoPayment.STATUS_RECEIVED
payment.tx_hashes = json.dumps(["rapid1", "rapid2"])
payment.received_amount = 100000000 # Set existing balance
tx3 = [
{"txid": "rapid1", "amount": 40000000, "confirmations": 3},
{"txid": "rapid2", "amount": 60000000, "confirmations": 2},
{
"txid": "rapid3",
"amount": 50000000,
"confirmations": 1,
}, # NEW transaction
]
with patch("make_post_sell.lib.crypto_watcher.logger") as mock_logger:
process_payment(self.request, payment, tx3)
# Should detect duplicate
mock_logger.warning.assert_called_once()
# Amount should not change
self.assertEqual(payment.received_amount, 100000000)
def test_concurrent_transactions_within_processing_window(self):
"""Test two legitimate transactions arriving within same processing cycle."""
# Scenario: Customer sends partial payment, then immediately sends the rest
# Both arrive before either is processed (within same 20-second window)
payment = self._create_test_payment()
payment.id = "concurrent-payment"
payment.status = CryptoPayment.STATUS_PENDING
payment.received_amount = 0
payment.expected_amount = 100000000 # Expecting 0.001 XMR
payment.tx_hashes = json.dumps([])
payment.confirmations_required = 10
payment.coin_type = "XMR"
# Both transactions arrive in same batch (realistic scenario)
# Customer sends 0.0006 XMR then quickly sends another 0.0004 XMR
concurrent_txs = [
{
"txid": "tx1-partial",
"amount": 60000000,
"confirmations": 10,
}, # 0.0006 XMR - enough confirmations
{
"txid": "tx2-completion",
"amount": 40000000,
"confirmations": 10,
}, # 0.0004 XMR - enough confirmations
]
self.request.dbsession.add = MagicMock()
with patch("make_post_sell.lib.crypto_watcher.finalize_invoice"):
process_payment(self.request, payment, concurrent_txs)
# Should process BOTH transactions as they're both new and legitimate
self.assertEqual(
payment.received_amount, 100000000
) # 0.0006 + 0.0004 = 0.001 XMR
self.assertEqual(
payment.status, CryptoPayment.STATUS_CONFIRMED
) # Should be confirmed
# Should track both transaction IDs
stored_txids = json.loads(payment.tx_hashes)
self.assertIn("tx1-partial", stored_txids)
self.assertIn("tx2-completion", stored_txids)
self.assertEqual(len(stored_txids), 2)
def test_sequential_overpayment_detection(self):
"""Test sequential transactions where second is detected as duplicate."""
# Scenario: Customer sends first payment (gets confirmed), then sends second payment
# The second should be detected as duplicate and handled by scanner, not main loop
payment = self._create_test_payment()
payment.id = "sequential-overpay"
payment.status = CryptoPayment.STATUS_PENDING
payment.received_amount = 0
payment.expected_amount = 100000000 # Expecting 0.001 XMR
payment.tx_hashes = json.dumps([])
payment.confirmations_required = 10
payment.coin_type = "XMR"
# First transaction - legitimate payment
first_tx = [
{"txid": "tx1-legitimate", "amount": 100000000, "confirmations": 10}
]
self.request.dbsession.add = MagicMock()
with patch("make_post_sell.lib.crypto_watcher.finalize_invoice"):
process_payment(self.request, payment, first_tx)
# Verify first payment was processed
self.assertEqual(payment.received_amount, 100000000)
self.assertEqual(payment.status, CryptoPayment.STATUS_CONFIRMED)
# Now second transaction arrives (this is the duplicate scenario)
# Payment is already confirmed, so this should be rejected by main loop
payment.tx_hashes = json.dumps(["tx1-legitimate"]) # Update stored txids
second_tx = [
{
"txid": "tx1-legitimate",
"amount": 100000000,
"confirmations": 15,
}, # Same TX, more confirmations
{
"txid": "tx2-duplicate",
"amount": 50000000,
"confirmations": 1,
}, # NEW duplicate transaction
]
with patch("make_post_sell.lib.crypto_watcher.logger") as mock_logger:
process_payment(self.request, payment, second_tx)
# Should detect duplicate and reject
mock_logger.warning.assert_called_once()
warning_msg = mock_logger.warning.call_args[0][0]
self.assertIn("DUPLICATE PAYMENT DETECTED", warning_msg)
self.assertIn("1 new transactions", warning_msg)
# Amount should NOT change (duplicate rejected)
self.assertEqual(payment.received_amount, 100000000) # Still 0.001 XMR
self.assertEqual(
payment.status, CryptoPayment.STATUS_CONFIRMED
) # Still confirmed
# Should update confirmation count but not add duplicate
# Note: confirmation count is minimum of all transactions including the duplicate (1)
self.assertEqual(payment.current_confirmations, 1)
def test_second_processing_cycle_rejects_duplicates(self):
"""Test that subsequent processing cycles correctly reject already-seen transactions."""
# Scenario: First cycle processes two concurrent transactions,
# then 20 seconds later the system processes again and sees the same transactions
payment = self._create_test_payment()
payment.id = "cycle-test"
payment.status = CryptoPayment.STATUS_PENDING
payment.received_amount = 0
payment.expected_amount = 100000000
payment.tx_hashes = json.dumps([])
payment.confirmations_required = 10
payment.coin_type = "XMR"
# First processing cycle - concurrent transactions
first_cycle_txs = [
{"txid": "concurrent1", "amount": 50000000, "confirmations": 10},
{"txid": "concurrent2", "amount": 50000000, "confirmations": 10},
]
self.request.dbsession.add = MagicMock()
with patch("make_post_sell.lib.crypto_watcher.finalize_invoice"):
process_payment(self.request, payment, first_cycle_txs)
# Verify first cycle worked
self.assertEqual(payment.received_amount, 100000000)
self.assertEqual(payment.status, CryptoPayment.STATUS_CONFIRMED)
stored_txids = json.loads(payment.tx_hashes)
self.assertEqual(len(stored_txids), 2)
# Second processing cycle - same transactions with more confirmations
# Plus a new transaction (this is the duplicate scenario)
second_cycle_txs = [
{
"txid": "concurrent1",
"amount": 50000000,
"confirmations": 15,
}, # Same as before
{
"txid": "concurrent2",
"amount": 50000000,
"confirmations": 15,
}, # Same as before
{
"txid": "new-duplicate",
"amount": 25000000,
"confirmations": 1,
}, # NEW transaction
]
# Reset mocks for second cycle
self.request.dbsession.add.reset_mock()
with patch("make_post_sell.lib.crypto_watcher.logger") as mock_logger:
process_payment(self.request, payment, second_cycle_txs)
# Should detect the new transaction as a duplicate
mock_logger.warning.assert_called_once()
warning_msg = mock_logger.warning.call_args[0][0]
self.assertIn("DUPLICATE PAYMENT DETECTED", warning_msg)
self.assertIn("1 new transactions", warning_msg)
# Amount should NOT change (no new funds added)
self.assertEqual(payment.received_amount, 100000000)
# Only update confirmation count, not add new funds
# Note: confirmation count is minimum of all transactions including the duplicate (1)
self.assertEqual(payment.current_confirmations, 1) # Updated confirmations
class TestWalletDrainPrevention(unittest.TestCase):
"""Test that the system prevents wallet drain scenarios."""
def setUp(self):
self.config = testing.setUp()
self.request = testing.DummyRequest()
self.request.dbsession = MagicMock()
self.request.registry = MagicMock()
self.request.registry.settings = {
"crypto.xmr.wallet_rpc_uri": "http://localhost:18082",
"crypto.xmr.wallet_rpc_user": "test",
"crypto.xmr.wallet_rpc_password": "test",
}
def tearDown(self):
testing.tearDown()
def _create_test_payment(self, **kwargs):
"""Helper to create a test payment with mock dependencies."""
defaults = {
"invoice": MagicMock(id="invoice-123", user=MagicMock()),
"user": MagicMock(id="user-123"),
"shop": MagicMock(id="shop-123"),
"address": "4TestAddress...",
"account_index": 0,
"subaddress_index": 1,
"coin_type": "XMR",
"expected_amount": 100000000,
"rate_locked_usd_per_coin": Decimal("150.00"),
"quote_expires_at_ms": int((time.time() + 900) * 1000),
"confirmations_required": 10,
"refund_address": "4RefundAddress...",
"shop_sweep_to_address": "4ShopAddress...",
}
defaults.update(kwargs)
payment = CryptoPayment(**defaults)
# Override properties that were set in __init__
for key, value in kwargs.items():
if hasattr(payment, key):
setattr(payment, key, value)
return payment
@patch("make_post_sell.lib.crypto_watcher.get_crypto_client")
def test_sweep_all_not_called_multiple_times(self, mock_get_client):
"""Verify sweep_all is not called multiple times for same payment."""
# Mock crypto client
mock_client = MagicMock()
mock_get_client.return_value = mock_client
# Create a payment ready for sweep
payment = self._create_test_payment()
payment.id = "sweep-test"
payment.status = CryptoPayment.STATUS_CONFIRMED
payment.received_amount = 100000000
payment.shop_sweep_to_address = "4ABC..."
payment.account_index = 0
payment.subaddress_index = 99
payment.coin_type = "XMR"
payment.invoice = MagicMock()
payment.swept_tx_hash = None
# Mock sweep_all response
mock_client._call.return_value = {
"tx_hash_list": ["sweep-tx-123"],
"amount_list": [95000000],
"fee_list": [5000000],
}
# Test logic - sweep should only happen once per payment
# First sweep
payment.swept_tx_hash = None
# Simulate sweep logic
if not payment.swept_tx_hash:
payment.swept_tx_hash = "sweep-tx-123"
call_count = 1
else:
call_count = 0
# Second attempt should be skipped
if not payment.swept_tx_hash: # Already set, so this won't execute
call_count += 1
# Should only sweep once
self.assertEqual(call_count, 1)
self.assertEqual(payment.swept_tx_hash, "sweep-tx-123")
def test_wallet_scan_creates_separate_duplicate_payments(self):
"""Test that scanner creates separate payment objects for duplicates."""
# Mock database query
mock_payment = self._create_test_payment()
mock_payment.id = "original"
mock_payment.status = CryptoPayment.STATUS_CONFIRMED
mock_payment.received_amount = 100000000
mock_payment.coin_type = "XMR"
mock_payment.account_index = 0
mock_payment.subaddress_index = 50
self.request.dbsession.query.return_value.filter.return_value.first.return_value = (
mock_payment
)
self.request.dbsession.add = MagicMock()
self.request.dbsession.flush = MagicMock()
# Mock duplicate payment creation
with patch(
"make_post_sell.lib.crypto_watcher._create_duplicate_payment"
) as mock_create:
mock_duplicate = self._create_test_payment()
mock_duplicate.id = "duplicate-1"
mock_duplicate.status = CryptoPayment.STATUS_DOUBLEPAY_REFUND
mock_create.return_value = mock_duplicate
with patch(
"make_post_sell.lib.crypto_watcher.process_payment"
) as mock_process:
# Simulate finding duplicate payment in scan
tx = {"amount": 50000000, "confirmations": 5}
# Test the duplicate detection logic
should_process = _should_process_late_payment(mock_payment, tx)
self.assertTrue(should_process)
# Test duplicate creation
duplicate = _create_duplicate_payment(mock_payment, tx, "XMR")
self.assertEqual(
duplicate.status, CryptoPayment.STATUS_DOUBLEPAY_REFUND
)
self.assertEqual(duplicate.received_amount, 50000000)
# Verify duplicate properties
self.assertEqual(duplicate.coin_type, "XMR")
self.assertEqual(duplicate.account_index, mock_payment.account_index)
self.assertEqual(
duplicate.subaddress_index, mock_payment.subaddress_index
)
class TestEdgeCasesAndErrorHandling(unittest.TestCase):
"""Test edge cases and error scenarios."""
def setUp(self):
self.config = testing.setUp()
self.request = testing.DummyRequest()
self.request.dbsession = MagicMock()
def tearDown(self):
testing.tearDown()
def _create_test_payment(self, **kwargs):
"""Helper to create a test payment with mock dependencies."""
defaults = {
"invoice": MagicMock(id="invoice-123", user=MagicMock()),
"user": MagicMock(id="user-123"),
"shop": MagicMock(id="shop-123"),
"address": "4TestAddress...",
"account_index": 0,
"subaddress_index": 1,
"coin_type": "XMR",
"expected_amount": 100000000,
"rate_locked_usd_per_coin": Decimal("150.00"),
"quote_expires_at_ms": int((time.time() + 900) * 1000),
"confirmations_required": 10,
"refund_address": "4RefundAddress...",
"shop_sweep_to_address": "4ShopAddress...",
}
defaults.update(kwargs)
payment = CryptoPayment(**defaults)
# Override properties that were set in __init__
for key, value in kwargs.items():
if hasattr(payment, key):
setattr(payment, key, value)
return payment
def test_malformed_tx_hashes_json(self):
"""Test handling of corrupted tx_hashes JSON."""
payment = self._create_test_payment()
payment.id = "corrupt-json"
payment.status = CryptoPayment.STATUS_PENDING # Start with pending
payment.received_amount = 0 # No funds yet
payment.tx_hashes = "not-valid-json{[" # Corrupted
payment.expected_amount = 100000000
payment.coin_type = "XMR"
payment.confirmations_required = 10
incoming = [{"txid": "new-tx", "amount": 50000000, "confirmations": 1}]
# Should handle gracefully and treat as empty list
self.request.dbsession.add = MagicMock()
with patch("make_post_sell.lib.crypto_watcher.finalize_invoice"):
process_payment(self.request, payment, incoming)
# Should process normally since corrupt JSON is treated as []
self.assertEqual(payment.received_amount, 50000000)
def test_payment_with_no_txid(self):
"""Test handling of transactions without txid."""
payment = self._create_test_payment()
payment.id = "no-txid"
payment.status = CryptoPayment.STATUS_PENDING
payment.received_amount = 0
payment.tx_hashes = json.dumps([])
payment.expected_amount = 100000000
payment.coin_type = "DOGE"
payment.confirmations_required = 3
# Transaction missing txid - this won't be processed because
# transactions without txid can't be tracked for deduplication
incoming = [
{"amount": 100000000, "confirmations": 3}
] # No txid, meets confirmation requirement
self.request.dbsession.add = MagicMock()
with patch("make_post_sell.lib.crypto_watcher.finalize_invoice"):
process_payment(self.request, payment, incoming)
# Should NOT process because txid is required for tracking
self.assertEqual(payment.received_amount, 0) # No change
self.assertEqual(json.loads(payment.tx_hashes), [])
def test_zero_amount_transactions(self):
"""Test handling of zero-amount transactions."""
payment = self._create_test_payment()
payment.id = "zero-amount"
payment.status = CryptoPayment.STATUS_PENDING
payment.received_amount = 0
payment.tx_hashes = json.dumps([])
payment.expected_amount = 100000000
payment.coin_type = "XMR"
incoming = [
{"txid": "zero-tx", "amount": 0, "confirmations": 10},
{"txid": "real-tx", "amount": 100000000, "confirmations": 10},
]
self.request.dbsession.add = MagicMock()
with patch("make_post_sell.lib.crypto_watcher.finalize_invoice"):
process_payment(self.request, payment, incoming)
# Should only count non-zero transaction
self.assertEqual(payment.received_amount, 100000000)
def test_payment_status_change_during_processing(self):
"""Test handling when payment status changes mid-processing."""
payment = self._create_test_payment()
payment.id = "status-change"
payment.status = CryptoPayment.STATUS_RECEIVED
payment.received_amount = 100000000
payment.expected_amount = 100000000
payment.tx_hashes = json.dumps(["tx1"])
# Mock refresh to simulate status change
def mock_refresh(obj):
obj.status = CryptoPayment.STATUS_CANCELLED
self.request.dbsession.refresh = mock_refresh
self.request.dbsession.add = MagicMock()
# This would be caught in the main processing loop
# The payment would be skipped due to status change
if __name__ == "__main__":
unittest.main()