feat(pos): mpesa related fixes & additions (#24306)

* fix: switching of mode of payments * feat: transaction limit for mpesa integration * feat: resend payment request if one fails * feat: make new request only for failed ones * fix: invalid amount for mpesa request * fix: payment successful message not shown * fix: url and business shortcode for live env * fix: duplicate items validation for pos invoices * fix: pos closing entry queued status * fix: peroid end date for amended pos closing
2026-05-27 00:44:45 +00:00 · 2021-02-11 14:06:15 +05:30
parent a46a83b476
commit 0681a480c2
12 changed files with 433 additions and 131 deletions
--- a/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_connector.py
+++ b/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_connector.py
@@ -5,7 +5,7 @@ import datetime

 class MpesaConnector():
 	def __init__(self, env="sandbox", app_key=None, app_secret=None, sandbox_url="https://sandbox.safaricom.co.ke",
-		live_url="https://safaricom.co.ke"):
+		live_url="https://api.safaricom.co.ke"):
 		"""Setup configuration for Mpesa connector and generate new access token."""
 		self.env = env
 		self.app_key = app_key
@@ -102,14 +102,14 @@ class MpesaConnector():
 			"BusinessShortCode": business_shortcode,
 			"Password": encoded.decode("utf-8"),
 			"Timestamp": time,
-			"TransactionType": "CustomerPayBillOnline",
 			"Amount": amount,
 			"PartyA": int(phone_number),
-			"PartyB": business_shortcode,
+			"PartyB": reference_code,
 			"PhoneNumber": int(phone_number),
 			"CallBackURL": callback_url,
 			"AccountReference": reference_code,
-			"TransactionDesc": description
+			"TransactionDesc": description,
+			"TransactionType": "CustomerPayBillOnline" if self.env == "sandbox" else "CustomerBuyGoodsOnline"
 		}
 		headers = {'Authorization': 'Bearer {0}'.format(self.authentication_token), 'Content-Type': "application/json"}

--- a/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_settings.json
+++ b/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_settings.json
@@ -11,8 +11,10 @@
  "consumer_secret",
  "initiator_name",
  "till_number",
+  "transaction_limit",
  "sandbox",
  "column_break_4",
+  "business_shortcode",
  "online_passkey",
  "security_credential",
  "get_account_balance",
@@ -84,10 +86,24 @@
   "fieldname": "get_account_balance",
   "fieldtype": "Button",
   "label": "Get Account Balance"
+  },
+  {
+   "depends_on": "eval:(doc.sandbox==0)",
+   "fieldname": "business_shortcode",
+   "fieldtype": "Data",
+   "label": "Business Shortcode",
+   "mandatory_depends_on": "eval:(doc.sandbox==0)"
+  },
+  {
+   "default": "150000",
+   "fieldname": "transaction_limit",
+   "fieldtype": "Float",
+   "label": "Transaction Limit",
+   "non_negative": 1
  }
 ],
 "links": [],
- "modified": "2020-09-25 20:21:38.215494",
+ "modified": "2021-01-29 12:02:16.106942",
 "modified_by": "Administrator",
 "module": "ERPNext Integrations",
 "name": "Mpesa Settings",
--- a/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_settings.py
+++ b/erpnext/erpnext_integrations/doctype/mpesa_settings/mpesa_settings.py
@@ -33,13 +33,34 @@ class MpesaSettings(Document):
 		create_mode_of_payment('Mpesa-' + self.payment_gateway_name, payment_type="Phone")

 	def request_for_payment(self, **kwargs):
-		if frappe.flags.in_test:
-			from erpnext.erpnext_integrations.doctype.mpesa_settings.test_mpesa_settings import get_payment_request_response_payload
-			response = frappe._dict(get_payment_request_response_payload())
-		else:
-			response = frappe._dict(generate_stk_push(**kwargs))
+		args = frappe._dict(kwargs)
+		request_amounts = self.split_request_amount_according_to_transaction_limit(args)

