""" inventory/tests/test_pos_mpesa.py M-Pesa STK Push from the admin POS: sending prompts, the shared Safaricom callback, and completing a sale against a paid prompt. Daraja is patched. """ import json from datetime import timedelta from decimal import Decimal from unittest.mock import patch from django.contrib.auth import get_user_model from django.test import TestCase, override_settings from django.urls import reverse from django.utils import timezone from catalog_app.models import Author, Book, Catalog from inventory.models import PosMpesaPayment, Sale, StockLocation from inventory.services.inventory_service import receive_stock from store.services.mpesa_service import QUERY_PROCESSING, MpesaError, MpesaService from system_logs.models import SystemLog User = get_user_model() MPESA_SETTINGS = dict( MOCK_MPESA=False, MPESA_STK_ENABLED=True, MPESA_SANDBOX=True, MPESA_CONSUMER_KEY="key", MPESA_CONSUMER_SECRET="secret", MPESA_SHORTCODE="174379", MPESA_PASSKEY="passkey", MPESA_CALLBACK_URL="https://example.test/api/payments/mpesa/callback/", ) CALLBACK_URL = "/api/payments/mpesa/callback/" def success_callback(checkout_id, amount=1, receipt="POSRCPT01"): return {"Body": {"stkCallback": { "MerchantRequestID": "m-1", "CheckoutRequestID": checkout_id, "ResultCode": 0, "ResultDesc": "The service request is processed successfully.", "CallbackMetadata": {"Item": [ {"Name": "Amount", "Value": amount}, {"Name": "MpesaReceiptNumber", "Value": receipt}, {"Name": "TransactionDate", "Value": 20261006120000}, {"Name": "PhoneNumber", "Value": 254713621589}, ]}, }}} def failure_callback(checkout_id, code=1032): return {"Body": {"stkCallback": { "MerchantRequestID": "m-1", "CheckoutRequestID": checkout_id, "ResultCode": code, "ResultDesc": "Request Cancelled by user.", }}} def push_result(checkout_id): return { "MerchantRequestID": "m-1", "CheckoutRequestID": checkout_id, "ResponseCode": "0", "ResponseDescription": "Success. Request accepted for processing", } @override_settings(**MPESA_SETTINGS) class PosMpesaTestBase(TestCase): def setUp(self): self.user = User.objects.create_superuser("cashier", "c@example.test", "pass") self.client.force_login(self.user) author = Author.objects.create(name="POS Author") catalog = Catalog.objects.create(name="POS Catalog", slug="pos-catalog") self.book = Book.objects.create( catalog=catalog, title="POS Book", author=author, price=Decimal("1500.00"), my_keep_percent=Decimal("25.00"), track_inventory=True, is_listed=True, ) self.location, _ = StockLocation.objects.get_or_create( code="MAIN-WH", defaults={"name": "Main Warehouse", "location_type": StockLocation.TYPE_WAREHOUSE}, ) receive_stock(book=self.book, location=self.location, quantity=10, user=self.user) def send_prompt(self, amount="1500.00", previous_id=None): return self.client.post( reverse("admin:inventory_sale_pos_mpesa_stk_push"), data=json.dumps({"phone": "0713621589", "amount": amount, "previous_id": previous_id}), content_type="application/json", ) def make_payment(self, status=PosMpesaPayment.STATUS_PAID, amount="1500.00", checkout="ws_POS_1"): return PosMpesaPayment.objects.create( phone_number="254713621589", amount=Decimal(amount), checkout_request_id=checkout, status=status, mpesa_receipt_number="POSRCPT01" if status == PosMpesaPayment.STATUS_PAID else "", created_by=self.user, ) def complete_sale(self, payment_id, qty=1, price="1500.00"): cart = [{"bookId": self.book.pk, "qty": qty, "price": price, "discount": 0}] return self.client.post(reverse("admin:inventory_sale_pos"), { "cart_data": json.dumps(cart), "stock_location_id": self.location.pk, "sale_channel": Sale.CHANNEL_WALK_IN, "payment_method": Sale.PAYMENT_MPESA, "mpesa_payment_id": str(payment_id), "mpesa_code": "", }) class SendPromptTests(PosMpesaTestBase): @patch.object(MpesaService, "stk_push", return_value=push_result("ws_POS_1")) def test_prompt_creates_pending_payment(self, push): resp = self.send_prompt() self.assertEqual(resp.status_code, 200) payment = PosMpesaPayment.objects.get() self.assertEqual(payment.status, PosMpesaPayment.STATUS_PENDING) self.assertEqual(payment.amount, Decimal("1500.00")) self.assertEqual(payment.checkout_request_id, "ws_POS_1") self.assertEqual(payment.phone_number, "254713621589") self.assertEqual(push.call_args.kwargs["amount"], 1500) self.assertEqual(push.call_args.kwargs["account_reference"], "TAIBOOKSTORE") @patch.object(MpesaService, "stk_push") def test_empty_cart_is_rejected(self, push): self.assertEqual(self.send_prompt(amount="0").status_code, 400) push.assert_not_called() @patch.object(MpesaService, "stk_push") @patch.object(MpesaService, "stk_query", return_value={"state": QUERY_PROCESSING, "result_code": None, "result_desc": "", "raw": {}}) def test_open_previous_prompt_blocks_new_one(self, _query, push): previous = self.make_payment(status=PosMpesaPayment.STATUS_PENDING) resp = self.send_prompt(previous_id=previous.pk) self.assertEqual(resp.status_code, 409) push.assert_not_called() @patch.object(MpesaService, "stk_push") @patch.object(MpesaService, "stk_query", side_effect=MpesaError("M-Pesa status check failed (HTTP 429).")) def test_unconfirmable_previous_prompt_blocks_new_one(self, _query, push): previous = self.make_payment(status=PosMpesaPayment.STATUS_PENDING) resp = self.send_prompt(previous_id=previous.pk) self.assertEqual(resp.status_code, 409) push.assert_not_called() @patch.object(MpesaService, "stk_push", return_value=push_result("ws_POS_2")) @patch.object(MpesaService, "stk_query", side_effect=MpesaError("M-Pesa status check failed (HTTP 429).")) def test_long_expired_unconfirmable_prompt_can_be_replaced(self, _query, push): previous = self.make_payment(status=PosMpesaPayment.STATUS_PENDING) PosMpesaPayment.objects.filter(pk=previous.pk).update(created_at=timezone.now() - timedelta(minutes=4)) resp = self.send_prompt(previous_id=previous.pk) self.assertEqual(resp.status_code, 200) push.assert_called_once() previous.refresh_from_db() self.assertIn("superseded", previous.result_desc) @patch.object(MpesaService, "stk_push") def test_unused_paid_previous_prompt_is_reused(self, push): previous = self.make_payment() resp = self.send_prompt(previous_id=previous.pk) self.assertEqual(resp.json()["id"], previous.pk) push.assert_not_called() @override_settings(MPESA_STK_ENABLED=False) class StkSwitchedOffTests(PosMpesaTestBase): @patch.object(MpesaService, "stk_push") def test_prompt_endpoint_refuses(self, push): resp = self.send_prompt() self.assertEqual(resp.status_code, 403) push.assert_not_called() self.assertFalse(PosMpesaPayment.objects.exists()) self.assertTrue(SystemLog.objects.filter(event="mpesa.pos.disabled").exists()) def test_pos_page_greys_out_prompt_and_shows_manual_code(self): html = self.client.get(reverse("admin:inventory_sale_pos")).content.decode() self.assertIn('title="M-Pesa prompts are not live yet"', html) self.assertIn("not live yet", html) self.assertNotIn("Customer paid via Paybill instead?", html) class CallbackTests(PosMpesaTestBase): def test_callback_marks_pos_payment_paid(self): payment = self.make_payment(status=PosMpesaPayment.STATUS_PENDING) self.client.post(CALLBACK_URL, success_callback("ws_POS_1"), content_type="application/json") payment.refresh_from_db() self.assertEqual(payment.status, PosMpesaPayment.STATUS_PAID) self.assertEqual(payment.mpesa_receipt_number, "POSRCPT01") def test_failure_after_paid_is_ignored(self): payment = self.make_payment(status=PosMpesaPayment.STATUS_PENDING) self.client.post(CALLBACK_URL, success_callback("ws_POS_1"), content_type="application/json") self.client.post(CALLBACK_URL, failure_callback("ws_POS_1"), content_type="application/json") payment.refresh_from_db() self.assertEqual(payment.status, PosMpesaPayment.STATUS_PAID) def test_cancel_marks_failed(self): payment = self.make_payment(status=PosMpesaPayment.STATUS_PENDING) self.client.post(CALLBACK_URL, failure_callback("ws_POS_1"), content_type="application/json") payment.refresh_from_db() self.assertEqual(payment.status, PosMpesaPayment.STATUS_FAILED) @override_settings(MPESA_SANDBOX=False) def test_live_underpayment_is_not_marked_paid(self): payment = self.make_payment(status=PosMpesaPayment.STATUS_PENDING) self.client.post(CALLBACK_URL, success_callback("ws_POS_1", amount=100), content_type="application/json") payment.refresh_from_db() self.assertEqual(payment.status, PosMpesaPayment.STATUS_PENDING) def test_late_receipt_reaches_completed_sale(self): payment = self.make_payment() payment.mpesa_receipt_number = "" payment.save() self.complete_sale(payment.pk) sale = Sale.objects.get() self.assertEqual(sale.mpesa_code, "") self.client.post(CALLBACK_URL, success_callback("ws_POS_1", receipt="LATE123"), content_type="application/json") sale.refresh_from_db() self.assertEqual(sale.mpesa_code, "LATE123") class CompleteSaleTests(PosMpesaTestBase): def test_paid_prompt_completes_sale_with_receipt(self): payment = self.make_payment() resp = self.complete_sale(payment.pk) sale = Sale.objects.get() self.assertRedirects(resp, reverse("admin:inventory_sale_receipt", args=[sale.pk]), fetch_redirect_response=False) self.assertEqual(sale.mpesa_code, "POSRCPT01") payment.refresh_from_db() self.assertEqual(payment.sale, sale) def test_unpaid_prompt_blocks_sale(self): payment = self.make_payment(status=PosMpesaPayment.STATUS_PENDING) resp = self.complete_sale(payment.pk) self.assertEqual(resp.status_code, 200) self.assertContains(resp, "hasn") self.assertFalse(Sale.objects.exists()) refused = SystemLog.objects.get(event="pos.sale.refused") self.assertEqual(refused.reference, f"POS-{payment.pk}") self.assertEqual(refused.user, self.user) self.assertEqual(refused.request_data["cart"][0]["bookId"], self.book.pk) def test_payment_smaller_than_cart_blocks_sale(self): payment = self.make_payment(amount="1500.00") self.complete_sale(payment.pk, qty=2) self.assertFalse(Sale.objects.exists()) payment.refresh_from_db() self.assertIsNone(payment.sale) def test_payment_cannot_back_two_sales(self): payment = self.make_payment() self.complete_sale(payment.pk) self.complete_sale(payment.pk) self.assertEqual(Sale.objects.count(), 1)