From fe4209ba92160c60868e5c7d530a2b74909999ae Mon Sep 17 00:00:00 2001 From: Okeyo Gichana Mayaka Date: Wed, 15 Nov 2023 11:04:27 +0300 Subject: [PATCH] Completed transfer of mpesa b2c integration application to new repository. --- .gitignore | 12 +- README.md | 22 +- .../mpesa_b2c/doctype/__init__.py | 5 + .../doctype/csf_ke_custom_exceptions.py | 40 ++ .../doctype/daraja_access_tokens/__init__.py | 0 .../daraja_access_tokens.js | 22 + .../daraja_access_tokens.json | 66 +++ .../daraja_access_tokens.py | 21 + .../test_daraja_access_tokens.py | 134 +++++ .../doctype/mpesa_b2c_payment/__init__.py | 0 .../mpesa_b2c_payment/encoding_credentials.py | 87 +++ .../mpesa_b2c_payment/mpesa_b2c_payment.js | 102 ++++ .../mpesa_b2c_payment/mpesa_b2c_payment.json | 132 +++++ .../mpesa_b2c_payment/mpesa_b2c_payment.py | 556 ++++++++++++++++++ .../test_mpesa_b2c_payment.py | 201 +++++++ .../__init__.py | 0 .../mpesa_b2c_payments_transactions.js | 8 + .../mpesa_b2c_payments_transactions.json | 127 ++++ .../mpesa_b2c_payments_transactions.py | 59 ++ .../test_mpesa_b2c_payments_transactions.py | 141 +++++ .../doctype/mpesa_b2c_settings/__init__.py | 0 .../mpesa_b2c_settings/mpesa_b2c_settings.js | 57 ++ .../mpesa_b2c_settings.json | 126 ++++ .../mpesa_b2c_settings/mpesa_b2c_settings.py | 64 ++ .../test_mpesa_b2c_settings.py | 141 +++++ 25 files changed, 2117 insertions(+), 6 deletions(-) create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/__init__.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/csf_ke_custom_exceptions.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/__init__.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.js create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.json create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/test_daraja_access_tokens.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/__init__.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/encoding_credentials.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.js create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.json create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/test_mpesa_b2c_payment.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/__init__.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.js create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.json create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/test_mpesa_b2c_payments_transactions.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/__init__.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.js create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.json create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.py create mode 100644 navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/test_mpesa_b2c_settings.py diff --git a/.gitignore b/.gitignore index ba04025..a278cdc 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,14 @@ *.pyc *.egg-info *.swp +*.rest + +*.pem +*.cer + +logfile.txt tags -node_modules -__pycache__ \ No newline at end of file + +csf_ke/docs/current +/.vscode +/__pycache__/ diff --git a/README.md b/README.md index f6fa0e3..b8e054d 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,21 @@ -## MPesa B2C +## EXPNext MPesa B2C Integration -MPesa B2C Integration for ERPNext +### Installation -#### License +Using bench, [install ERPNext](https://github.com/frappe/bench#installation) as mentioned here. -mit \ No newline at end of file +Once ERPNext is installed, add MPesa B2c app to your bench by running + +```sh +$ bench get-app https://github.com/navariltd/navari-mpesa-b2c.git +``` + +After that, you can install MPesa B2C app on required site by running + +```sh +$ bench --site [site.name] install-app navari_mpesa_b2c +``` + +### License + +MIT. See [license.txt](https://github.com/navariltd/navari-mpesa-b2c/blob/develop/license.txt) for more information. diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/__init__.py b/navari_mpesa_b2c/mpesa_b2c/doctype/__init__.py new file mode 100644 index 0000000..19d868c --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/__init__.py @@ -0,0 +1,5 @@ +import frappe +from frappe.utils import logger + +logger.set_log_level("DEBUG") +api_logger = frappe.logger("api", allow_site=True, file_count=50) diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/csf_ke_custom_exceptions.py b/navari_mpesa_b2c/mpesa_b2c/doctype/csf_ke_custom_exceptions.py new file mode 100644 index 0000000..5086edb --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/csf_ke_custom_exceptions.py @@ -0,0 +1,40 @@ +"""Custom Exceptions and Errors raised by modules in the CSF_KE application""" + + +class InvalidReceiverMobileNumberError(Exception): + """Raised when receiver's mobile number fails validation""" + + +class InsufficientPaymentAmountError(Exception): + """Raised when the payment amount is less than the required KShs. 10""" + + +class IncorrectStatusError(Exception): + """Raised when status is Errored but no errod description or error code has been supplied""" + + +class InvalidTokenExpiryTime(Exception): + """ + Raised when the access token's expiry time is earlier + or the same as the access token's fetch time. + It should always be 1 hour after the fetch time. + """ + + +class InvalidURLError(Exception): + """Raised when URLs fail validation""" + + +class InvalidAuthenticationCertificateFileError(Exception): + """Raised when an invalid certificate file, i.e. not a .cer or .pem, is uploaded""" + + +class UnExistentB2CPaymentRecordError(Exception): + """Raised when referencing a B2C Payment that does not exist""" + + +class InformationMismatchError(Exception): + """ + Raised when there's a mismatch in any of the B2C Payment's records + and the corresponding B2C Payments Transaction's records + """ diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/__init__.py b/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.js b/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.js new file mode 100644 index 0000000..2e60d87 --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.js @@ -0,0 +1,22 @@ +// Copyright (c) 2023, Navari Limited and contributors +// For license information, please see license.txt + +frappe.ui.form.on("Daraja Access Tokens", { + validate: function (frm) { + if (frm.doc.expiry_time && frm.doc.token_fetch_time) { + expiryTime = new Date(frm.doc.expiry_time); + fetchTime = new Date(frm.doc.token_fetch_time); + + if (expiryTime <= fetchTime) { + frappe.msgprint({ + message: __( + "Token Expiry Time cannot be earlier than or the same as Token Fetch Time" + ), + indicator: "red", + title: "Validation Error", + }); + frappe.validated = false; + } + } + }, +}); diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.json b/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.json new file mode 100644 index 0000000..039d193 --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.json @@ -0,0 +1,66 @@ +{ + "actions": [], + "autoname": "autoincrement", + "creation": "2023-11-01 11:56:09.082783", + "default_view": "List", + "doctype": "DocType", + "editable_grid": 1, + "engine": "InnoDB", + "field_order": [ + "access_token", + "token_fetch_time", + "column_break_snir", + "expiry_time" + ], + "fields": [ + { + "fieldname": "access_token", + "fieldtype": "Password", + "in_list_view": 1, + "label": "Access Token", + "reqd": 1 + }, + { + "fieldname": "token_fetch_time", + "fieldtype": "Datetime", + "label": "Token Fetch Time", + "reqd": 1 + }, + { + "fieldname": "column_break_snir", + "fieldtype": "Column Break" + }, + { + "fieldname": "expiry_time", + "fieldtype": "Datetime", + "in_list_view": 1, + "label": "Expiry", + "reqd": 1 + } + ], + "index_web_pages_for_search": 1, + "links": [], + "modified": "2023-11-15 10:42:44.625615", + "modified_by": "Administrator", + "module": "MPesa B2C", + "name": "Daraja Access Tokens", + "naming_rule": "Autoincrement", + "owner": "Administrator", + "permissions": [ + { + "create": 1, + "delete": 1, + "email": 1, + "export": 1, + "print": 1, + "read": 1, + "report": 1, + "role": "System Manager", + "share": 1, + "write": 1 + } + ], + "sort_field": "modified", + "sort_order": "DESC", + "states": [] +} diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.py b/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.py new file mode 100644 index 0000000..217ba87 --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/daraja_access_tokens.py @@ -0,0 +1,21 @@ +# Copyright (c) 2023, Navari Limited and contributors +# For license information, please see license.txt + +from frappe.model.document import Document + +from ..csf_ke_custom_exceptions import InvalidTokenExpiryTime +from .. import api_logger + + +class DarajaAccessTokens(Document): + """Daraja Access Tokens controller class""" + + def validate(self) -> None: + """Run validations before saving document""" + if self.expiry_time and self.expiry_time <= self.token_fetch_time: + api_logger.error( + "Access Token Expiry time cannot be same or early than the fetch time" + ) + raise InvalidTokenExpiryTime( + "Access Token Expiry time cannot be same or early than the fetch time" + ) diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/test_daraja_access_tokens.py b/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/test_daraja_access_tokens.py new file mode 100644 index 0000000..2d8bee4 --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/daraja_access_tokens/test_daraja_access_tokens.py @@ -0,0 +1,134 @@ +# Copyright (c) 2023, Navari Limited and Contributors +# See license.txt + +import datetime +from unittest.mock import patch + +import frappe +import requests +from frappe.tests.utils import FrappeTestCase +from frappe.utils.password import get_decrypted_password + +from ..csf_ke_custom_exceptions import InvalidTokenExpiryTime +from ..mpesa_b2c_payment import mpesa_b2c_payment +from ..mpesa_b2c_payment.mpesa_b2c_payment import get_hashed_token + +TOKEN_ACCESS_TIME = datetime.datetime.now() + + +def create_access_token() -> None: + """Creates a valid access token record for testing""" + if frappe.flags.test_events_created: + return + + frappe.set_user("Administrator") + + expiry_time = TOKEN_ACCESS_TIME + datetime.timedelta(hours=1) + frappe.get_doc( + { + "doctype": "Daraja Access Tokens", + "access_token": "123456789", + "token_fetch_time": TOKEN_ACCESS_TIME, + "expiry_time": expiry_time, + } + ).insert() + + frappe.flags.test_events_created = True + + +class TestDarajaAccessTokens(FrappeTestCase): + """Testing the Daraja Access Tokens doctype""" + + def setUp(self) -> None: + create_access_token() + + def tearDown(self) -> None: + frappe.set_user("Administrator") + + def test_valid_access_token(self) -> None: + """Attempt to access an existing token""" + token = frappe.db.get_value( + "Daraja Access Tokens", + {"token_fetch_time": TOKEN_ACCESS_TIME}, + ["name", "token_fetch_time", "expiry_time"], + as_dict=True, + ) + access_token = get_decrypted_password( + "Daraja Access Tokens", token.name, "access_token" + ) + + self.assertEqual(access_token, "123456789") + self.assertEqual(token.token_fetch_time, TOKEN_ACCESS_TIME) + self.assertEqual( + token.expiry_time, TOKEN_ACCESS_TIME + datetime.timedelta(hours=1) + ) + + def test_create_incomplete_access_token(self) -> None: + """Attemp to create a record from incomplete data""" + with self.assertRaises(frappe.exceptions.MandatoryError): + frappe.get_doc( + { + "doctype": "Daraja Access Tokens", + "access_token": "123456789", + "token_fetch_time": TOKEN_ACCESS_TIME, + } + ).insert() + + def test_incorrect_datetime_type(self) -> None: + """Test passing strings to datetime fields""" + with self.assertRaises(TypeError): + frappe.get_doc( + { + "doctype": "Daraja Access Tokens", + "access_token": TOKEN_ACCESS_TIME + datetime.timedelta(hours=1), + "token_fetch_time": TOKEN_ACCESS_TIME, + "expiry_time": "123456789", + } + ).insert() + + def test_expiry_time_earlier_than_fetch_time(self) -> None: + """Test expiry time being early than fetch time""" + with self.assertRaises(InvalidTokenExpiryTime): + frappe.get_doc( + { + "doctype": "Daraja Access Tokens", + "access_token": "123456789", + "token_fetch_time": TOKEN_ACCESS_TIME, + "expiry_time": TOKEN_ACCESS_TIME - datetime.timedelta(hours=1), + } + ).insert() + + def test_get_hashed_token(self) -> None: + """Tests the get_hashed_token() function from the b2c_payment module""" + hashed_token = get_hashed_token() + token = frappe.db.get_value( + "Daraja Access Tokens", + {"token_fetch_time": TOKEN_ACCESS_TIME}, + ["name"], + ) + + self.assertEqual(hashed_token, token) + + @patch.object(mpesa_b2c_payment.requests, "get") + def test_get_access_tokens(self, mock_response) -> None: + """Tests the get_access_tokens() function from the b2c_payment module""" + mock_response.return_value.status_code = 200 + mock_response.return_value.text = { + "access_token": "987654321", + "expires_in": "3599", + } + + token, status_code = mpesa_b2c_payment.get_access_tokens( + "123456789", "secret", "https://example.com/authorise" + ) + self.assertIsInstance(token, dict) + self.assertEqual(token["access_token"], "987654321") + self.assertEqual(token["expires_in"], "3599") + self.assertEqual(status_code, 200) + + def test_get_access_tokens_error_response(self) -> None: + """Tests instances the get_access_tokens() function from the b2c_payment module fails""" + with self.assertRaises(requests.HTTPError): + mpesa_b2c_payment.get_access_tokens( + "123456789", "secret", "https://example.com/authorise" + ) diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/__init__.py b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/encoding_credentials.py b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/encoding_credentials.py new file mode 100644 index 0000000..cd6fa7e --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/encoding_credentials.py @@ -0,0 +1,87 @@ +"""Utility functions that handle the encoding and decoding of credentials""" + +import base64 +import hashlib +import os + +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + +KEY_LEN = 32 +IV_LEN = 16 + + +def evp_bytes_to_key( + password: bytes, salt: bytes, key_len=KEY_LEN, iv_len=IV_LEN +) -> tuple[bytes, bytes]: + """Derives the key and the IV from the password and the salt using evp_bytes_to_key function""" + dtot = hashlib.md5(password + salt).digest() + d = [dtot] + while len(dtot) < (iv_len + key_len): + d.append(hashlib.md5(d[-1] + password + salt).digest()) + dtot += d[-1] + return dtot[:key_len], dtot[key_len : key_len + iv_len] + + +def pkcs7_pad(data: bytes, block_size: int = 16) -> bytes: + """Pad the certificate data with PKCS#7""" + pad_len = block_size - (len(data) % block_size) + return data + bytes([pad_len] * pad_len) + + +def pkcs7_unpad(data: bytes) -> bytes: + """Unpad the decrypted data with PKCS#7""" + pad_len = data[-1] + return data[:-pad_len] + + +def openssl_encrypt_encode(password: bytes, cert_file: str) -> bytes: + """ + Defines a function that encrypts and encodes the password + using a certificate file and OpenSSL and Base64 encoding + """ + with open(cert_file, "rb") as f: + cert_data = f.read() + + # Generate a random salt of 8 bytes. + salt = os.urandom(8) + + key, iv = evp_bytes_to_key(password, salt) + + # Create a cipher object using AES-256 in CBC mode with the derived key and IV + cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend()) + + padded_data = pkcs7_pad(cert_data) + + # Encrypt the padded data using the cipher object + encryptor = cipher.encryptor() + encrypted_data = encryptor.update(padded_data) + encryptor.finalize() + + # Encode the encrypted data in base64 + encoded_data = base64.b64encode(encrypted_data) + + # Return the encoded data with the salt as prefix + return salt + encoded_data + + +def openssl_decrypt_decode(password: bytes, encoded_data: bytes) -> bytes: + """Defines a function that decrypts and decodes a file with OpenSSL and base64""" + # Extract the salt from the first 8 bytes of the encoded data + salt = encoded_data[:8] + + # Decode the rest of the encoded data from base64 + encrypted_data = base64.b64decode(encoded_data[8:]) + + key, iv = evp_bytes_to_key(password, salt) + + # Create a cipher object using AES-256 in CBC mode with the derived key and IV + cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend()) + + # Decrypt the encrypted data using the cipher object + decryptor = cipher.decryptor() + decrypted_data = decryptor.update(encrypted_data) + decryptor.finalize() + + unpadded_data = pkcs7_unpad(decrypted_data) + + # Return the original certificate data + return unpadded_data diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.js b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.js new file mode 100644 index 0000000..e9e32c0 --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.js @@ -0,0 +1,102 @@ +// Copyright (c) 2023, Navari Limited and contributors +// For license information, please see license.txt + +frappe.ui.form.on("MPesa B2C Payment", { + validate: function (frm) { + if (frm.doc.partyb) { + if (!validatePhoneNumber(frm.doc.partyb)) { + // Validate if the receiver's mobile number is valid + frappe.msgprint({ + title: __("Validation Error"), + indicator: "red", + message: __("The Receiver (mobile number) entered is incorrect."), + }); + frappe.validated = false; + } + } + + if (frm.doc.amount < 10) { + // Validate amount to be greater then KShs. 10 + frappe.msgprint({ + title: __("Validation Error"), + indicator: "red", + message: __( + "Amount entered is less than the least acceptable amount of Kshs. 10" + ), + }); + frappe.validated = false; + } + }, + refresh: function (frm) { + if ( + !frm.doc.__islocal && + (frm.doc.status === "Not Initiated" || frm.doc.status === "Timed-Out") + ) { + // Only render the Initiate Payment button if document is saved, and + // payment status is "Not Initiated" or "Timed-Out" + frm.add_custom_button("Initiate Payment", async function () { + frappe.call({ + method: + "csf_ke.csf_ke.doctype.mpesa_b2c_payment.mpesa_b2c_payment.initiate_payment", + args: { + // Create request with a partial payload + partial_payload: { + name: frm.doc.name, + OriginatorConversationID: frm.doc.originatorconversationid, + CommandID: frm.doc.commandid, + Amount: frm.doc.amount, + PartyB: frm.doc.partyb, + Remarks: frm.doc.remarks, + Occassion: frm.doc.occassion, + }, + }, + callback: function (response) { + // Redirect upon response. Response received is success since error responses + // raise an HTTPError on the server-side + if (response.message === "No certificate file found in server") { + frappe.msgprint({ + title: __("Authentication Error"), + indicator: "red", + message: __(response.message), + }); + } else if (response.message === "successful") { + location.reload(); + } else { + // TODO: Add proper cases + frappe.msgprint(`${response}`); + } + }, + }); + }); + } + + if (!frm.doc.originatorconversationid) { + // Set uuidv4 compliant string + frm.set_value("originatorconversationid", generateUUIDv4()); + } + }, +}); + +function generateUUIDv4() { + // Generates a uuid4 string conforming to RFC standards + let uuid = "xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx".replace( + /[xy]/g, + function (c) { + let r = (Math.random() * 16) | 0, + v = c === "x" ? r : (r & 0x3) | 0x8; + return v.toString(16); + } + ); + return uuid; +} + +function validatePhoneNumber(input) { + // Validates the receiver phone numbers + if (input.startsWith("2547")) { + const pattern = /^2547\d{8}$/; + return pattern.test(input); + } else { + const pattern = /^(25410|25411)\d{7}$/; + return pattern.test(input); + } +} diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.json b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.json new file mode 100644 index 0000000..b1ce85c --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.json @@ -0,0 +1,132 @@ +{ + "actions": [], + "autoname": "naming_series:", + "creation": "2023-10-30 11:38:30.870409", + "default_view": "List", + "doctype": "DocType", + "editable_grid": 1, + "engine": "InnoDB", + "field_order": [ + "naming_series", + "section_break_pujd", + "originatorconversationid", + "commandid", + "remarks", + "status", + "error_description", + "column_break_jg71", + "partyb", + "amount", + "occassion", + "error_code" + ], + "fields": [ + { + "fieldname": "naming_series", + "fieldtype": "Select", + "hidden": 1, + "label": "Naming Series", + "options": "MPESA-B2C-" + }, + { + "fieldname": "section_break_pujd", + "fieldtype": "Section Break" + }, + { + "description": "This is randomly generated. Do not change", + "fieldname": "originatorconversationid", + "fieldtype": "Data", + "in_list_view": 1, + "label": "Originator Conversation ID", + "read_only": 1, + "reqd": 1 + }, + { + "fieldname": "commandid", + "fieldtype": "Select", + "in_list_view": 1, + "label": "Command ID", + "options": "\nSalaryPayment\nBusinessPayment\nPromotionPayment", + "reqd": 1 + }, + { + "fieldname": "remarks", + "fieldtype": "Data", + "label": "Remarks", + "reqd": 1 + }, + { + "default": "Not Initiated", + "fieldname": "status", + "fieldtype": "Select", + "label": "Payment Status", + "options": "\nNot Initiated\nPaid\nPending\nErrored\nTimed-Out", + "read_only": 1, + "reqd": 1 + }, + { + "depends_on": "eval:doc.status === \"Errored\";", + "fieldname": "error_description", + "fieldtype": "Data", + "label": "Error Description", + "read_only": 1 + }, + { + "fieldname": "column_break_jg71", + "fieldtype": "Column Break" + }, + { + "description": "Enter a phone number starting with 254. Format: 2547xxxxxxxx", + "fieldname": "partyb", + "fieldtype": "Data", + "in_list_view": 1, + "label": "Receiver", + "reqd": 1 + }, + { + "fieldname": "amount", + "fieldtype": "Float", + "in_list_view": 1, + "label": "Amount", + "reqd": 1 + }, + { + "fieldname": "occassion", + "fieldtype": "Data", + "label": "Occassion" + }, + { + "depends_on": "eval:doc.status === \"Errored\";", + "fieldname": "error_code", + "fieldtype": "Data", + "label": "Error Code", + "read_only": 1 + } + ], + "index_web_pages_for_search": 1, + "links": [], + "modified": "2023-11-15 10:43:07.830136", + "modified_by": "Administrator", + "module": "MPesa B2C", + "name": "MPesa B2C Payment", + "naming_rule": "By \"Naming Series\" field", + "owner": "Administrator", + "permissions": [ + { + "create": 1, + "delete": 1, + "email": 1, + "export": 1, + "print": 1, + "read": 1, + "report": 1, + "role": "System Manager", + "share": 1, + "write": 1 + } + ], + "sort_field": "modified", + "sort_order": "DESC", + "states": [], + "track_changes": 1 +} diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.py b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.py new file mode 100644 index 0000000..fb76b9e --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/mpesa_b2c_payment.py @@ -0,0 +1,556 @@ +# Copyright (c) 2023, Navari Limited and contributors +# For license information, please see license.txt + +import ast +import base64 +import datetime +import json +import re +from typing import Literal +from uuid import uuid4 + +import frappe +import requests +from frappe.model.document import Document +from frappe.utils.file_manager import get_file_path +from frappe.utils.password import get_decrypted_password + +from csf_ke.csf_ke.doctype import api_logger +from csf_ke.csf_ke.doctype.mpesa_b2c_payment.encoding_credentials import ( + openssl_encrypt_encode, +) + +from ..csf_ke_custom_exceptions import ( + IncorrectStatusError, + InsufficientPaymentAmountError, + InvalidReceiverMobileNumberError, +) + + +class MPesaB2CPayment(Document): + """MPesa B2C Payment Class""" + + def validate(self) -> None: + """Validations""" + self.error = "" + + if not self.originatorconversationid: + # Generate random UUID4 + self.originatorconversationid = str(uuid4()) + + if self.partyb and not validate_receiver_mobile_number(self.partyb): + # Validate mobile number of receiver, i.e. PartyB + self.error = ( + "The Receiver (mobile number) entered is incorrect for payment: %s." + ) + + api_logger.error(self.error, self.name) + raise InvalidReceiverMobileNumberError(self.error, self.name) + + if self.amount < 10: + # Validates payment amount + self.error = "Amount entered is less than the least acceptable amount of Kshs. 10, for payment: %s" + + api_logger.error(self.error, self.name) + raise InsufficientPaymentAmountError(self.error, self.name) + + if not self.status: + self.status = "Not Initiated" + + if self.status == "Errored": + if not (self.error_description and self.error_code): + self.error = "Status 'Errored' needs to have a corresponding error_code and error_description for payment: %s" + + api_logger.error(self.error, self.name) + raise IncorrectStatusError(self.error, self.name) + + +@frappe.whitelist(methods="POST") +def initiate_payment(partial_payload: str) -> None: + """ + This endpoint initiates the payment process. + The endpoint first checks if a valid (meaning un-expired) access token is available. + If none is found, it fetches one from the authorization url provided in the MPesa B2C Settings and + proceeds to initiate a payment request to the payment url also specified in the MPesa B2C Settings. + If a valid token is found, a payment initialization request is placed immediately. + """ + partial_payload = json.loads(frappe.form_dict.partial_payload) + b2c_settings = frappe.db.get_singles_dict("MPesa B2C Settings") + + payment_document = frappe.db.get_value( + "MPesa B2C Payment", + {"name": partial_payload.get("name")}, + ["name", "status"], + as_dict=True, + ) + hashed_token = get_hashed_token() + + if not hashed_token: + consumer_key, consumer_secret, authorization_url = get_b2c_settings( + b2c_settings + ) + response, status_code = get_access_tokens( + consumer_key, consumer_secret, authorization_url + ) + + if status_code == requests.codes.ok: + # If response code is 200, proceed + bearer_token = save_access_token_to_database(response) + make_payment(bearer_token, b2c_settings, partial_payload, payment_document) + else: + bearer_token = get_decrypted_password( + "Daraja Access Tokens", hashed_token, "access_token" + ) + + make_payment(bearer_token, b2c_settings, partial_payload, payment_document) + + +@frappe.whitelist(allow_guest=True) +def results_callback_url(Result: dict) -> None: + """ + Handles results response from Safaricom after successful B2C Payment request. + For a complete description of the response parameters: https://developer.safaricom.co.ke/APIs/BusinessToCustomer + """ + results = ast.literal_eval(json.dumps(Result)) + ( + originator_conversation_id, + result_type, + result_code, + results_description, + transaction_id, + ) = get_result_details(results) + + if result_type == 0: + if result_code == 0: + handle_successful_result_response( + results, originator_conversation_id, transaction_id + ) + else: + handle_unsuccessful_result_response( + transaction_id, + originator_conversation_id, + result_code, + results_description, + ) + else: + handle_duplicate_request(originator_conversation_id) + + +@frappe.whitelist(allow_guest=True) +def queue_timeout_url(response): + """Handles timeout responses from Safaricom""" + # TODO: Properly handle timeout responses. Not clearly specified in Safaricom's documentations + frappe.msgprint(f"{response}") + + +def get_hashed_token() -> str | list[None]: + """ + Checks if a valid (read un-expired) token is present in the database, + fetches and returns it. Otherwise, returns an empty list + """ + current_time = datetime.datetime.now() + hashed_token = frappe.db.sql( + f""" + SELECT name, access_token + FROM `tabDaraja Access Tokens` + WHERE expiry_time > '{current_time.strftime("%Y-%m-%d %H:%M:%S")}' + ORDER BY creation DESC + LIMIT 1 + """, + as_dict=True, + ) + + if hashed_token: + return hashed_token[0].name + + return [] + + +def get_b2c_settings(b2c_settings: Document) -> tuple[str, str, str]: + """Gets the consumer key, secret, and authorization url from the MPesa B2C Settings doctype""" + consumer_key = b2c_settings.get("consumer_key") + authorization_url = b2c_settings.get("authorization_url") + consumer_secret = get_decrypted_password( + "MPesa B2C Settings", "MPesa B2C Settings", "consumer_secret" + ) + return consumer_key, consumer_secret, authorization_url + + +def get_access_tokens( + consumer_key: str, consumer_secret: str, url: str +) -> tuple[str, int]: + """ + Gets the access token from the authorization url specified in the MPesa B2C Settings doctype. + """ + keys = f"{consumer_key}:{consumer_secret}" + encoded_credentials = base64.b64encode(keys.encode()).decode() + + try: + response = requests.get( + url, + headers={ + "Authorization": f"Basic {encoded_credentials}", + "Content-Type": "application/json", + }, + timeout=60, + ) + + response.raise_for_status() # Raise HTTPError if status code >= 400 + + except requests.HTTPError: + api_logger.exception("Exception Encountered when fetching access token") + raise + + except requests.exceptions.ConnectionError: + api_logger.exception("Exception Encountered when fetching access token") + raise + + except Exception: + api_logger.exception("Exception Encountered") + raise + + return response.text, response.status_code + + +def save_access_token_to_database(response: str) -> str: + """ + Deserialises the response object and saves the access token to the database, + returning the access token + """ + token_fetch_time = datetime.datetime.now() + response = json.loads(response) + + expiry_time = datetime.datetime.now() + datetime.timedelta( + seconds=int(response.get("expires_in")) + ) + + new_token = frappe.new_doc("Daraja Access Tokens") + new_token.access_token = response.get("access_token") + new_token.expiry_time = expiry_time + new_token.token_fetch_time = token_fetch_time + new_token.save() + + api_logger.info( + "Access token fetched and saved successfully at %s expiring at %s", + token_fetch_time, + expiry_time, + ) + return response.get("access_token") + + +def get_certificate_file(certificate_path: str) -> str | Literal[-1]: + """ + Gets the specified certificate's file path in the server. + This is the path of the file attached under the Authorisation Certificate File + in the MPesa B2C Settings field. + """ + if certificate_path: + certificate: str | None = get_file_path(certificate_path) + + return certificate + + api_logger.error( + "No valid Authentication Certificate file (*.cer or *.pem) found in the server." + ) + return -1 + + +def generate_payload( + b2c_settings: Document, + partial_payload: dict[str, str | int], + security_credentials: str, +) -> str: + """Generates an MPesa B2C API conforming payload to send in order to initiate payment""" + partial_payload_from_settings = { + "PartyA": b2c_settings.organisation_shortcode, + "InitiatorName": b2c_settings.initiator_name, + "SecurityCredential": security_credentials, + "QueueTimeOutURL": b2c_settings.queue_timeout_url, + "ResultURL": b2c_settings.results_url, + } + + partial_payload.update(partial_payload_from_settings) + + return json.dumps(partial_payload) + + +def get_result_details(results: dict) -> tuple[str, str, str | int, str, str | int]: + """ + Takes the results callback's result object and returns the + Originator Conversation ID, the Result Type, Result Code, + Result Description, and Transaction ID respectively + """ + originator_conversation_id = results.get("OriginatorConversationID") + result_type = int(results.get("ResultType")) + result_code = int(results.get("ResultCode")) + results_description = results.get("ResultDesc") + transaction_id = results.get("TransactionID") + + return ( + originator_conversation_id, + result_type, + result_code, + results_description, + transaction_id, + ) + + +def handle_successful_result_response( + results: dict, originator_conversation_id: str, transaction_id: str +) -> None: + """ + Handles the results callback's responses with a successful ResultCode, i.e. ResultCode of 0 + """ + mpesa_b2c_payment_document = frappe.db.get_value( + "MPesa B2C Payment", + {"originatorconversationid": originator_conversation_id}, + ["name"], + as_dict=True, + ) + + result_parameters = results.get("ResultParameters").get("ResultParameter") + transaction_values = extract_transaction_values(result_parameters, transaction_id) + transaction_values.update( + {"mpesa_b2c_payment_name": mpesa_b2c_payment_document.name} + ) + + update_doctype_single_values( + "MPesa B2C Payment", mpesa_b2c_payment_document, "status", "Paid" + ) + + transaction = save_transaction_to_database( + "MPesa B2C Payments Transactions", transaction_values + ) + + frappe.response["transaction"] = transaction + + +def handle_unsuccessful_result_response( + transaction_id: str, + originator_conversation_id: str, + result_code: int, + results_description: str, +) -> None: + """ + Handles the results callback's responses with an unsuccessful ResultCode, i.e. ResultCode != 0 + """ + mpesa_b2c_payment_document = frappe.db.get_value( + "MPesa B2C Payment", + {"originatorconversationid": originator_conversation_id}, + as_dict=True, + ) + + api_logger.info( + "Transaction %s with originator conversation id %s Errored with code: %s, description: %s", + transaction_id, + originator_conversation_id, + result_code, + results_description, + ) + update_doctype_single_values( + "MPesa B2C Payment", mpesa_b2c_payment_document, "status", "Errored" + ) + update_doctype_single_values( + "MPesa B2C Payment", mpesa_b2c_payment_document, "error_code", result_code + ) + update_doctype_single_values( + "MPesa B2C Payment", + mpesa_b2c_payment_document, + "error_description", + results_description, + ) + + +def handle_duplicate_request(originator_conversation_id: str) -> None: + """ + Logs instances where multiple requests from same B2C payment record are initiated. + Normally, only one payment can be initiated from the client. + """ + mpesa_b2c_payment = frappe.db.get_value( + "MPesa B2C Payment", + {"originatorconversationid": originator_conversation_id}, + ["name"], + as_dict=True, + ) + api_logger.info( + "Duplicate Request Encountered for: %s and originator conversation id: %s", + mpesa_b2c_payment.name, + originator_conversation_id, + ) + + +def extract_transaction_values( + result_parameters: dict, transaction_id: str +) -> dict[str, str | int]: + """ + Parses the ResultParameters of successful responses to the results callback endpoint + and returns the values as a dictionary. + Fields parsed include: TransactionAmount, TransactionReceipt, B2CRecipientIsRegisteredCustomer, + B2CChargesPaidAccountAvailableFunds, ReceiverPartyPublicName, TransactionCompletedDateTime, + B2CUtilityAccountAvailableFunds, B2CWorkingAccountAvailableFunds + """ + transaction_values = {} + + for item in result_parameters: + if item["Key"] == "TransactionAmount": + transaction_values["transaction_amount"] = item["Value"] + + elif item["Key"] == "TransactionReceipt" and item["Value"] == transaction_id: + transaction_values["transaction_id"] = item["Value"] + + elif item["Key"] == "B2CRecipientIsRegisteredCustomer": + transaction_values["recipient_is_registered_customer"] = item["Value"] + + elif item["Key"] == "B2CChargesPaidAccountAvailableFunds": + transaction_values["charges_paid_acct_avlbl_funds"] = item["Value"] + + elif item["Key"] == "ReceiverPartyPublicName": + transaction_values["receiver_public_name"] = item["Value"] + + elif item["Key"] == "TransactionCompletedDateTime": + transaction_datetime = datetime.datetime.strptime( + item["Value"], "%d.%m.%Y %H:%M:%S" + ).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] + transaction_values["transaction_completed_datetime"] = transaction_datetime + + elif item["Key"] == "B2CUtilityAccountAvailableFunds": + transaction_values["utility_acct_avlbl_funds"] = item["Value"] + + elif item["Key"] == "B2CWorkingAccountAvailableFunds": + transaction_values["working_acct_avlbl_funds"] = item["Value"] + + return transaction_values + + +def send_payload(payload: str, access_token: str, url: str) -> tuple[str, int]: + """Sends request to payment processing url with payload""" + try: + response = requests.post( + url, + data=payload, + headers={ + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + }, + timeout=60, + ) + + response.raise_for_status() # Raise HTTPError if status code >= 400 + + except requests.HTTPError: + api_logger.exception("Exception Encountered when sending payment request") + raise + + except requests.exceptions.ConnectionError: + api_logger.exception("Exception Encountered when sending payment request") + raise + + except Exception: + api_logger.exception("Exception Encountered") + raise + + frappe.msgprint("Payment Request Successful", title="Successful", indicator="green") + return response.text, response.status_code + + +def make_payment( + bearer_token: str, + b2c_settings: Document, + partial_payload: dict[str, str | int], + payment_document: Document, +) -> None: + """ + Handles making the Payment request. + This function sends the final response to the client after initiating the payment request. + """ + initiator_password = get_decrypted_password( + "MPesa B2C Settings", "MPesa B2C Settings", "initiator_password" + ) + payment_url = b2c_settings.get("payment_url") + certificate_relative_path = b2c_settings.get("certificate_file") + + certificate = get_certificate_file(certificate_relative_path) + + if isinstance(certificate, str): + security_credentials = openssl_encrypt_encode( + initiator_password.encode(), certificate + )[8:].decode() + + payload = generate_payload(b2c_settings, partial_payload, security_credentials) + + response, status_code = send_payload(payload, bearer_token, payment_url) + + update_doctype_single_values( + "MPesa B2C Payment", payment_document, "status", "Pending" + ) + + api_logger.info( + "Successful payment initiation for originator conversation id: %s with status code: %s", + partial_payload["OriginatorConversationID"], + status_code, + ) + frappe.response["message"] = "successful" + frappe.response["info"] = {"response": response, "status_code": status_code} + + # This return is important since without it, execution will continue + # to below and overwrite the "message" key in the response causing + # the client to enter an incorrect state + return + + frappe.response["message"] = "No certificate file found in server" + return + + +def update_doctype_single_values( + doctype: str, document_to_update: Document, field: str, new_value: str +) -> None: + """ + Updates the specified doctype's field with the specified values. + Note: Only one field is updated at a time + """ + frappe.db.set_value( + doctype, document_to_update.name, field, new_value, update_modified=True + ) + + api_logger.info( + "%s's %s's %s updated to %s", + doctype, + document_to_update.name, + field, + new_value, + ) + + +def save_transaction_to_database( + doctype: str, + update_values: dict[str, str | int | float], +) -> Document: + """ + Saves Transaction details to database after successful B2C Payment and returns the record + """ + update_values.update({"doctype": doctype}) + + transaction = frappe.get_doc(update_values) + transaction.insert(ignore_permissions=True) + + api_logger.info( + "Transaction ID: %s, originator conversation id: %s, amount: %s, transaction time: %s saved.", + update_values["transaction_id"], + update_values["mpesa_b2c_payment_name"], + update_values["transaction_amount"], + update_values["transaction_completed_datetime"], + ) + + return transaction + + +def validate_receiver_mobile_number(receiver: str) -> bool: + """Validates the Receiver's mobile number""" + receiver = receiver.replace("+", "").strip() + pattern1 = re.compile(r"^2547\d{8}$") + pattern2 = re.compile(r"(25410|25411)\d{7}$") + + if receiver.startswith("2547"): + return bool(pattern1.match(receiver)) + + return bool(pattern2.match(receiver)) diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/test_mpesa_b2c_payment.py b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/test_mpesa_b2c_payment.py new file mode 100644 index 0000000..32fa8e8 --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payment/test_mpesa_b2c_payment.py @@ -0,0 +1,201 @@ +# Copyright (c) 2023, Navari Limited and Contributors +# See license.txt + + +import frappe +import pymysql +from frappe.tests.utils import FrappeTestCase + +from ..csf_ke_custom_exceptions import ( + IncorrectStatusError, + InsufficientPaymentAmountError, + InvalidReceiverMobileNumberError, +) + + +def create_mpesa_b2c_payment() -> None: + """Create a valid b2c payment""" + if frappe.flags.test_events_created: + return + + frappe.set_user("Administrator") + + frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "commandid": "SalaryPayment", + "remarks": "test remarks", + "status": "Not Initiated", + "partyb": "254708993268", + "amount": 10, + "occassion": "Testing", + } + ).insert() + + frappe.flags.test_events_created = True + + +class TestMPesaB2CPayment(FrappeTestCase): + """B2C Payment Tests""" + + def setUp(self) -> None: + create_mpesa_b2c_payment() + + def tearDown(self) -> None: + frappe.set_user("Administrator") + + def test_invalid_receiver_number(self) -> None: + """Tests invalid receivers""" + with self.assertRaises(InvalidReceiverMobileNumberError): + frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "commandid": "SalaryPayment", + "remarks": "test remarks", + "status": "Not Initiated", + "partyb": "2547089932680", + "amount": 10, + "occassion": "Testing", + } + ).insert() + + frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "commandid": "SalaryPayment", + "remarks": "test remarks", + "status": "Not Initiated", + "partyb": "25470899326", + "amount": 10, + "occassion": "Testing", + } + ).insert() + + frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "commandid": "SalaryPayment", + "remarks": "test remarks", + "status": "Not Initiated", + "partyb": 254103456789, + "amount": 10, + "occassion": "Testing", + } + ).insert() + + frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "commandid": "SalaryPayment", + "remarks": "test remarks", + "status": "Not Initiated", + "partyb": 254113456789, + "amount": 10, + "occassion": "Testing", + } + ).insert() + + def test_empty_mandatory_fields(self) -> None: + """Tests when a mandatory field is not field""" + with self.assertRaises(frappe.exceptions.MandatoryError): + frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "commandid": "SalaryPayment", + "status": "Not Initiated", + "partyb": "254708993268", + "amount": 10, + "occassion": "Testing", + } + ).insert() + + frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "remarks": "testing remarks", + "status": "Not Initiated", + "partyb": "254708993268", + "amount": 10, + "occassion": "Testing", + } + ).insert() + + def test_insufficient_amount(self) -> None: + """Tests when an insufficient payment amount has been supplied""" + with self.assertRaises(InsufficientPaymentAmountError): + frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "commandid": "SalaryPayment", + "remarks": "test remarks", + "status": "Not Initiated", + "partyb": "254708993268", + "amount": 9.9999999, + "occassion": "Testing", + } + ).insert() + + def test_incredibly_large_amount(self) -> None: + """Tests when an incredibly large number has been supplied""" + large_number = 9999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999999 + with self.assertRaises(pymysql.err.DataError): + frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "commandid": "SalaryPayment", + "remarks": "test remarks", + "status": "Not Initiated", + "partyb": "254708993268", + "amount": large_number, + "occassion": "Testing", + } + ).insert() + + def test_valid_originator_conversation_id_length(self) -> None: + """Test that the created b2c settings have valid length uuids""" + new_mpesa_b2c_payment = frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "commandid": "SalaryPayment", + "remarks": "test remarks", + "status": "Not Initiated", + "partyb": "254708993268", + "amount": 10, + "occassion": "Testing", + } + ).insert() + + self.assertEqual(len(new_mpesa_b2c_payment.originatorconversationid), 36) + + def test_invalid_errored_status_no_code_or_error_description(self) -> None: + """Tests when status is set to errored without a description or error code""" + with self.assertRaises(IncorrectStatusError): + frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "commandid": "SalaryPayment", + "remarks": "test remarks", + "status": "Errored", + "partyb": "254708993268", + "amount": 10, + "occassion": "Testing", + } + ).insert() + + def test_status_set_to_not_initiated_when_not_supplied(self) -> None: + """ + Tests when status is not given on creation of a record. + Should be set to 'Not Initiated' + """ + new_doc = frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "commandid": "SalaryPayment", + "remarks": "test remarks", + "partyb": "254708993268", + "amount": 10, + "occassion": "Testing", + } + ).insert() + + self.assertEqual(new_doc.status, "Not Initiated") diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/__init__.py b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.js b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.js new file mode 100644 index 0000000..054ddf3 --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.js @@ -0,0 +1,8 @@ +// Copyright (c) 2023, Navari Ltd and contributors +// For license information, please see license.txt + +// frappe.ui.form.on("MPesa B2C Payments Transactions", { +// refresh(frm) { + +// }, +// }); diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.json b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.json new file mode 100644 index 0000000..6a13dad --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.json @@ -0,0 +1,127 @@ +{ + "actions": [], + "autoname": "field:transaction_id", + "creation": "2023-11-07 14:25:41.290770", + "default_view": "List", + "doctype": "DocType", + "editable_grid": 1, + "engine": "InnoDB", + "field_order": [ + "b2c_payment_name", + "section_break_ovxt", + "transaction_id", + "receiver_public_name", + "charges_paid_acct_avlbl_funds", + "utility_acct_avlbl_funds", + "column_break_iset", + "transaction_amount", + "recipient_is_registered_customer", + "working_acct_avlbl_funds", + "transaction_completed_datetime" + ], + "fields": [ + { + "fieldname": "b2c_payment_name", + "fieldtype": "Link", + "label": "B2C Payment Name", + "options": "MPesa B2C Payment", + "read_only": 1, + "reqd": 1 + }, + { + "fieldname": "section_break_ovxt", + "fieldtype": "Section Break" + }, + { + "fieldname": "transaction_id", + "fieldtype": "Data", + "in_list_view": 1, + "label": "Transaction ID", + "read_only": 1, + "reqd": 1, + "unique": 1 + }, + { + "fieldname": "receiver_public_name", + "fieldtype": "Data", + "label": "Receiver Party Public Name", + "read_only": 1, + "reqd": 1 + }, + { + "fieldname": "charges_paid_acct_avlbl_funds", + "fieldtype": "Float", + "in_list_view": 1, + "label": "B2C Charges Paid Account Available Funds", + "read_only": 1, + "reqd": 1 + }, + { + "fieldname": "utility_acct_avlbl_funds", + "fieldtype": "Float", + "label": "B2C Utility Account Available Funds", + "read_only": 1, + "reqd": 1 + }, + { + "fieldname": "column_break_iset", + "fieldtype": "Column Break" + }, + { + "fieldname": "transaction_amount", + "fieldtype": "Float", + "label": "Transaction Amount", + "read_only": 1, + "reqd": 1 + }, + { + "fieldname": "recipient_is_registered_customer", + "fieldtype": "Select", + "label": "B2C Recipient Is Registered Customer", + "options": "\nN\nY", + "read_only": 1, + "reqd": 1 + }, + { + "fieldname": "working_acct_avlbl_funds", + "fieldtype": "Float", + "in_list_view": 1, + "label": "B2C Working Account Available Funds", + "read_only": 1, + "reqd": 1 + }, + { + "fieldname": "transaction_completed_datetime", + "fieldtype": "Datetime", + "in_list_view": 1, + "label": "Transaction Completed Date Time", + "read_only": 1, + "reqd": 1 + } + ], + "index_web_pages_for_search": 1, + "links": [], + "modified": "2023-11-15 10:43:20.399768", + "modified_by": "Administrator", + "module": "MPesa B2C", + "name": "MPesa B2C Payments Transactions", + "naming_rule": "By fieldname", + "owner": "Administrator", + "permissions": [ + { + "create": 1, + "delete": 1, + "email": 1, + "export": 1, + "print": 1, + "read": 1, + "report": 1, + "role": "System Manager", + "share": 1, + "write": 1 + } + ], + "sort_field": "modified", + "sort_order": "DESC", + "states": [] +} diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.py b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.py new file mode 100644 index 0000000..00bdd1f --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/mpesa_b2c_payments_transactions.py @@ -0,0 +1,59 @@ +# Copyright (c) 2023, Navari Limited and contributors +# For license information, please see license.txt + +import frappe +from frappe.model.document import Document + +from .. import api_logger +from ..csf_ke_custom_exceptions import ( + InformationMismatchError, + UnExistentB2CPaymentRecordError, +) + + +class MPesaB2CPaymentsTransactions(Document): + """B2C Payments Transactions""" + + def validate(self) -> None: + """B2C Payments Transactions validations""" + + if self.b2c_payment_name: + b2c_payment = frappe.db.get_value( + "MPesa B2C Payment", + {"name": self.b2c_payment_name}, + ["name", "status", "amount"], + as_dict=True, + ) + + if not b2c_payment: + api_logger.error( + "The B2C payment record with originator conversation ID: %s does not exist", + self.b2c_payment_name, + ) + raise UnExistentB2CPaymentRecordError( + f"The B2C payment record with originator conversation ID: {self.b2c_payment_name} does not exist", + ) + + if ( + b2c_payment.status == "Errored" + or b2c_payment.status == "Not Initiated" + or b2c_payment.status == "Timed-Out" + or b2c_payment.status == "Pending" + ): + api_logger.error( + "Incorrect B2C Payment Status: %s for B2C Payment: %s", + b2c_payment.status, + self.b2c_payment_name, + ) + raise InformationMismatchError( + f"Incorrect B2C Payment Status: {b2c_payment.status} for B2C Payment: {self.b2c_payment_name}" + ) + + if self.transaction_amount != b2c_payment.amount: + api_logger.error( + "Incorrect Transaction and B2C Payment Amount for B2C payment: %s", + self.b2c_payment_name, + ) + raise InformationMismatchError( + f"Incorrect Transaction and B2C Payment Amount for B2C payment: {self.b2c_payment_name}" + ) diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/test_mpesa_b2c_payments_transactions.py b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/test_mpesa_b2c_payments_transactions.py new file mode 100644 index 0000000..aa882e9 --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_payments_transactions/test_mpesa_b2c_payments_transactions.py @@ -0,0 +1,141 @@ +# Copyright (c) 2023, Navari Limited and Contributors +# See license.txt + +import datetime +import random +from uuid import uuid4 + +import frappe +from frappe.tests.utils import FrappeTestCase + +from ..csf_ke_custom_exceptions import InformationMismatchError + +ORIGINATOR_CONVERSATION_ID = str(uuid4()) +ORIGINATOR_CONVERSATION_ID_2 = str(uuid4()) + + +def create_b2c_payment_transaction() -> None: + """Create a valid b2c payment""" + if frappe.flags.test_events_created: + return + + frappe.set_user("Administrator") + + doc = frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "originatorconversationid": ORIGINATOR_CONVERSATION_ID, + "commandid": "SalaryPayment", + "remarks": "test remarks", + "status": "Paid", + "partyb": "254712345678", + "amount": 10, + "occassion": "Testing", + } + ).insert() + + frappe.get_doc( + { + "doctype": "MPesa B2C Payment", + "originatorconversationid": ORIGINATOR_CONVERSATION_ID_2, + "commandid": "SalaryPayment", + "remarks": "test remarks", + "status": random.choice(["Pending", "Not Initiated", "Timed-Out"]), + "partyb": "254712345678", + "amount": 10, + "occassion": "Testing", + } + ).insert() + + frappe.get_doc( + { + "doctype": "MPesa B2C Payments Transactions", + "b2c_payment_name": doc.name, + "transaction_id": 951753654, + "transaction_amount": 10, + "receiver_public_name": "Jane Doe", + "recipient_is_registered_customer": "Y", + "charges_paid_acct_avlbl_funds": 100, + "working_acct_avlbl_funds": 1000000, + "utility_acct_avlbl_funds": 10000000, + "transaction_completed_datetime": datetime.datetime.now(), + } + ).insert() + + frappe.flags.test_events_created = True + + +class TestMPesaB2CPaymentsTransactions(FrappeTestCase): + """B2C Payments Transactions Tests""" + + def setUp(self) -> None: + create_b2c_payment_transaction() + + def tearDown(self) -> None: + frappe.set_user("Administrator") + + def test_mismatch_in_amount(self) -> None: + """Tests a mismatch in the amount paid and the transaction amount""" + payment = frappe.db.get_value( + "MPesa B2C Payment", + {"originatorconversationid": ORIGINATOR_CONVERSATION_ID}, + ["name"], + as_dict=True, + ) + with self.assertRaises(InformationMismatchError): + frappe.get_doc( + { + "doctype": "MPesa B2C Payments Transactions", + "b2c_payment_name": payment.name, + "transaction_id": random.randint(1000000, 100000000), + "transaction_amount": 9.9999, + "receiver_public_name": "Jane Doe", + "recipient_is_registered_customer": "Y", + "charges_paid_acct_avlbl_funds": 100, + "working_acct_avlbl_funds": 1000000, + "utility_acct_avlbl_funds": 10000000, + "transaction_completed_datetime": datetime.datetime.now(), + } + ).insert() + + def test_mismatch_in_payment_status(self) -> None: + """Tests creating a transaction when payment status is not 'Paid'""" + payment = frappe.db.get_value( + "MPesa B2C Payment", + {"originatorconversationid": ORIGINATOR_CONVERSATION_ID_2}, + ["name"], + as_dict=True, + ) + with self.assertRaises(InformationMismatchError): + frappe.get_doc( + { + "doctype": "MPesa B2C Payments Transactions", + "b2c_payment_name": payment.name, + "transaction_id": random.randint(1000000, 100000000), + "transaction_amount": 9.9999, + "receiver_public_name": "Jane Doe", + "recipient_is_registered_customer": "Y", + "charges_paid_acct_avlbl_funds": 100, + "working_acct_avlbl_funds": 1000000, + "utility_acct_avlbl_funds": 10000000, + "transaction_completed_datetime": datetime.datetime.now(), + } + ).insert() + + def test_creating_transaction_for_non_existent_payment(self) -> None: + """Tests creating transaction for non-existent payment record""" + with self.assertRaises(frappe.exceptions.LinkValidationError): + frappe.get_doc( + { + "doctype": "MPesa B2C Payments Transactions", + "b2c_payment_name": "MPESA-B2C-0000", + "transaction_id": random.randint(1000000, 100000000), + "transaction_amount": 9.9999, + "receiver_public_name": "Jane Doe", + "recipient_is_registered_customer": "Y", + "charges_paid_acct_avlbl_funds": 100, + "working_acct_avlbl_funds": 1000000, + "utility_acct_avlbl_funds": 10000000, + "transaction_completed_datetime": datetime.datetime.now(), + } + ).insert() diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/__init__.py b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.js b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.js new file mode 100644 index 0000000..73f914d --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.js @@ -0,0 +1,57 @@ +// Copyright (c) 2023, Navari Limited and contributors +// For license information, please see license.txt + +frappe.ui.form.on("MPesa B2C Settings", { + validate: function (frm) { + if ( + frm.doc.results_url && + frm.doc.queue_timeout_url && + frm.doc.authorization_url && + frm.doc.payment_url + ) { + if ( + !( + validateURL(frm.doc.results_url) && + validateURL(frm.doc.queue_timeout_url) && + validateURL(frm.doc.authorization_url) && + validateURL(frm.doc.payment_url) + ) + ) { + frappe.msgprint({ + message: __("The URLs Registered are not valid. Please review them"), + indicator: "red", + title: "Validation Error", + }); + frappe.validated = false; + } + } + + if (frm.doc.certificate_file) { + if (!frm.doc.certificate_file.endsWith(".cer")) { + frappe.msgprint({ + message: __( + `The certificate file uploaded is not valid. Please upload a .CER file` + ), + indicator: "red", + title: "Validation Error", + }); + frappe.validated = false; + } + } + }, +}); + +function validateURL(url) { + // validates the input parameter to a valid URL. + // url: string, returnType: boolean + const pattern = new RegExp( + "^((https?|ftp|file):\\/\\/)?" + + "((([a-zA-Z\\d]([a-zA-Z\\d-]*[a-zA-Z\\d])*)\\.)+[a-zA-Z]{2,}|" + + "((\\d{1,3}\\.){3}\\d{1,3}))" + + "(\\:\\d+)?(\\/[-a-zA-Z\\d%_.~+]*)*" + + "(\\?[;&a-z\\d%_.~+=-]*)?" + + "(\\#[-a-z\\d_]*)?$", + "i" + ); + return pattern.test(url); +} diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.json b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.json new file mode 100644 index 0000000..75bfcbf --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.json @@ -0,0 +1,126 @@ +{ + "actions": [], + "allow_rename": 1, + "creation": "2023-10-30 10:51:27.687329", + "default_view": "List", + "doctype": "DocType", + "editable_grid": 1, + "engine": "InnoDB", + "field_order": [ + "title", + "consumer_key", + "initiator_name", + "results_url", + "authorization_url", + "column_break_it7j", + "organisation_shortcode", + "consumer_secret", + "initiator_password", + "queue_timeout_url", + "payment_url", + "section_break_tos4", + "certificate_file" + ], + "fields": [ + { + "fieldname": "title", + "fieldtype": "Data", + "label": "Title" + }, + { + "fieldname": "consumer_key", + "fieldtype": "Data", + "in_list_view": 1, + "label": "Consumer Key", + "reqd": 1 + }, + { + "fieldname": "initiator_name", + "fieldtype": "Data", + "label": "Initiator Name" + }, + { + "description": "Mandatory Format: (http|https)://domain/api/method/csf_ke.csf_ke.doctype.b2c_payment.b2c_payment.results_callback_url", + "fieldname": "results_url", + "fieldtype": "Data", + "label": "Results URL" + }, + { + "description": "Mandatory Format: (http|https)://domain.*", + "fieldname": "authorization_url", + "fieldtype": "Data", + "label": "Authorization URL", + "reqd": 1 + }, + { + "fieldname": "column_break_it7j", + "fieldtype": "Column Break" + }, + { + "fieldname": "organisation_shortcode", + "fieldtype": "Data", + "label": "Organisation ShortCode", + "reqd": 1 + }, + { + "fieldname": "consumer_secret", + "fieldtype": "Password", + "in_list_view": 1, + "label": "Consumer Secret", + "reqd": 1 + }, + { + "fieldname": "initiator_password", + "fieldtype": "Password", + "label": "Initiator Password", + "reqd": 1 + }, + { + "description": "Mandatory Format: (http|https)://domain/api/method/csf_ke.csf_ke.doctype.b2c_payment.b2c_payment.queue_timeout_url", + "fieldname": "queue_timeout_url", + "fieldtype": "Data", + "label": "Queue TimeOut URL" + }, + { + "description": "Mandatory Format: (http|https)://domain.*", + "fieldname": "payment_url", + "fieldtype": "Data", + "label": "Payment URL", + "reqd": 1 + }, + { + "fieldname": "section_break_tos4", + "fieldtype": "Section Break" + }, + { + "description": "A Certificate file (.cer or .pem) received from Safaricom.", + "fieldname": "certificate_file", + "fieldtype": "Attach", + "label": "Authorisation Certificate File", + "reqd": 1 + } + ], + "index_web_pages_for_search": 1, + "issingle": 1, + "links": [], + "modified": "2023-11-15 10:42:57.886201", + "modified_by": "Administrator", + "module": "MPesa B2C", + "name": "MPesa B2C Settings", + "owner": "Administrator", + "permissions": [ + { + "create": 1, + "delete": 1, + "email": 1, + "print": 1, + "read": 1, + "role": "System Manager", + "share": 1, + "write": 1 + } + ], + "sort_field": "modified", + "sort_order": "DESC", + "states": [] +} diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.py b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.py new file mode 100644 index 0000000..741d7a7 --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/mpesa_b2c_settings.py @@ -0,0 +1,64 @@ +# Copyright (c) 2023, Navari Limited and contributors +# For license information, please see license.txt + +import re + +from frappe.model.document import Document + +from ..csf_ke_custom_exceptions import InvalidURLError + +from .. import api_logger +from ..csf_ke_custom_exceptions import ( + InvalidAuthenticationCertificateFileError, +) + + +class MPesaB2CSettings(Document): + """MPesa B2C Settings Doctype""" + + def validate(self) -> None: + """Validate upon creating a record of the B2C Settings""" + if ( + self.results_url + and self.queue_timeout_url + and self.authorization_url + and self.payment_url + ): + if not ( + validate_url(self.results_url) + and validate_url(self.queue_timeout_url) + and validate_url(self.authorization_url) + and validate_url(self.payment_url) + ): + api_logger.error( + "The URLs Registered are not valid. Please review them" + ) + raise InvalidURLError( + "The URLs Registered are not valid. Please review them" + ) + + if self.certificate_file: + if not ( + self.certificate_file.endswith(".cer") + or self.certificate_file.endswith(".pem") + ): + api_logger.error("Invalid Authentication Certificate file uploaded") + raise InvalidAuthenticationCertificateFileError( + "Invalid Authentication Certificate file uploaded" + ) + + +def validate_url(url: str) -> bool: + """ + Validates the input parameter to a valid URL. + """ + pattern = re.compile( + r"^((https?|ftp|file):\/\/)?" + + r"((([a-zA-Z\d]([a-zA-Z\d-]*[a-zA-Z\d])*)\.)+[a-zA-Z]{2,}|" + + r"((\d{1,3}\.){3}\d{1,3}))" + + r"(:\d+)?(\/[-a-zA-Z\d%_.~+]*)*" + + r"(\?[;&a-z\d%_.~+=-]*)?" + + r"(\#[-a-z\d_]*)?$", + re.IGNORECASE, + ) + return bool(pattern.match(url)) diff --git a/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/test_mpesa_b2c_settings.py b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/test_mpesa_b2c_settings.py new file mode 100644 index 0000000..b664ea8 --- /dev/null +++ b/navari_mpesa_b2c/mpesa_b2c/doctype/mpesa_b2c_settings/test_mpesa_b2c_settings.py @@ -0,0 +1,141 @@ +# Copyright (c) 2023, Navari Limited and Contributors +# See license.txt + +import frappe +from frappe.tests.utils import FrappeTestCase + +from ..mpesa_b2c_payment.mpesa_b2c_payment import get_b2c_settings, get_certificate_file +from ..csf_ke_custom_exceptions import ( + InvalidAuthenticationCertificateFileError, + InvalidURLError, +) + + +def create_b2c_settings(): + """Setup context for tests""" + if frappe.flags.test_events_created: + return + + frappe.set_user("Administrator") + + # Create a valid singles record during setUp context + frappe.get_doc( + { + "doctype": "MPesa B2C Settings", + "consumer_key": "1234567890", + "initiator_name": "tester", + "results_url": "https://example.com/api/method/handler", + "authorization_url": "https://example.com/api/method/handler", + "organisation_shortcode": "951753", + "consumer_secret": "secret", + "initiator_password": "password", + "queue_timeout_url": "https://example.com/api/method/handler", + "payment_url": "https://example.com/api/method/handler", + } + ).insert(ignore_mandatory=True) + + frappe.flags.test_events_created = True + + +class TestMPesaB2CSettings(FrappeTestCase): + """MPesa B2C Settings Tests""" + + def setUp(self) -> None: + create_b2c_settings() + + def tearDown(self) -> None: + frappe.set_user("Administrator") + + def test_invalid_urls_in_b2c_settings(self) -> None: + """Tests for cases when an invalid url is supplied""" + with self.assertRaises(InvalidURLError): + frappe.get_doc( + { + "doctype": "MPesa B2C Settings", + "consumer_key": "1234567890", + "initiator_name": "tester", + "results_url": "https://example.com/api/method/handler", + "authorization_url": "https://example.com/api/method/handler", + "organisation_shortcode": "951753", + "consumer_secret": "secret", + "initiator_password": "password", + "queue_timeout_url": "https://example.com/api/method/handler", + "payment_url": "jkl", + } + ).insert(ignore_mandatory=True) + + def test_override_b2c_settings(self) -> None: + """Test instances where the B2C Settings have been overridden""" + frappe.get_doc( + { + "doctype": "MPesa B2C Settings", + "consumer_key": "987654321", + "initiator_name": "tester2", + "results_url": "https://example2.com/api/method/handler", + "authorization_url": "https://example2.com/api/method/handler", + "organisation_shortcode": "951753", + "consumer_secret": "secret", + "initiator_password": "password", + "queue_timeout_url": "https://example2.com/api/method/handler", + "payment_url": "https://example2.com/api/method/handler", + } + ).insert(ignore_mandatory=True) + + new_doc = frappe.db.get_singles_dict("MPesa B2C Settings") + + self.assertEqual(new_doc.initiator_name, "tester2") + self.assertEqual(new_doc.payment_url, "https://example2.com/api/method/handler") + self.assertEqual(new_doc.consumer_key, "987654321") + + def test_invalid_certificate_file(self) -> None: + """Tests when a user uploads an invalid certificate file""" + with self.assertRaises(InvalidAuthenticationCertificateFileError): + frappe.get_doc( + { + "doctype": "MPesa B2C Settings", + "consumer_key": "987654321", + "initiator_name": "tester2", + "results_url": "https://example2.com/api/method/handler", + "authorization_url": "https://example2.com/api/method/handler", + "organisation_shortcode": "951753", + "consumer_secret": "secret", + "initiator_password": "password", + "queue_timeout_url": "https://example2.com/api/method/handler", + "payment_url": "https://example2.com/api/method/handler", + "certificate_file": "/files/AuthorizationCertificate", + } + ).insert() + + def test_get_b2c_settings_function(self) -> None: + """Tests the get_b2c_settings() function from the b2c payment module""" + b2c_settings = frappe.db.get_singles_dict("MPesa B2C Settings") + consumer_key, consumer_secret, authorization_url = get_b2c_settings( + b2c_settings + ) + + self.assertEqual(consumer_key, "1234567890") + self.assertEqual(authorization_url, "https://example.com/api/method/handler") + self.assertEqual(consumer_secret, "secret") + + def test_get_certificate_file_function(self) -> None: + """Tests the get_certificate_file() function from the b2c payment module""" + certificate_file_path = "/files/AuthorizationCertificate.cer" + frappe.get_doc( + { + "doctype": "MPesa B2C Settings", + "consumer_key": "1234567890", + "initiator_name": "tester", + "results_url": "https://example.com/api/method/handler", + "authorization_url": "https://example.com/api/method/handler", + "organisation_shortcode": "951753", + "consumer_secret": "secret", + "initiator_password": "password", + "queue_timeout_url": "https://example.com/api/method/handler", + "payment_url": "https://example.com/api/method/handler", + "certificate_file": certificate_file_path, + } + ).insert() + + certificate = get_certificate_file(certificate_file_path) + + self.assertTrue(certificate, certificate.endswith(certificate_file_path))