-		self.handle_api_response("CheckoutRequestID", kwargs, response)
+		for i, amount in enumerate(request_amounts):
+			args.request_amount = amount
+			if frappe.flags.in_test:
+				from erpnext.erpnext_integrations.doctype.mpesa_settings.test_mpesa_settings import get_payment_request_response_payload
+				response = frappe._dict(get_payment_request_response_payload(amount))
+			else:
+				response = frappe._dict(generate_stk_push(**args))
+
+			self.handle_api_response("CheckoutRequestID", args, response)
+
+	def split_request_amount_according_to_transaction_limit(self, args):
+		request_amount = args.request_amount
+		if request_amount > self.transaction_limit:
+			# make multiple requests
+			request_amounts = []
+			requests_to_be_made = frappe.utils.ceil(request_amount / self.transaction_limit) # 480/150 = ceil(3.2) = 4
+			for i in range(requests_to_be_made):
+				amount = self.transaction_limit
+				if i == requests_to_be_made - 1:
+					amount = request_amount - (self.transaction_limit * i) # for 4th request, 480 - (150 * 3) = 30
+				request_amounts.append(amount)
+		else:
+			request_amounts = [request_amount]
+		
+		return request_amounts

 	def get_account_balance_info(self):
 		payload = dict(
@@ -67,7 +88,8 @@ class MpesaSettings(Document):
 			req_name = getattr(response, global_id)
 			error = None

-		create_request_log(request_dict, "Host", "Mpesa", req_name, error)
+		if not frappe.db.exists('Integration Request', req_name):
+			create_request_log(request_dict, "Host", "Mpesa", req_name, error)

 		if error:
 			frappe.throw(_(getattr(response, "errorMessage")), title=_("Transaction Error"))
@@ -80,6 +102,8 @@ def generate_stk_push(**kwargs):

 		mpesa_settings = frappe.get_doc("Mpesa Settings", args.payment_gateway[6:])
 		env = "production" if not mpesa_settings.sandbox else "sandbox"
+		# for sandbox, business shortcode is same as till number
+		business_shortcode = mpesa_settings.business_shortcode if env == "production" else mpesa_settings.till_number

 		connector = MpesaConnector(env=env,
 			app_key=mpesa_settings.consumer_key,
@@ -87,10 +111,12 @@ def generate_stk_push(**kwargs):

 		mobile_number = sanitize_mobile_number(args.sender)

-		response = connector.stk_push(business_shortcode=mpesa_settings.till_number,
-			passcode=mpesa_settings.get_password("online_passkey"), amount=args.grand_total,
+		response = connector.stk_push(
+			business_shortcode=business_shortcode, amount=args.request_amount,
+			passcode=mpesa_settings.get_password("online_passkey"),
 			callback_url=callback_url, reference_code=mpesa_settings.till_number,
-			phone_number=mobile_number, description="POS Payment")
+			phone_number=mobile_number, description="POS Payment"
+		)

 		return response

@@ -108,29 +134,72 @@ def verify_transaction(**kwargs):
 	transaction_response = frappe._dict(kwargs["Body"]["stkCallback"])

 	checkout_id = getattr(transaction_response, "CheckoutRequestID", "")
-	request = frappe.get_doc("Integration Request", checkout_id)
-	transaction_data = frappe._dict(loads(request.data))
+	integration_request = frappe.get_doc("Integration Request", checkout_id)
+	transaction_data = frappe._dict(loads(integration_request.data))
+	total_paid = 0 # for multiple integration request made against a pos invoice
+	success = False # for reporting successfull callback to point of sale ui

 	if transaction_response['ResultCode'] == 0:
-		if request.reference_doctype and request.reference_docname:
+		if integration_request.reference_doctype and integration_request.reference_docname:
 			try:
-				doc = frappe.get_doc(request.reference_doctype,
-					request.reference_docname)
-				doc.run_method("on_payment_authorized", 'Completed')
-
 				item_response = transaction_response["CallbackMetadata"]["Item"]
+				amount = fetch_param_value(item_response, "Amount", "Name")
 				mpesa_receipt = fetch_param_value(item_response, "MpesaReceiptNumber", "Name")
-				frappe.db.set_value("POS Invoice", doc.reference_name, "mpesa_receipt_number", mpesa_receipt)
-				request.handle_success(transaction_response)
+				pr = frappe.get_doc(integration_request.reference_doctype, integration_request.reference_docname)
+
+				mpesa_receipts, completed_payments = get_completed_integration_requests_info(
+					integration_request.reference_doctype,
+					integration_request.reference_docname,
+					checkout_id
+				)
+
+				total_paid = amount + sum(completed_payments)
+				mpesa_receipts = ', '.join(mpesa_receipts + [mpesa_receipt])
+
+				if total_paid >= pr.grand_total:
+					pr.run_method("on_payment_authorized", 'Completed')
+					success = True
+
+				frappe.db.set_value("POS Invoice", pr.reference_name, "mpesa_receipt_number", mpesa_receipts)
+				integration_request.handle_success(transaction_response)
 			except Exception:
-				request.handle_failure(transaction_response)
+				integration_request.handle_failure(transaction_response)
 				frappe.log_error(frappe.get_traceback())

 	else:
-		request.handle_failure(transaction_response)
+		integration_request.handle_failure(transaction_response)

-	frappe.publish_realtime('process_phone_payment', doctype="POS Invoice",
-		docname=transaction_data.payment_reference, user=request.owner, message=transaction_response)
+	frappe.publish_realtime(
+		event='process_phone_payment',
+		doctype="POS Invoice",
+		docname=transaction_data.payment_reference,
+		user=integration_request.owner,
+		message={
+			'amount': total_paid,
+			'success': success,
+			'failure_message': transaction_response["ResultDesc"] if transaction_response['ResultCode'] != 0 else ''
+		},
+	)
+
+def get_completed_integration_requests_info(reference_doctype, reference_docname, checkout_id):
+	output_of_other_completed_requests = frappe.get_all("Integration Request", filters={
+		'name': ['!=', checkout_id],
+		'reference_doctype': reference_doctype,
+		'reference_docname': reference_docname,
+		'status': 'Completed'
+	}, pluck="output")
+
+	mpesa_receipts, completed_payments = [], []
+
+	for out in output_of_other_completed_requests:
+		out = frappe._dict(loads(out))
+		item_response = out["CallbackMetadata"]["Item"]
+		completed_amount = fetch_param_value(item_response, "Amount", "Name")
+		completed_mpesa_receipt = fetch_param_value(item_response, "MpesaReceiptNumber", "Name")
+		completed_payments.append(completed_amount)
+		mpesa_receipts.append(completed_mpesa_receipt)
+	
+	return mpesa_receipts, completed_payments

 def get_account_balance(request_payload):
 	"""Call account balance API to send the request to the Mpesa Servers."""
--- a/erpnext/erpnext_integrations/doctype/mpesa_settings/test_mpesa_settings.py
+++ b/erpnext/erpnext_integrations/doctype/mpesa_settings/test_mpesa_settings.py
@@ -9,6 +9,10 @@ from erpnext.erpnext_integrations.doctype.mpesa_settings.mpesa_settings import p
 from erpnext.accounts.doctype.pos_invoice.test_pos_invoice import create_pos_invoice

 class TestMpesaSettings(unittest.TestCase):
+	def tearDown(self):
+		frappe.db.sql('delete from `tabMpesa Settings`')
+		frappe.db.sql('delete from `tabIntegration Request` where integration_request_service = "Mpesa"')
+
 	def test_creation_of_payment_gateway(self):
 		create_mpesa_settings(payment_gateway_name="_Test")

@@ -40,6 +44,8 @@ class TestMpesaSettings(unittest.TestCase):
 			}
 		}))

