""" store/tests/test_mpesa.py M-Pesa STK flow: callback idempotency, resend/retry, status query, stale sweep. Daraja HTTP calls are patched — nothing leaves the test process. """ from datetime import timedelta from decimal import Decimal from io import StringIO from unittest.mock import MagicMock, patch from django.core.management import call_command from django.test import override_settings from django.utils import timezone from rest_framework.test import APITestCase from catalog_app.models import Author, Book, Catalog from store.models import MpesaTransaction, Order, OrderItem from store.services import mpesa_service from store.services.mpesa_service import ( QUERY_FAILED, QUERY_PROCESSING, QUERY_SUCCESS, MpesaService, ) PUSH_URL = "/api/payments/mpesa/stk-push/" CALLBACK_URL = "/api/payments/mpesa/callback/" QUERY_URL = "/api/payments/mpesa/query/" 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/", ) def success_callback(checkout_id, amount=1000, receipt="RCP123ABC"): 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": 254708374149}, ]}, }}} def failure_callback(checkout_id, code=1032, desc="Request cancelled by user."): return {"Body": {"stkCallback": { "MerchantRequestID": "m-1", "CheckoutRequestID": checkout_id, "ResultCode": code, "ResultDesc": desc, }}} def query_result(state, code=None, desc=""): return {"state": state, "result_code": code, "result_desc": desc, "raw": {"test": True}} def push_result(checkout_id): return { "MerchantRequestID": "m-new", "CheckoutRequestID": checkout_id, "ResponseCode": "0", "ResponseDescription": "Success. Request accepted for processing", } @override_settings(**MPESA_SETTINGS) class MpesaFlowTestBase(APITestCase): def setUp(self): author = Author.objects.create(name="Test Author") catalog = Catalog.objects.create(name="Test Catalog", slug="test-catalog") # Stock after checkout already reserved 2 of the original 10. self.book = Book.objects.create( catalog=catalog, title="Test Book", author=author, price=Decimal("500.00"), stock=8, is_listed=True, ) self.order = Order.objects.create( total_amount=Decimal("1000.00"), final_amount=Decimal("1000.00"), customer_name="Test Customer", phone_number="254708374149", status=Order.STATUS_PAYMENT_INITIATED, checkout_request_id="ws_CO_1", merchant_request_id="m-1", ) OrderItem.objects.create( order=self.order, book=self.book, unit_price=Decimal("500.00"), quantity=2, total_price=Decimal("1000.00"), book_snapshot={"title": "Test Book"}, ) MpesaTransaction.objects.create( order=self.order, checkout_request_id="ws_CO_1", merchant_request_id="m-1", phone_number="254708374149", amount=Decimal("1000.00"), ) def age_prompt(self, seconds): past = timezone.now() - timedelta(seconds=seconds) MpesaTransaction.objects.filter(order=self.order).update(created_at=past) Order.objects.filter(pk=self.order.pk).update(updated_at=past) def stock(self): self.book.refresh_from_db() return self.book.stock def status(self): self.order.refresh_from_db() return self.order.status class CallbackTests(MpesaFlowTestBase): def test_success_marks_paid_with_receipt(self): resp = self.client.post(CALLBACK_URL, success_callback("ws_CO_1"), format="json") self.assertEqual(resp.data["ResultCode"], 0) self.assertEqual(self.status(), Order.STATUS_PAID) self.assertEqual(self.order.mpesa_receipt_number, "RCP123ABC") self.assertEqual(self.stock(), 8) def test_duplicate_success_is_harmless(self): self.client.post(CALLBACK_URL, success_callback("ws_CO_1"), format="json") self.client.post(CALLBACK_URL, success_callback("ws_CO_1"), format="json") self.assertEqual(self.status(), Order.STATUS_PAID) self.assertEqual(self.stock(), 8) def test_failure_releases_stock_once_even_if_repeated(self): self.client.post(CALLBACK_URL, failure_callback("ws_CO_1"), format="json") self.client.post(CALLBACK_URL, failure_callback("ws_CO_1"), format="json") self.assertEqual(self.status(), Order.STATUS_FAILED) self.assertEqual(self.stock(), 10) def test_failure_for_superseded_prompt_is_ignored(self): Order.objects.filter(pk=self.order.pk).update(checkout_request_id="ws_CO_2") MpesaTransaction.objects.create( order=self.order, checkout_request_id="ws_CO_2", phone_number="254708374149", amount=Decimal("1000.00"), ) self.client.post(CALLBACK_URL, failure_callback("ws_CO_1"), format="json") self.assertEqual(self.status(), Order.STATUS_PAYMENT_INITIATED) self.assertEqual(self.stock(), 8) def test_late_success_after_failure_marks_paid_and_retakes_stock(self): self.client.post(CALLBACK_URL, failure_callback("ws_CO_1", 1037, "Timeout"), format="json") self.assertEqual(self.stock(), 10) self.client.post(CALLBACK_URL, success_callback("ws_CO_1"), format="json") self.assertEqual(self.status(), Order.STATUS_PAID) self.assertEqual(self.stock(), 8) def test_success_on_superseded_prompt_still_marks_paid(self): Order.objects.filter(pk=self.order.pk).update(checkout_request_id="ws_CO_2") self.client.post(CALLBACK_URL, success_callback("ws_CO_1"), format="json") self.assertEqual(self.status(), Order.STATUS_PAID) @override_settings(MPESA_SANDBOX=False) def test_underpayment_is_not_marked_paid(self): self.client.post(CALLBACK_URL, success_callback("ws_CO_1", amount=10), format="json") self.assertEqual(self.status(), Order.STATUS_PAYMENT_INITIATED) def test_sandbox_one_shilling_payment_marks_paid_and_logs_real_amount(self): self.client.post(CALLBACK_URL, success_callback("ws_CO_1", amount=1), format="json") self.assertEqual(self.status(), Order.STATUS_PAID) self.assertEqual(self.order.final_amount, Decimal("1000.00")) logged = MpesaTransaction.objects.filter(order=self.order, result_code=0).get() self.assertEqual(logged.amount, Decimal("1000.00")) def test_unknown_checkout_is_acknowledged(self): resp = self.client.post(CALLBACK_URL, success_callback("ws_CO_unknown"), format="json") self.assertEqual(resp.status_code, 200) self.assertEqual(self.status(), Order.STATUS_PAYMENT_INITIATED) def test_query_confirmed_payment_gets_receipt_from_late_callback(self): with patch.object(MpesaService, "stk_query", return_value=query_result(QUERY_SUCCESS, 0)): self.age_prompt(60) self.client.post(QUERY_URL, {"order_reference": self.order.order_reference}, format="json") self.assertEqual(self.status(), Order.STATUS_PAID) self.assertEqual(self.order.mpesa_receipt_number, "") self.client.post(CALLBACK_URL, success_callback("ws_CO_1"), format="json") self.order.refresh_from_db() self.assertEqual(self.order.mpesa_receipt_number, "RCP123ABC") class ResendTests(MpesaFlowTestBase): def push(self): return self.client.post( PUSH_URL, {"order_reference": self.order.order_reference, "phone_number": "0708374149"}, format="json", ) @patch.object(MpesaService, "stk_push") @patch.object(MpesaService, "stk_query", return_value=query_result(QUERY_PROCESSING)) def test_open_prompt_blocks_resend(self, _query, push): resp = self.push() self.assertEqual(resp.status_code, 409) push.assert_not_called() @patch.object(MpesaService, "stk_push", return_value=push_result("ws_CO_2")) @patch.object(MpesaService, "stk_query", return_value=query_result(QUERY_FAILED, 1037, "Timeout")) def test_expired_prompt_is_replaced(self, _query, push): resp = self.push() self.assertEqual(resp.status_code, 200) push.assert_called_once() self.assertEqual(self.status(), Order.STATUS_PAYMENT_INITIATED) self.assertEqual(self.order.checkout_request_id, "ws_CO_2") # Released on failure, re-reserved for the new prompt. self.assertEqual(self.stock(), 8) @patch.object(MpesaService, "stk_push") @patch.object(MpesaService, "stk_query", return_value=query_result(QUERY_SUCCESS, 0)) def test_already_paid_prompt_is_not_resent(self, _query, push): resp = self.push() self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data["status"], Order.STATUS_PAID) push.assert_not_called() @patch.object(MpesaService, "stk_push", return_value=push_result("ws_CO_2")) def test_retry_after_failure_reserves_stock(self, push): self.client.post(CALLBACK_URL, failure_callback("ws_CO_1"), format="json") self.assertEqual(self.stock(), 10) resp = self.push() self.assertEqual(resp.status_code, 200) self.assertEqual(self.stock(), 8) self.assertEqual(self.status(), Order.STATUS_PAYMENT_INITIATED) @patch.object(MpesaService, "stk_push") def test_retry_after_failure_without_stock_is_refused(self, push): self.client.post(CALLBACK_URL, failure_callback("ws_CO_1"), format="json") Book.objects.filter(pk=self.book.pk).update(stock=1) resp = self.push() self.assertEqual(resp.status_code, 409) push.assert_not_called() self.assertEqual(self.status(), Order.STATUS_FAILED) self.assertEqual(self.stock(), 1) @patch.object(MpesaService, "stk_push") def test_cancelling_failed_order_does_not_release_stock_again(self, push): self.client.post(CALLBACK_URL, failure_callback("ws_CO_1"), format="json") self.order.refresh_from_db() self.order.status = Order.STATUS_CANCELLED self.order.save() self.assertEqual(self.stock(), 10) @override_settings(MPESA_STK_ENABLED=False) class StkSwitchedOffTests(MpesaFlowTestBase): @patch.object(MpesaService, "stk_push") def test_store_prompt_endpoint_refuses(self, push): Order.objects.filter(pk=self.order.pk).update(status=Order.STATUS_PENDING, checkout_request_id="") resp = self.client.post( PUSH_URL, {"order_reference": self.order.order_reference, "phone_number": "0708374149"}, format="json", ) self.assertEqual(resp.status_code, 403) push.assert_not_called() self.assertEqual(self.status(), Order.STATUS_PENDING) class QueryEndpointTests(MpesaFlowTestBase): def query(self): return self.client.post(QUERY_URL, {"order_reference": self.order.order_reference}, format="json") @patch.object(MpesaService, "stk_query") def test_fresh_prompt_is_not_queried(self, stk_query): resp = self.query() self.assertEqual(resp.data["state"], QUERY_PROCESSING) stk_query.assert_not_called() @patch.object(MpesaService, "stk_query", return_value=query_result(QUERY_FAILED, 1032, "Cancelled")) def test_old_prompt_is_queried_and_applied(self, _query): self.age_prompt(60) resp = self.query() self.assertEqual(resp.data["status"], Order.STATUS_FAILED) self.assertEqual(self.stock(), 10) @patch.object(MpesaService, "stk_query") def test_non_initiated_order_is_not_queried(self, stk_query): Order.objects.filter(pk=self.order.pk).update(status=Order.STATUS_PAID) resp = self.query() self.assertEqual(resp.data["status"], Order.STATUS_PAID) stk_query.assert_not_called() class ExpireCommandTests(MpesaFlowTestBase): @patch.object(MpesaService, "stk_query", return_value=query_result(QUERY_PROCESSING)) def test_stale_unanswered_prompt_is_failed(self, _query): self.age_prompt(15 * 60) call_command("expire_mpesa_payments", stdout=StringIO()) self.assertEqual(self.status(), Order.STATUS_FAILED) self.assertEqual(self.stock(), 10) @patch.object(MpesaService, "stk_query") def test_recent_prompt_is_left_alone(self, stk_query): call_command("expire_mpesa_payments", stdout=StringIO()) stk_query.assert_not_called() self.assertEqual(self.status(), Order.STATUS_PAYMENT_INITIATED) @override_settings(**MPESA_SETTINGS) class ChargeAmountTests(APITestCase): def sent_amount(self): response = MagicMock(status_code=200, ok=True) response.json.return_value = push_result("ws_CO_9") | {"ResponseCode": "0"} with patch.object(MpesaService, "_get_access_token", return_value="token"), \ patch("store.services.mpesa_service.requests.post", return_value=response) as post: MpesaService().stk_push("254713621589", 1500, "ORD-TEST") return post.call_args.kwargs["json"]["Amount"] def test_sandbox_charges_one_shilling(self): self.assertEqual(self.sent_amount(), 1) @override_settings(MPESA_SANDBOX=False) def test_live_charges_full_amount(self): self.assertEqual(self.sent_amount(), 1500) @override_settings(**MPESA_SETTINGS) class AccessTokenCacheTests(APITestCase): def setUp(self): mpesa_service._TOKEN_CACHE.clear() def tearDown(self): mpesa_service._TOKEN_CACHE.clear() def test_token_is_reused_between_calls(self): response = MagicMock(status_code=200) response.json.return_value = {"access_token": "tok-1", "expires_in": "3599"} with patch("store.services.mpesa_service.requests.get", return_value=response) as get: self.assertEqual(MpesaService()._get_access_token(), "tok-1") self.assertEqual(MpesaService()._get_access_token(), "tok-1") get.assert_called_once() @override_settings(**MPESA_SETTINGS) class StkQueryParsingTests(APITestCase): def run_query(self, http_status, body): response = MagicMock(status_code=http_status, ok=http_status < 400) response.json.return_value = body with patch.object(MpesaService, "_get_access_token", return_value="token"), \ patch("store.services.mpesa_service.requests.post", return_value=response): return MpesaService().stk_query("ws_CO_1") def test_still_processing_error_body(self): result = self.run_query(500, { "requestId": "1", "errorCode": "500.001.1001", "errorMessage": "The transaction is being processed", }) self.assertEqual(result["state"], QUERY_PROCESSING) def test_cancelled(self): result = self.run_query(200, {"ResponseCode": "0", "ResultCode": "1032", "ResultDesc": "Cancelled"}) self.assertEqual(result["state"], QUERY_FAILED) self.assertEqual(result["result_code"], 1032) def test_success(self): result = self.run_query(200, {"ResponseCode": "0", "ResultCode": "0", "ResultDesc": "OK"}) self.assertEqual(result["state"], QUERY_SUCCESS)