Pesapal API 3.0 in Django: Payments, IPN and Transaction Status

This guide accepts payments through Pesapal API 3.0 from a Django app. Pesapal hosts the payment page, so your customer chooses how to pay there (Pesapal’s status responses name methods such as M-Pesa, MTN, Tigo and cards). You’ll build:

• a client.py: a cached token, IPN registration, orders and transaction status,
• a PesapalPayment model and a services.py that saves each order before sending it, settles only from GetTransactionStatus, and never settles a payment twice,
• views for starting a payment, the customer’s return and the IPN,
• management commands to register the IPN and to reconcile pending payments,
• tests that run every case without calling Pesapal.

Every code block was run in fresh Django 5.2 and 6.1 projects, and the tests at the end pass. API facts come from Pesapal’s API 3.0 documentation, checked in November 2026. The Node.js version of this guide has the same design and explains each step in more detail.

How it works

1. Token: POST /api/Auth/RequestToken with your consumer key and secret. It lasts at most 5 minutes.
2. Register your IPN URL once per environment with POST /api/URLSetup/RegisterIPN, and keep the ipn_id.
3. Submit an order with POST /api/Transactions/SubmitOrderRequest. Pesapal answers with an order_tracking_id and a redirect_url; send the customer there.
4. Pesapal sends the customer back to your callback_url and calls your IPN URL, both with OrderTrackingId, OrderMerchantReference and OrderNotificationType. Neither includes the status.
5. Check the status with GET /api/Transactions/GetTransactionStatus?orderTrackingId=...: status_code 1 COMPLETED, 2 FAILED, 3 REVERSED, 0 INVALID.

new ──start──▶ pending ──status 1, same amount & currency──▶ paid ──status 3──▶ reversed
                  ├──status 1, other amount or currency──▶ review
                  ├──status 2──▶ failed
                  └──no completed payment after an hour──▶ expired

Before you start

1. Get sandbox keys: Pesapal publishes demo keys, linked from the authentication page of its developer docs. Live keys are emailed when your Pesapal business account opens.
2. Your IPN URL must be public. For local testing, use a tunnel.
3. Install requests, create the app and set the keys in your environment. Never commit them:

pip install "django>=5.2" requests
python manage.py startapp pesapal
mkdir -p pesapal/management/commands
touch pesapal/management/__init__.py pesapal/management/commands/__init__.py
export PESAPAL_CONSUMER_KEY=your-consumer-key
export PESAPAL_CONSUMER_SECRET=your-consumer-secret

Add "pesapal" to INSTALLED_APPS, and the settings:

# settings.py (add to your existing settings)
import os

PESAPAL = {
    "BASE_URL": os.environ.get("PESAPAL_BASE_URL", "https://cybqa.pesapal.com/pesapalv3"),  # live: https://pay.pesapal.com/v3
    "CONSUMER_KEY": os.environ.get("PESAPAL_CONSUMER_KEY", ""),
    "CONSUMER_SECRET": os.environ.get("PESAPAL_CONSUMER_SECRET", ""),
    "IPN_ID": os.environ.get("PESAPAL_IPN_ID", ""),
    "CALLBACK_URL": os.environ.get("PESAPAL_CALLBACK_URL", "https://your-app.com/pesapal/callback/"),
}

The model

# pesapal/models.py
import uuid

from django.db import models


def new_reference():
    # Pesapal's "id": letters, numbers, - _ . : only, at most 50 characters, unique for every order request.
    return f"pay-{uuid.uuid4()}"