+		integration_request.delete()
+
 	def test_processing_of_callback_payload(self):
 		create_mpesa_settings(payment_gateway_name="Payment")
 		mpesa_account = frappe.db.get_value("Payment Gateway Account", {"payment_gateway": 'Mpesa-Payment'}, "payment_account")
@@ -56,10 +62,16 @@ class TestMpesaSettings(unittest.TestCase):
 		# test payment request creation
 		self.assertEquals(pr.payment_gateway, "Mpesa-Payment")

-		callback_response = get_payment_callback_payload()
+		# submitting payment request creates integration requests with random id
+		integration_req_ids = frappe.get_all("Integration Request", filters={
+			'reference_doctype': pr.doctype,
+			'reference_docname': pr.name,
+		}, pluck="name")
+
+		callback_response = get_payment_callback_payload(Amount=500, CheckoutRequestID=integration_req_ids[0])
 		verify_transaction(**callback_response)
 		# test creation of integration request
-		integration_request = frappe.get_doc("Integration Request", "ws_CO_061020201133231972")
+		integration_request = frappe.get_doc("Integration Request", integration_req_ids[0])

 		# test integration request creation and successful update of the status on receiving callback response
 		self.assertTrue(integration_request)
@@ -69,8 +81,120 @@ class TestMpesaSettings(unittest.TestCase):
 		integration_request.reload()
 		self.assertEquals(pos_invoice.mpesa_receipt_number, "LGR7OWQX0R")
 		self.assertEquals(integration_request.status, "Completed")
