Add STK Push Transaction Query feature

.gitignore

@@ -32,4 +32,5 @@ docs/_build/
 
 # File based database
 *.sqlite3
-.vscode
+.vscode
+venv/

django_daraja/mpesa/core.py

@@ -4,7 +4,7 @@
 from datetime import datetime
 import json
 from .exceptions import MpesaInvalidParameterException, MpesaConnectionError
-from .utils import encrypt_security_credential, mpesa_access_token, format_phone_number, api_base_url, mpesa_config, mpesa_response
+from .utils import encrypt_security_credential, mpesa_access_token, format_phone_number, api_base_url, mpesa_config, mpesa_response, mpesa_query_response
 from decouple import config
 
 class MpesaClient:
@@ -124,6 +124,56 @@ def stk_push(self, phone_number, amount, account_reference, transaction_desc, ca
 			raise MpesaConnectionError('Connection failed')
 		except Exception as ex:
 			raise MpesaConnectionError(str(ex))
+
+	def transaction_query(self, checkout_request_id):
+		"""
+		Queries the status of a Lipa na MPESA Online Payment (STK Push)
+
+		Args:
+			checkout_request_id (str): -- The Checkout Request ID of the transaction to be queried. This is the same value returned in the response of the original request (STK Push request).
+
+		Returns:
+			MpesaQueryResponse: MpesaQueryResponse object containing the details of the API response
+
+		Raises:
+			MpesaInvalidParameterException: Invalid parameter passed
+			MpesaConnectionError: Connection error
+
+		"""
+
+		if str(checkout_request_id).strip() == '':
+			raise MpesaInvalidParameterException('Checkout Request ID cannot be blank')
+
+		url = api_base_url() + 'mpesa/stkpushquery/v1/query'
+		mpesa_environment = mpesa_config('MPESA_ENVIRONMENT')
+		if mpesa_environment == 'sandbox':
+			business_short_code = mpesa_config('MPESA_EXPRESS_SHORTCODE')
+		else:
+			business_short_code = mpesa_config('MPESA_SHORTCODE')
+		passkey = mpesa_config('MPESA_PASSKEY')
+		timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
+		password = base64.b64encode((business_short_code + passkey + timestamp).encode('ascii')).decode('utf-8') 
+
+		data = {
+			'BusinessShortCode': business_short_code,
+			'Password': password,
+			'Timestamp': timestamp,
+			'CheckoutRequestID': checkout_request_id,
+		}
+
+		headers = {
+			'Authorization': 'Bearer ' + mpesa_access_token(),
+			'Content-type': 'application/json'
+		}
+
+		try:
+			r = requests.post(url, json=data, headers=headers)
+			response = mpesa_query_response(r)
+			return response
+		except requests.exceptions.ConnectionError:
+			raise MpesaConnectionError('Connection failed')
+		except Exception as ex:
+			raise MpesaConnectionError(str(ex))
 
 	def b2c_payment(self, phone_number, amount, transaction_desc, callback_url, occassion, command_id):
 		"""

django_daraja/mpesa/utils.py

@@ -31,6 +31,39 @@ class MpesaResponse(Response):
 	merchant_request_id = ''
 	checkout_request_id = ''
 
+class MpesaQueryResponse(Response):
+	request_id = ''
+	error_code = ''
+	error_message = ''
+	response_code = ''
+	response_description = ''
+	result_code = ''
+	result_desc = ''
+	checkout_request_id = ''
+	merchant_request_id = ''
+
+
+def mpesa_query_response(r):
+	"""
+	Create MpesaQueryResponse object from requests.Response object
+
+	Arguments:
+		r (requests.Response) -- The response to convert
+	"""
+
+	r.__class__ = MpesaQueryResponse
+	json_response = r.json()
+	r.response_code = json_response.get('ResponseCode', '')
+	r.response_description = json_response.get('ResponseDescription', '')
+	r.result_code = json_response.get('ResultCode', '')
+	r.result_desc = json_response.get('ResultDesc', '')
+	r.checkout_request_id = json_response.get('CheckoutRequestID', '')
+	r.merchant_request_id = json_response.get('MerchantRequestID', '')
+	r.request_id = json_response.get('requestId', '')
+	r.error_code = json_response.get('errorCode')
+	r.error_message = json_response.get('errorMessage', '')
+	return r
+
 
 def mpesa_response(r):
 	"""

tests/settings.py

@@ -26,7 +26,7 @@
 # SECURITY WARNING: don't run with debug turned on in production!
 DEBUG = True
 
-ALLOWED_HOSTS = ['testserver']
+ALLOWED_HOSTS = ['testserver', '*']
 
 
 # Application definition

tests/test_transaction_query.py

@@ -0,0 +1,72 @@
+# -*- coding: utf-8 -*-
+'''
+Test STK Push Transaction Query
+'''
+
+from django.test import TestCase
+from django_daraja.mpesa.core import MpesaClient
+from decouple import config
+from django_daraja.mpesa.exceptions import MpesaInvalidParameterException
+import time
+
+
+class MpesaStkPushTestCase(TestCase):
+
+	cl = MpesaClient()
+	callback_url = 'https://api.darajambili.com/express-payment'
+
+
+	def test_transaction_query_success(self):
+		'''
+		Test successful transaction query
+		'''
+
+		phone_number = config('LNM_PHONE_NUMBER')
+		amount = 1
+		account_reference = 'reference'
+		transaction_desc = 'Description'
+		time.sleep(10) # Wait for any ongoing STK push to complete
+		stk_response = self.cl.stk_push(
+			phone_number, amount, account_reference, transaction_desc, self.callback_url)
+		if stk_response.response_code == '0':
+			time.sleep(10) # Wait for STK push to complete
+			status_response = self.cl.transaction_query(stk_response.checkout_request_id)
+			self.assertEqual(status_response.response_code, '0')
+			self.assertEqual(status_response.response_description,
+					         'The service request has been accepted successsfully')
+			self.assertTrue(len(status_response.merchant_request_id) > 0)
+			self.assertTrue(len(status_response.checkout_request_id) > 0)
+		else:
+			self.fail('STK push failed')
+
+	def test_transaction_being_processed(self):
+		'''
+		Test that transaction query returns 'The service request is being processed' when transaction is being processed
+		'''
+
+		phone_number = config('LNM_PHONE_NUMBER')
+		amount = 1
+		account_reference = 'reference'
+		transaction_desc = 'Description'
+		time.sleep(10)
+		stk_response = self.cl.stk_push(
+			phone_number, amount, account_reference, transaction_desc, self.callback_url)
+		if stk_response.response_code == '0':
+			status_response = self.cl.transaction_query(
+				stk_response.checkout_request_id)
+			self.assertEqual(status_response.request_id, stk_response.checkout_request_id)
+			self.assertEqual(status_response.error_message,
+			                 'The transaction is being processed')
+		else:
+			self.fail('STK push failed')
+
+	def test_transaction_query_empty_checkout_request_id(self):
+		'''
+		Test that Transaction Query with empty Checkout Request ID raises MpesaInvalidParameterException
+		'''
+
+		with self.assertRaises(MpesaInvalidParameterException):
+			checkout_request_id = ''
+			self.cl.transaction_query(checkout_request_id)
+
+