daraja-mock

Local test server for the Safaricom M-Pesa Daraja v3 API.

Test your M-Pesa integration without a Safaricom account, sandbox credentials, or internet connection. Configure scenarios to simulate user cancellation, insufficient funds, timeouts, and more — all from a single in-process server.

Install

pip install daraja-mock

Quickstart

from daraja_mock import DarajaMock, Scenario

mock = DarajaMock()

def test_stk_push_success():
    with mock.run() as base_url:
        # Point your MpesaClient at base_url instead of api.safaricom.co.ke
        response = requests.post(
            f"{base_url}/mpesa/stkpush/v1/processrequest",
            json={
                "BusinessShortCode": "174379",
                "Amount": 100,
                "PhoneNumber": "254712345678",
                "CallBackURL": "https://yourapp.com/callback",
                "AccountReference": "Order001",
                "TransactionDesc": "Payment",
            }
        )
    assert response.json()["ResponseCode"] == "0"
    assert mock.last_stk_checkout_id  # store this to query status later

def test_stk_push_user_cancels():
    # STK initiated OK, but user cancels on phone
    mock.queue_scenarios(Scenario.SUCCESS, Scenario.USER_CANCELLED)

    with mock.run() as base_url:
        init = requests.post(f"{base_url}/mpesa/stkpush/v1/processrequest", json={"Amount": 100})
        status = requests.post(f"{base_url}/mpesa/stkpushquery/v1/query", json={
            "CheckoutRequestID": init.json()["CheckoutRequestID"]
        })

    assert status.json()["ResultCode"] == "1032"  # user cancelled

Scenarios

# Single scenario — all calls use this
mock.set_scenario(Scenario.INSUFFICIENT_FUNDS)

# Queue — each call consumes one, then falls back to set_scenario
mock.queue_scenarios(Scenario.SUCCESS, Scenario.USER_CANCELLED, Scenario.TIMED_OUT)

Endpoints implemented

Callback simulation

For webhook-based flows, build a realistic callback payload and POST it to your handler:

# Simulate Safaricom posting to your callback URL
payload = mock.build_stk_callback(
    checkout_request_id="ws_CO_123",
    scenario=Scenario.USER_CANCELLED,
)

# POST to your FastAPI/Flask/Django handler
response = test_client.post("/mpesa/stk/callback", json=payload)
assert response.status_code == 200

Inspect calls

with mock.run() as base_url:
    # ... make calls ...
    pass

# After the context
assert len(mock.calls) == 2
assert mock.calls[0].endpoint == "/oauth/v1/generate"
assert mock.calls[1].body["Amount"] == 100

Standalone server

# Default port 8765
python -m daraja_mock

# Custom port
python -m daraja_mock --port 9000

Then point any HTTP client (Postman, curl, your app) at http://localhost:8765.

Use with mpesa-python

import pytest
from daraja_mock import DarajaMock, Scenario
from mpesa import MpesaClient  # github.com/gabrielmahia/mpesa-python

@pytest.fixture
def mpesa_client():
    mock = DarajaMock()
    with mock.run() as base_url:
        client = MpesaClient(
            consumer_key="test_key",
            consumer_secret="test_secret",
            shortcode="174379",
            passkey="test_passkey",
            base_url=base_url,
        )
        yield client, mock

def test_full_stk_flow(mpesa_client):
    client, mock = mpesa_client
    result = client.stk_push("0712345678", 100, "Order001")
    assert result.checkout_request_id == mock.last_stk_checkout_id

Design decisions

No external dependencies. The server runs on Python's stdlib HTTPServer. No FastAPI, no httpx, no pytest-asyncio. This means it works in any test environment without dependency conflicts.

Thread-safe context manager. Each mock.run() starts a server in a daemon thread and tears it down cleanly on exit. Multiple mocks can run concurrently on different ports.

Queue-based scenarios. Real M-Pesa flows have two steps (initiate + query). queue_scenarios lets you specify each step independently: SUCCESS initiation followed by USER_CANCELLED status.

Part of the nairobi-stack East Africa engineering ecosystem. Maintained by Gabriel Mahia. Kenya × USA.