+		
+		frappe.db.set_value("Customer", "_Test Customer", "default_currency", "")
+		integration_request.delete()
+		pr.reload()
+		pr.cancel()
+		pr.delete()
+		pos_invoice.delete()
+
+	def test_processing_of_multiple_callback_payload(self):
+		create_mpesa_settings(payment_gateway_name="Payment")
+		mpesa_account = frappe.db.get_value("Payment Gateway Account", {"payment_gateway": 'Mpesa-Payment'}, "payment_account")
+		frappe.db.set_value("Account", mpesa_account, "account_currency", "KES")
+		frappe.db.set_value("Mpesa Settings", "Payment", "transaction_limit", "500")
+		frappe.db.set_value("Customer", "_Test Customer", "default_currency", "KES")
+
+		pos_invoice = create_pos_invoice(do_not_submit=1)
+		pos_invoice.append("payments", {'mode_of_payment': 'Mpesa-Payment', 'account': mpesa_account, 'amount': 1000})
+		pos_invoice.contact_mobile = "093456543894"
+		pos_invoice.currency = "KES"
+		pos_invoice.save()
+
+		pr = pos_invoice.create_payment_request()
+		# test payment request creation
+		self.assertEquals(pr.payment_gateway, "Mpesa-Payment")
+
+		# submitting payment request creates integration requests with random id
+		integration_req_ids = frappe.get_all("Integration Request", filters={
+			'reference_doctype': pr.doctype,
+			'reference_docname': pr.name,
+		}, pluck="name")
+
+		# create random receipt nos and send it as response to callback handler
+		mpesa_receipt_numbers = [frappe.utils.random_string(5) for d in integration_req_ids]
+
+		integration_requests = []
+		for i in range(len(integration_req_ids)):
+			callback_response = get_payment_callback_payload(
+				Amount=500,
+				CheckoutRequestID=integration_req_ids[i],
+				MpesaReceiptNumber=mpesa_receipt_numbers[i]
+			)
+			# handle response manually
+			verify_transaction(**callback_response)
+			# test completion of integration request
+			integration_request = frappe.get_doc("Integration Request", integration_req_ids[i])
+			self.assertEquals(integration_request.status, "Completed")
+			integration_requests.append(integration_request)
+
+		# check receipt number once all the integration requests are completed
+		pos_invoice.reload()
+		self.assertEquals(pos_invoice.mpesa_receipt_number, ', '.join(mpesa_receipt_numbers))

 		frappe.db.set_value("Customer", "_Test Customer", "default_currency", "")
+		[d.delete() for d in integration_requests]
+		pr.reload()
+		pr.cancel()
+		pr.delete()
+		pos_invoice.delete()
+	
+	def test_processing_of_only_one_succes_callback_payload(self):
+		create_mpesa_settings(payment_gateway_name="Payment")
+		mpesa_account = frappe.db.get_value("Payment Gateway Account", {"payment_gateway": 'Mpesa-Payment'}, "payment_account")
+		frappe.db.set_value("Account", mpesa_account, "account_currency", "KES")
+		frappe.db.set_value("Mpesa Settings", "Payment", "transaction_limit", "500")
+		frappe.db.set_value("Customer", "_Test Customer", "default_currency", "KES")
+
+		pos_invoice = create_pos_invoice(do_not_submit=1)
+		pos_invoice.append("payments", {'mode_of_payment': 'Mpesa-Payment', 'account': mpesa_account, 'amount': 1000})
+		pos_invoice.contact_mobile = "093456543894"
+		pos_invoice.currency = "KES"
+		pos_invoice.save()
+
+		pr = pos_invoice.create_payment_request()
+		# test payment request creation
+		self.assertEquals(pr.payment_gateway, "Mpesa-Payment")
+
+		# submitting payment request creates integration requests with random id
+		integration_req_ids = frappe.get_all("Integration Request", filters={
+			'reference_doctype': pr.doctype,
+			'reference_docname': pr.name,
+		}, pluck="name")
+
+		# create random receipt nos and send it as response to callback handler
+		mpesa_receipt_numbers = [frappe.utils.random_string(5) for d in integration_req_ids]
+
+		callback_response = get_payment_callback_payload(
+			Amount=500,
+			CheckoutRequestID=integration_req_ids[0],
+			MpesaReceiptNumber=mpesa_receipt_numbers[0]
+		)
+		# handle response manually
+		verify_transaction(**callback_response)
+		# test completion of integration request
+		integration_request = frappe.get_doc("Integration Request", integration_req_ids[0])
+		self.assertEquals(integration_request.status, "Completed")
+
+		# now one request is completed
+		# second integration request fails
+		# now retrying payment request should make only one integration request again
+		pr = pos_invoice.create_payment_request()
+		new_integration_req_ids = frappe.get_all("Integration Request", filters={
+			'reference_doctype': pr.doctype,
+			'reference_docname': pr.name,
+			'name': ['not in', integration_req_ids]
+		}, pluck="name")
+
+		self.assertEquals(len(new_integration_req_ids), 1)
+
+		frappe.db.set_value("Customer", "_Test Customer", "default_currency", "")
+		frappe.db.sql("delete from `tabIntegration Request` where integration_request_service = 'Mpesa'")
+		pr.reload()
+		pr.cancel()
+		pr.delete()
+		pos_invoice.delete()

 def create_mpesa_settings(payment_gateway_name="Express"):
 	if frappe.db.exists("Mpesa Settings", payment_gateway_name):