class PesapalPayment(models.Model):
    # Final: nothing changes them again. paid is not final: Pesapal can still report it REVERSED.
    FINAL = ("failed", "review", "reversed", "expired")

    reference = models.CharField(max_length=50, unique=True, default=new_reference, editable=False)
    order_id = models.CharField(max_length=64)  # your own order
    amount_cents = models.PositiveBigIntegerField()  # 1500.50 is 150050, so amounts compare exactly
    currency = models.CharField(max_length=3)
    email = models.CharField(max_length=254, blank=True)
    phone = models.CharField(max_length=20, blank=True)
    description = models.CharField(max_length=100)
    status = models.CharField(max_length=10, default="new")  # new | pending | paid | failed | review | reversed | expired
    tracking_id = models.CharField(max_length=64, unique=True, null=True, blank=True)  # Pesapal's order_tracking_id
    note = models.CharField(max_length=200, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        indexes = [models.Index(fields=["status", "created_at"])]

Run python manage.py makemigrations pesapal and python manage.py migrate.

The Pesapal client

# pesapal/client.py
from datetime import datetime, timedelta, timezone
from urllib.parse import quote

import requests
from django.conf import settings
from django.core.cache import cache


class PesapalError(Exception):
    """ambiguous=True: the request may have worked. Look it up before doing anything else."""

    def __init__(self, message, status=0, ambiguous=False, error=None):
        super().__init__(message)
        self.status = status
        self.ambiguous = ambiguous
        self.error = error or {}


def call(method, path, body=None, auth=True):
    headers = {"Accept": "application/json", "Content-Type": "application/json"}
    if auth:
        headers["Authorization"] = f"Bearer {token()}"
    try:
        res = requests.request(method, settings.PESAPAL["BASE_URL"].rstrip("/") + path, json=body, headers=headers, timeout=30)
    except requests.RequestException as exc:
        raise PesapalError(f"Pesapal unreachable: {exc}", ambiguous=True) from exc
    try:
        data = res.json()
    except ValueError:
        data = None
    if res.status_code >= 500 or not isinstance(data, dict):
        raise PesapalError(f"Pesapal HTTP {res.status_code}", res.status_code, ambiguous=True)
    # Successful answers can carry an error object whose fields are all null: only a code or message is an error.
    err = data.get("error")
    err = err if isinstance(err, dict) and (err.get("code") or err.get("message")) else None
    if not res.ok or err:
        code = (err or {}).get("code") or res.status_code
        raise PesapalError(f"Pesapal error: {code} {(err or {}).get('message') or data.get('message') or ''}".strip(), res.status_code, error=err)
    return data


def _expiry(value):
    try:
        when = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except ValueError:
        return datetime.now(timezone.utc) + timedelta(minutes=4)
    return when if when.tzinfo else when.replace(tzinfo=timezone.utc)  # Pesapal gives it in UTC


def token():
    """A bearer token, cached until 30 seconds before its expiryDate. Pesapal tokens last at most 5 minutes."""
    cached = cache.get("pesapal_token")
    if cached:
        return cached
    data = call("POST", "/api/Auth/RequestToken", {
        "consumer_key": settings.PESAPAL["CONSUMER_KEY"], "consumer_secret": settings.PESAPAL["CONSUMER_SECRET"],
    }, auth=False)
    if not isinstance(data.get("token"), str):
        raise PesapalError("No token in the response")
    now = datetime.now(timezone.utc)
    seconds = (min(_expiry(data.get("expiryDate")), now + timedelta(minutes=5)) - now).total_seconds() - 30
    if seconds > 0:
        cache.set("pesapal_token", data["token"], int(seconds))
    return data["token"]


def register_ipn(url, method="POST"):
    """Registers your IPN URL. Keep the ipn_id: it is the notification_id of every order in this environment."""
    return call("POST", "/api/URLSetup/RegisterIPN", {"url": url, "ipn_notification_type": method})["ipn_id"]


def submit_order(order):
    """Creates the order. Returns order_tracking_id, merchant_reference and redirect_url. Nothing is paid yet."""
    data = call("POST", "/api/Transactions/SubmitOrderRequest", order)
    if not data.get("order_tracking_id") or not data.get("redirect_url"):
        raise PesapalError("Order response without order_tracking_id or redirect_url")
    return data


def transaction_status(tracking_id):
    """The source of truth: status_code 0 INVALID, 1 COMPLETED, 2 FAILED, 3 REVERSED. Any failure here is ambiguous."""
    try:
        return call("GET", f"/api/Transactions/GetTransactionStatus?orderTrackingId={quote(tracking_id, safe='')}")
    except PesapalError as exc:
        exc.ambiguous = True
        raise

Settle each payment once

# pesapal/services.py
import re
from datetime import timedelta
from decimal import Decimal, InvalidOperation

from django.conf import settings
from django.utils import timezone

from . import client
from .models import PesapalPayment


def save(queryset, **fields):
    """queryset.update() skips auto_now, so set updated_at here: the reconcile command relies on it."""
    return queryset.update(updated_at=timezone.now(), **fields)


def fresh(payment):
    payment.refresh_from_db()
    return payment


def create(order_id, amount, currency, email, phone, description):
    """Create the payment record. Its reference is fixed for life and used for only one order request."""
    m = re.fullmatch(r"(\d{1,12})(?:\.(\d{1,2}))?", str(amount).strip())
    if not m:
        raise ValueError("Enter an amount like 1500 or 1500.50")
    cents = int(m.group(1)) * 100 + int((m.group(2) or "").ljust(2, "0"))
    if cents <= 0:
        raise ValueError("Amount must be more than zero")
    if not re.fullmatch(r"[A-Z]{3}", str(currency)):
        raise ValueError("Use an ISO currency code, e.g. KES, UGX or TZS")
    if not email and not phone:
        raise ValueError("Pesapal needs an email address or a phone number")
    return PesapalPayment.objects.create(order_id=str(order_id)[:64], amount_cents=cents, currency=currency,
                                         email=email or "", phone=phone or "", description=str(description)[:100])


def start(payment):
    """Saves the payment as pending, then creates the Pesapal order. Returns the page to send the customer to."""
    if save(PesapalPayment.objects.filter(pk=payment.pk, status="new"), status="pending") != 1:
        raise ValueError("This payment was already sent to Pesapal")  # a double click
    billing = {k: v for k, v in (("email_address", payment.email), ("phone_number", payment.phone)) if v}
    try:
        order = client.submit_order({
            "id": payment.reference,
            "currency": payment.currency,
            "amount": payment.amount_cents / 100,
            "description": payment.description,
            "callback_url": settings.PESAPAL["CALLBACK_URL"],
            "notification_id": settings.PESAPAL["IPN_ID"],
            "billing_address": billing,
        })
    except client.PesapalError as exc:
        if not exc.ambiguous:
            save(PesapalPayment.objects.filter(pk=payment.pk, status="pending", tracking_id__isnull=True),
                 status="failed", note=str(exc.error.get("code") or exc)[:200])
        # Ambiguous: the order may exist. The payment stays pending; its IPN or callback can still settle it.
        raise
    PesapalPayment.objects.filter(pk=payment.pk).update(tracking_id=order["order_tracking_id"])
    return order["redirect_url"]


def settle(payment, tx):
    """Applies a GetTransactionStatus answer. Every update names the status it expects, so a stale copy can't undo a result."""
    if tx.get("merchant_reference") != payment.reference:
        return fresh(payment)  # not this payment
    row = PesapalPayment.objects.filter(pk=payment.pk)
    try:
        code = int(tx.get("status_code"))
    except (TypeError, ValueError):
        code = -1
    if code == 3:  # a reversal only follows a payment
        save(row.filter(status="paid"), status="reversed", note="REVERSED")
    elif code == 1:
        try:
            same = Decimal(str(tx.get("amount"))) * 100 == payment.amount_cents and str(tx.get("currency") or "").upper() == payment.currency
        except InvalidOperation:
            same = False
        if same:
            save(row.filter(status="pending"), status="paid", note=str(tx.get("confirmation_code") or "")[:200])
        else:
            save(row.filter(status="pending"), status="review", note=f"PAID {tx.get('amount')} {tx.get('currency')}"[:200])
    elif code == 2:
        save(row.filter(status="pending"), status="failed", note=str(tx.get("description") or "FAILED")[:200])
    # 0 (INVALID) or anything else: not paid yet. reconcile() expires it eventually.
    return fresh(payment)


def handle_notification(tracking_id, reference):
    """
    The callback redirect or an IPN. Neither carries the status: they only say which order to look up,
    and the tracking id must be the one Pesapal gave us for this reference.
    """
    if not tracking_id or not reference:
        return None
    payment = PesapalPayment.objects.filter(reference=reference).first()
    if payment is None or (payment.tracking_id and payment.tracking_id != tracking_id):
        return None  # not ours, or not this payment's order
    tx = client.transaction_status(tracking_id)  # raises if Pesapal is unreachable
    if not payment.tracking_id and tx.get("merchant_reference") == payment.reference:
        PesapalPayment.objects.filter(pk=payment.pk, tracking_id__isnull=True).update(tracking_id=tracking_id)  # the order whose answer we lost
    return settle(payment, tx)


def reconcile(payment):
    """For payments still pending: asks Pesapal, and expires those that never completed."""
    if payment.status != "pending":
        return payment
    if payment.tracking_id:
        try:
            payment = settle(payment, client.transaction_status(payment.tracking_id))
        except client.PesapalError:
            return payment  # Pesapal unreachable: try again on the next run
    if payment.status == "pending" and payment.created_at < timezone.now() - timedelta(hours=1):
        save(PesapalPayment.objects.filter(pk=payment.pk, status="pending"), status="expired", note="NO_PAYMENT")
    return fresh(payment)

The rules:

1. Save before you send. start() moves the payment from new to pending in one conditional update before calling Pesapal, so a double click can’t create two orders.
2. Never settle from the callback or IPN. They only say which order to look up, and the tracking id must match the one Pesapal gave you.
3. Paid means status 1 with the same amount and currency. Anything else goes to review.
4. Final means final. failed, review, reversed and expired never change, and paid changes only to reversed.
5. Reconcile. A cron command checks pending payments and expires them after an hour.

The views

# pesapal/views.py
import json
import logging

from django.http import HttpResponse, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_GET, require_POST

from . import client, services

log = logging.getLogger(__name__)


@require_POST
def start_payment(request):
    """Starts a payment from your checkout page (with its CSRF token) and returns the Pesapal page to send the customer to."""
    try:
        payment = services.create(request.POST.get("order_id", ""), request.POST.get("amount", ""), request.POST.get("currency", "").upper(),
                                  request.POST.get("email", ""), None, f"Order {request.POST.get('order_id', '')}")
    except ValueError as exc:
        return JsonResponse({"error": str(exc)}, status=400)
    try:
        return JsonResponse({"reference": payment.reference, "redirect_url": services.start(payment)})
    except client.PesapalError:
        return JsonResponse({"error": "Could not start the payment. Please try again."}, status=502)


@require_GET
def callback(request):
    """The customer comes back here. Check the status before showing anything."""
    try:
        payment = services.handle_notification(request.GET.get("OrderTrackingId"), request.GET.get("OrderMerchantReference"))
    except client.PesapalError:
        payment = None
    if payment is not None and payment.status == "paid":
        return HttpResponse("Payment received. Thank you!")
    return HttpResponse("We are confirming your payment. This page will update shortly.")


def _ipn_params(request):
    if request.method == "GET":
        return request.GET
    try:
        body = json.loads(request.body or b"{}")
        if isinstance(body, dict):
            return body
    except ValueError:
        pass
    return request.POST


@csrf_exempt  # Pesapal can't send a CSRF token; nothing in the IPN is trusted anyway
def ipn(request):
    """The IPN: Pesapal calls it, with the method you registered, when a payment status changes."""
    params = _ipn_params(request)
    tracking, reference = params.get("OrderTrackingId"), params.get("OrderMerchantReference")
    echo = {"orderNotificationType": params.get("OrderNotificationType"), "orderTrackingId": tracking, "orderMerchantReference": reference}
    try:
        services.handle_notification(tracking if isinstance(tracking, str) else None, reference if isinstance(reference, str) else None)
    except client.PesapalError as exc:
        log.warning("Pesapal IPN not processed for %s: %s", reference, exc)
        return JsonResponse({**echo, "status": 500}, status=500)
    return JsonResponse({**echo, "status": 200})

# pesapal/urls.py
from django.urls import path

from . import views

urlpatterns = [
    path("pesapal/pay/", views.start_payment),
    path("pesapal/callback/", views.callback),
    path("pesapal/ipn/", views.ipn),
]

Include them in your project’s urls.py with path("", include("pesapal.urls")), and set PESAPAL_CALLBACK_URL to your public callback URL.

The IPN answer. Pesapal’s API 3.0 pages say what it sends to your IPN URL but not what your server must answer. This view answers HTTP 200 with the three values it received and status: 200 once the payment is settled, and 500 if it couldn’t ask Pesapal. Confirm in the sandbox that Pesapal accepts it, and keep the reconcile command running: it settles any payment whose IPN went wrong.

Register the IPN and reconcile

# pesapal/management/commands/register_pesapal_ipn.py
from django.core.management.base import BaseCommand

from pesapal import client


class Command(BaseCommand):
    help = "Register the Pesapal IPN URL (once per environment) and print its ipn_id."

    def add_arguments(self, parser):
        parser.add_argument("url")
        parser.add_argument("--method", default="POST", choices=["GET", "POST"])

    def handle(self, *args, **options):
        self.stdout.write(f"PESAPAL_IPN_ID={client.register_ipn(options['url'], options['method'])}")

# pesapal/management/commands/reconcile_pesapal.py
from datetime import timedelta

from django.core.management.base import BaseCommand
from django.utils import timezone

from pesapal import services
from pesapal.models import PesapalPayment


class Command(BaseCommand):
    help = "Check pending Pesapal payments and expire the ones that never completed. Run it every few minutes."

    def handle(self, *args, **options):
        due = PesapalPayment.objects.filter(status="pending", updated_at__lte=timezone.now() - timedelta(minutes=1)).order_by("id")[:100]
        counts = {}
        for payment in due:
            status = services.reconcile(payment).status
            counts[status] = counts.get(status, 0) + 1
        self.stdout.write(", ".join(f"{n} {s}" for s, n in counts.items()) or "Nothing to check.")

Run python manage.py register_pesapal_ipn https://your-app.com/pesapal/ipn/ once in each environment and put the printed value in PESAPAL_IPN_ID. Run the reconcile command from cron: */5 * * * * cd /srv/app && .venv/bin/python manage.py reconcile_pesapal.

Test it without calling Pesapal

These tests patch requests with unittest.mock, including the all-null error object Pesapal’s status answers carry. They check that a payment is saved before the order is sent and sent only once, that the callback and IPN are never trusted, that another order’s tracking id can’t settle a payment, that a different amount goes to review, that a reversal only follows a payment, that final results never change, and that the token is cached.

# pesapal/tests.py
import json as jsonlib
from datetime import timedelta
from io import StringIO
from unittest import mock

from django.core.cache import cache
from django.core.management import call_command
from django.test import Client, TestCase, override_settings
from django.utils import timezone

from . import client, services
from .models import PesapalPayment

IPN_ID = "e7b1f5a6-1111-4d2c-9e3b-0d5c7e8a9f10"
SETTINGS = {"BASE_URL": "https://pesapal.test/v3", "CONSUMER_KEY": "key", "CONSUMER_SECRET": "secret", "IPN_ID": IPN_ID,
            "CALLBACK_URL": "https://shop.example.com/pesapal/callback/"}
NULL_ERROR = {"error_type": None, "code": None, "message": None}


def reply(status, data=None):
    res = mock.Mock(status_code=status, ok=200 <= status < 300)
    if data is None:
        res.json.side_effect = ValueError("no JSON")
    else:
        res.json.return_value = data
    return res


@override_settings(PESAPAL=SETTINGS)
class PesapalTests(TestCase):
    def setUp(self):
        cache.clear()
        self.orders = {}  # by tracking id, the way Pesapal keeps them
        self.tokens = 0
        self.submits = []  # each SubmitOrderRequest, with our payment's status at that moment
        self.next = "ok"  # next SubmitOrderRequest: ok, refused or timeout
        self.down = False
        patcher = mock.patch("pesapal.client.requests.request", side_effect=self.fake)
        patcher.start()
        self.addCleanup(patcher.stop)

    def fake(self, method, url, json=None, headers=None, timeout=None):
        path = url.removeprefix("https://pesapal.test/v3")
        if path == "/api/Auth/RequestToken":
            self.tokens += 1
            expiry = (timezone.now() + timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%S.%f")  # UTC, without an offset
            return reply(200, {"token": f"tok-{self.tokens}", "expiryDate": expiry, "error": None, "status": "200"})
        if headers.get("Authorization") != f"Bearer tok-{self.tokens}":
            return reply(401, {"error": {"type": "error", "code": "invalid_token", "message": "Unauthorized"}})
        if path == "/api/URLSetup/RegisterIPN":
            return reply(200, {"url": json["url"], "ipn_id": "ipn-guid-1", "error": None, "status": "200"})
        if path == "/api/Transactions/SubmitOrderRequest":
            self.submits.append({"order": json, "status": PesapalPayment.objects.get(reference=json["id"]).status})
            mode, self.next = self.next, "ok"
            if mode == "refused":
                return reply(200, {"error": {"type": "error", "code": "invalid_notification_id", "message": "Invalid notification id"}, "status": "500"})
            tracking = f"7e6b62d9-883e-440f-a63e-{len(self.orders) + 1:012d}"
            self.orders[tracking] = {"merchant_reference": json["id"], "amount": json["amount"], "currency": json["currency"],
                                     "status_code": 0, "payment_status_description": "INVALID"}
            if mode == "timeout":
                return reply(504)
            return reply(200, {"order_tracking_id": tracking, "merchant_reference": json["id"],
                               "redirect_url": f"https://cybqa.pesapal.com/iframe?OrderTrackingId={tracking}", "error": None, "status": "200"})
        if path.startswith("/api/Transactions/GetTransactionStatus"):
            if self.down:
                return reply(503)
            tx = self.orders.get(path.split("orderTrackingId=")[1])
            if not tx:
                return reply(200, {"error": {"error_type": "api_error", "code": "order_not_found", "message": "Order not found"}, "status": "500"})
            return reply(200, {**tx, "error": NULL_ERROR, "status": "200"})
        raise AssertionError(f"unexpected {method} {url}")

    def payment(self, amount="1500"):
        return services.create("A1001", amount, "KES", "jane@example.com", None, "Order A1001")

    def started(self, amount="1500"):
        p = self.payment(amount)
        services.start(p)
        return PesapalPayment.objects.get(pk=p.pk)

    def pay(self, tracking, **extra):
        self.orders[tracking].update(status_code=1, payment_status_description="Completed", confirmation_code="RKL12AB3CD", **extra)

    def ipn(self, p, tracking=None):
        body = {"OrderTrackingId": tracking or p.tracking_id, "OrderMerchantReference": p.reference, "OrderNotificationType": "IPNCHANGE"}
        return self.client.post("/pesapal/ipn/", jsonlib.dumps(body), content_type="application/json")

    def get(self, p):
        return PesapalPayment.objects.get(pk=p.pk)

    def test_saved_before_sending_sent_once_and_paid_only_from_the_status_check(self):
        p = self.payment()
        self.assertTrue(services.start(p).startswith("https://cybqa.pesapal.com/"))
        self.assertEqual([s["status"] for s in self.submits], ["pending"])
        with self.assertRaises(ValueError):
            services.start(p)  # a double click
        self.assertEqual(self.submits[0]["order"], {
            "id": p.reference, "currency": "KES", "amount": 1500.0, "description": "Order A1001",
            "callback_url": SETTINGS["CALLBACK_URL"], "notification_id": IPN_ID, "billing_address": {"email_address": "jane@example.com"},
        })
        p = self.get(p)
        res = self.ipn(p)
        self.assertEqual(res.json(), {"orderNotificationType": "IPNCHANGE", "orderTrackingId": p.tracking_id, "orderMerchantReference": p.reference, "status": 200})
        self.assertEqual(self.get(p).status, "pending")  # INVALID: not paid
        self.pay(p.tracking_id)
        res = self.client.get("/pesapal/callback/", {"OrderTrackingId": p.tracking_id, "OrderMerchantReference": p.reference, "OrderNotificationType": "CALLBACKURL"})
        self.assertContains(res, "Payment received")
        self.assertEqual((self.get(p).status, self.get(p).note), ("paid", "RKL12AB3CD"))

    def test_the_ipn_works_as_get_and_form_post_too(self):
        p = self.started()
        self.pay(p.tracking_id)
        self.client.get("/pesapal/ipn/", {"OrderTrackingId": p.tracking_id, "OrderMerchantReference": p.reference, "OrderNotificationType": "IPNCHANGE"})
        self.assertEqual(self.get(p).status, "paid")
        q = self.started()
        self.pay(q.tracking_id)
        strict = Client(enforce_csrf_checks=True)  # like Pesapal: no CSRF token
        res = strict.post("/pesapal/ipn/", {"OrderTrackingId": q.tracking_id, "OrderMerchantReference": q.reference})
        self.assertEqual((res.status_code, self.get(q).status), (200, "paid"))

    def test_the_callback_and_ipn_are_never_trusted(self):
        mine = self.started()
        other = self.started("10")
        self.pay(other.tracking_id, amount=10)
        self.ipn(mine, other.tracking_id)  # our reference, another order's tracking id
        self.assertEqual(self.get(mine).status, "pending")
        self.client.post("/pesapal/ipn/", jsonlib.dumps({"OrderTrackingId": mine.tracking_id, "OrderMerchantReference": "not-ours"}), content_type="application/json")
        self.client.post("/pesapal/ipn/", jsonlib.dumps({"OrderMerchantReference": mine.reference}), content_type="application/json")
        self.assertEqual(self.get(mine).status, "pending")
        self.assertEqual(services.settle(mine, {"merchant_reference": other.reference, "status_code": 1, "amount": 1500, "currency": "KES"}).status, "pending")

    def test_a_different_amount_or_currency_goes_to_review_and_final_means_final(self):
        p = self.started()
        self.pay(p.tracking_id, amount=150)
        self.ipn(p)
        self.assertEqual((self.get(p).status, self.get(p).note), ("review", "PAID 150 KES"))
        q = self.started()
        self.pay(q.tracking_id, currency="UGX")
        self.assertEqual(services.reconcile(q).status, "review")
        self.orders[q.tracking_id]["currency"] = "KES"
        self.assertEqual(services.settle(q, self.orders[q.tracking_id]).status, "review")

        f = self.started()
        self.orders[f.tracking_id].update(status_code=2, description="Insufficient funds")
        failed = services.reconcile(f)
        self.assertEqual((failed.status, failed.note), ("failed", "Insufficient funds"))
        self.pay(f.tracking_id)
        self.assertEqual(services.settle(f, self.orders[f.tracking_id]).status, "failed")

    def test_only_a_reversal_changes_a_paid_payment_and_a_reversal_needs_a_payment_first(self):
        p = self.started()
        self.orders[p.tracking_id]["status_code"] = 3
        self.assertEqual(services.reconcile(p).status, "pending")
        self.pay(p.tracking_id)
        paid = services.reconcile(self.get(p))
        self.assertEqual(paid.status, "paid")
        stale = self.get(p)
        stale.status = "pending"  # an old copy
        self.orders[p.tracking_id]["status_code"] = 2
        self.assertEqual(services.settle(stale, self.orders[p.tracking_id]).status, "paid")
        self.orders[p.tracking_id]["status_code"] = 3
        self.ipn(p)
        self.assertEqual(self.get(p).status, "reversed")

    def test_refused_lost_and_unreachable(self):
        self.next = "refused"
        res = self.client.post("/pesapal/pay/", {"order_id": "A1", "amount": "1500", "currency": "KES", "email": "jane@example.com"})
        self.assertEqual(res.status_code, 502)
        self.assertEqual((PesapalPayment.objects.get().status, PesapalPayment.objects.get().note), ("failed", "invalid_notification_id"))

        self.next = "timeout"
        lost = self.payment()
        with self.assertRaises(client.PesapalError) as ctx:
            services.start(lost)
        self.assertTrue(ctx.exception.ambiguous)
        self.assertEqual((self.get(lost).status, self.get(lost).tracking_id), ("pending", None))
        tracking = list(self.orders)[-1]  # Pesapal did create it: the IPN brings the tracking id
        self.pay(tracking)
        self.ipn(self.get(lost), tracking)
        self.assertEqual((self.get(lost).status, self.get(lost).tracking_id), ("paid", tracking))

        p = self.started()
        self.down = True
        self.assertEqual(services.reconcile(p).status, "pending")
        res = self.ipn(p)
        self.assertEqual((res.status_code, res.json()["status"]), (500, 500))

    def test_reconcile_expires_old_payments_and_the_commands_run(self):
        p = self.started()
        PesapalPayment.objects.update(created_at=timezone.now() - timedelta(hours=2), updated_at=timezone.now() - timedelta(hours=2))
        call_command("reconcile_pesapal", stdout=StringIO())
        self.assertEqual(self.get(p).status, "expired")
        self.pay(p.tracking_id)
        self.ipn(p)
        self.assertEqual(self.get(p).status, "expired")
        out = StringIO()
        call_command("register_pesapal_ipn", "https://shop.example.com/pesapal/ipn/", stdout=out)
        self.assertEqual(out.getvalue().strip(), "PESAPAL_IPN_ID=ipn-guid-1")

    def test_the_token_is_cached_and_bad_input_is_refused(self):
        self.started()
        self.started()
        self.assertEqual(self.tokens, 1)
        cache.clear()  # as if it expired
        self.started()
        self.assertEqual(self.tokens, 2)
        for amount, currency, email in [("1.234", "KES", "a@b.c"), ("0", "KES", "a@b.c"), ("5", "kes", "a@b.c"), ("5", "KES", None)]:
            with self.assertRaises(ValueError):
                services.create("x", amount, currency, email, None, "x")
        self.assertRegex(self.payment().reference, r"^[A-Za-z0-9\-_.:]{1,50}$")

python manage.py test pesapal

Common errors and fixes

Every Pesapal status and the error format are explained in the Pesapal error reference.

Going to production

• Set PESAPAL_BASE_URL to https://pay.pesapal.com/v3, use your live consumer key and secret, and register the IPN on the live site.
• Use a cache all your processes share for the token, and keep reconcile_pesapal running.
• Send one small real payment end to end before launch.

Want this done for you, alongside MTN MoMo, M-Pesa, Airtel Money, Flutterwave and Paystack? The Django starter kit includes Pesapal.

Using Laravel? See Pesapal API 3.0 in Laravel.

Using the Next.js App Router? See Pesapal API 3.0 in Next.js: the same design in route handlers.