from django.core.management.base import BaseCommand from payments.services import MpesaService from payments.models import MpesaB2CTransaction, MpesaB2BTransaction, MpesaReversal, MpesaBalanceInquiry from decimal import Decimal class Command(BaseCommand): help = 'Test advanced M-Pesa features (Balance, B2C, B2B, Reversals)' def add_arguments(self, parser): parser.add_argument('--test-balance', action='store_true', help='Test balance inquiry') parser.add_argument('--test-b2c', action='store_true', help='Test B2C payment') parser.add_argument('--test-b2b', action='store_true', help='Test B2B payment') parser.add_argument('--test-reversal', action='store_true', help='Test transaction reversal') parser.add_argument('--phone', type=str, help='Phone number for B2C test') parser.add_argument('--amount', type=str, help='Amount for tests') parser.add_argument('--transaction-id', type=str, help='Transaction ID for reversal test') def handle(self, *args, **options): mpesa_service = MpesaService() self.stdout.write("=" * 60) self.stdout.write("M-PESA ADVANCED FEATURES TEST") self.stdout.write("=" * 60) if options['test_balance']: self.test_balance_inquiry(mpesa_service) if options['test_b2c']: phone = options.get('phone', '0700000000') amount = options.get('amount', '10') self.test_b2c_payment(mpesa_service, phone, amount) if options['test_b2b']: amount = options.get('amount', '10') self.test_b2b_payment(mpesa_service, amount) if options['test_reversal']: transaction_id = options.get('transaction_id') amount = options.get('amount', '10') if transaction_id: self.test_reversal(mpesa_service, transaction_id, amount) else: self.stdout.write(self.style.ERROR("āŒ Transaction ID required for reversal test")) def test_balance_inquiry(self, mpesa_service): self.stdout.write("\nšŸ” Testing Balance Inquiry...") try: response = mpesa_service.check_account_balance() self.stdout.write(f"Response: {response}") if response.get('ResponseCode') == '0': # Create pending record conversation_id = response.get('ConversationID') originator_conversation_id = response.get('OriginatorConversationID') balance_inquiry = MpesaBalanceInquiry.objects.create( conversation_id=conversation_id, originator_conversation_id=originator_conversation_id ) self.stdout.write(self.style.SUCCESS("āœ… Balance inquiry initiated successfully")) self.stdout.write(f" Conversation ID: {conversation_id}") else: error_msg = response.get('errorMessage', response.get('ResponseDescription', 'Unknown error')) self.stdout.write(self.style.ERROR(f"āŒ Balance inquiry failed: {error_msg}")) except Exception as e: self.stdout.write(self.style.ERROR(f"āŒ Balance inquiry error: {str(e)}")) def test_b2c_payment(self, mpesa_service, phone, amount): self.stdout.write(f"\nšŸ’ø Testing B2C Payment to {phone} for KES {amount}...") try: response = mpesa_service.b2c_payment( phone_number=phone, amount=amount, command_id='BusinessPayment', remarks='Test B2C payment', occasion='Testing' ) self.stdout.write(f"Response: {response}") if response.get('ResponseCode') == '0': # Create pending record conversation_id = response.get('ConversationID') originator_conversation_id = response.get('OriginatorConversationID') b2c_transaction = MpesaB2CTransaction.objects.create( conversation_id=conversation_id, originator_conversation_id=originator_conversation_id, command_id='BusinessPayment', amount=Decimal(str(amount)), phone_number=phone, remarks='Test B2C payment', occasion='Testing' ) self.stdout.write(self.style.SUCCESS("āœ… B2C payment initiated successfully")) self.stdout.write(f" Conversation ID: {conversation_id}") else: error_msg = response.get('errorMessage', response.get('ResponseDescription', 'Unknown error')) self.stdout.write(self.style.ERROR(f"āŒ B2C payment failed: {error_msg}")) except Exception as e: self.stdout.write(self.style.ERROR(f"āŒ B2C payment error: {str(e)}")) def test_b2b_payment(self, mpesa_service, amount): self.stdout.write(f"\nšŸ¢ Testing B2B Payment for KES {amount}...") try: # Example B2B payment (you need to replace with actual business short code) response = mpesa_service.b2b_payment( receiver_party='600000', # Replace with actual shortcode amount=amount, account_reference='TEST123', command_id='BusinessPayBill', remarks='Test B2B payment' ) self.stdout.write(f"Response: {response}") if response.get('ResponseCode') == '0': # Create pending record conversation_id = response.get('ConversationID') originator_conversation_id = response.get('OriginatorConversationID') b2b_transaction = MpesaB2BTransaction.objects.create( conversation_id=conversation_id, originator_conversation_id=originator_conversation_id, command_id='BusinessPayBill', amount=Decimal(str(amount)), party_a=mpesa_service.business_short_code, party_b='600000', account_reference='TEST123', remarks='Test B2B payment' ) self.stdout.write(self.style.SUCCESS("āœ… B2B payment initiated successfully")) self.stdout.write(f" Conversation ID: {conversation_id}") else: error_msg = response.get('errorMessage', response.get('ResponseDescription', 'Unknown error')) self.stdout.write(self.style.ERROR(f"āŒ B2B payment failed: {error_msg}")) except Exception as e: self.stdout.write(self.style.ERROR(f"āŒ B2B payment error: {str(e)}")) def test_reversal(self, mpesa_service, transaction_id, amount): self.stdout.write(f"\nšŸ”„ Testing Transaction Reversal for {transaction_id}...") try: response = mpesa_service.reverse_transaction( transaction_id=transaction_id, amount=amount, receiver_party=mpesa_service.business_short_code, remarks='Test reversal', occasion='Testing' ) self.stdout.write(f"Response: {response}") if response.get('ResponseCode') == '0': # Create pending record conversation_id = response.get('ConversationID') originator_conversation_id = response.get('OriginatorConversationID') reversal = MpesaReversal.objects.create( conversation_id=conversation_id, originator_conversation_id=originator_conversation_id, original_transaction_id=transaction_id, amount=Decimal(str(amount)), receiver_party=mpesa_service.business_short_code, remarks='Test reversal', occasion='Testing' ) self.stdout.write(self.style.SUCCESS("āœ… Transaction reversal initiated successfully")) self.stdout.write(f" Conversation ID: {conversation_id}") else: error_msg = response.get('errorMessage', response.get('ResponseDescription', 'Unknown error')) self.stdout.write(self.style.ERROR(f"āŒ Transaction reversal failed: {error_msg}")) except Exception as e: self.stdout.write(self.style.ERROR(f"āŒ Transaction reversal error: {str(e)}")) self.stdout.write("\n" + "=" * 60) self.stdout.write("TEST COMPLETED") self.stdout.write("=" * 60)