@@ -160,16 +284,19 @@ def get_test_account_balance_response():
 	}
 		}

-def get_payment_request_response_payload():
+def get_payment_request_response_payload(Amount=500):
 	"""Response received after successfully calling the stk push process request API."""
+
+	CheckoutRequestID = frappe.utils.random_string(10)
+
 	return {
 		"MerchantRequestID": "8071-27184008-1",
-		"CheckoutRequestID": "ws_CO_061020201133231972",
+		"CheckoutRequestID": CheckoutRequestID,
 		"ResultCode": 0,
 		"ResultDesc": "The service request is processed successfully.",
 		"CallbackMetadata": {
 			"Item": [
-				{ "Name": "Amount", "Value": 500.0 },
+				{ "Name": "Amount", "Value": Amount },
 				{ "Name": "MpesaReceiptNumber", "Value": "LGR7OWQX0R" },
 				{ "Name": "TransactionDate", "Value": 20201006113336 },
 				{ "Name": "PhoneNumber", "Value": 254723575670 }
@@ -177,41 +304,26 @@ def get_payment_request_response_payload():
 		}
 	}

-
-def get_payment_callback_payload():
+def get_payment_callback_payload(Amount=500, CheckoutRequestID="ws_CO_061020201133231972", MpesaReceiptNumber="LGR7OWQX0R"):
 	"""Response received from the server as callback after calling the stkpush process request API."""
 	return {
 		"Body":{
-		"stkCallback":{
-			"MerchantRequestID":"19465-780693-1",
-			"CheckoutRequestID":"ws_CO_061020201133231972",
-			"ResultCode":0,
-			"ResultDesc":"The service request is processed successfully.",
-			"CallbackMetadata":{
-			"Item":[
-				{
-				"Name":"Amount",
-				"Value":500
-				},
-				{
-				"Name":"MpesaReceiptNumber",
-				"Value":"LGR7OWQX0R"
-				},
-				{
-				"Name":"Balance"
-				},
-				{
-				"Name":"TransactionDate",
-				"Value":20170727154800
-				},
-				{
-				"Name":"PhoneNumber",
-				"Value":254721566839
+			"stkCallback":{
+				"MerchantRequestID":"19465-780693-1",
+				"CheckoutRequestID":CheckoutRequestID,
+				"ResultCode":0,
+				"ResultDesc":"The service request is processed successfully.",
+				"CallbackMetadata":{
+					"Item":[
+						{ "Name":"Amount", "Value":Amount },
+						{ "Name":"MpesaReceiptNumber", "Value":MpesaReceiptNumber },
+						{ "Name":"Balance" },
+						{ "Name":"TransactionDate", "Value":20170727154800 },
+						{ "Name":"PhoneNumber", "Value":254721566839 }
+					]
 				}
-			]
 			}
 		}
-		}
 	}

 def get_account_balance_callback_payload():