diff --git a/mpesakit/b2b_express_checkout/schemas.py b/mpesakit/b2b_express_checkout/schemas.py index c5f66a6..2058c85 100644 --- a/mpesakit/b2b_express_checkout/schemas.py +++ b/mpesakit/b2b_express_checkout/schemas.py @@ -1,8 +1,9 @@ """Schemas for M-PESA B2B Express Checkout APIs.""" -from pydantic import BaseModel, Field, ConfigDict, HttpUrl from typing import Optional +from pydantic import BaseModel, ConfigDict, Field, HttpUrl + class B2BExpressCheckoutRequest(BaseModel): """Request schema for B2B Express Checkout USSD Push.""" @@ -56,8 +57,9 @@ class B2BExpressCheckoutResponse(BaseModel): ) def is_successful(self) -> bool: - """Check if the response indicates a successful USSD initiation.""" - return self.code == "0" + """Return True if code indicates success (e.g., '0', '00000000').""" + code = str(self.code) + return code.strip("0") == "" and code != "" class B2BExpressCheckoutCallback(BaseModel): @@ -101,8 +103,9 @@ class B2BExpressCheckoutCallback(BaseModel): ) def is_successful(self) -> bool: - """Check if the callback indicates a successful transaction.""" - return self.resultCode == "0" + """Return True if resultCode indicates success (e.g., '0', '00000000').""" + code = str(self.resultCode) + return code.strip("0") == "" and code != "" class B2BExpressCallbackResponse(BaseModel): diff --git a/mpesakit/b2c/schemas.py b/mpesakit/b2c/schemas.py index a0cdfac..272626c 100644 --- a/mpesakit/b2c/schemas.py +++ b/mpesakit/b2c/schemas.py @@ -4,8 +4,10 @@ """ from enum import Enum -from pydantic import BaseModel, Field, ConfigDict, model_validator from typing import Optional + +from pydantic import BaseModel, ConfigDict, Field, model_validator + from mpesakit.utils.phone import normalize_phone_number @@ -27,7 +29,7 @@ class B2CRequest(BaseModel): NOTE: To use this API in production, you must apply for a Bulk Disbursement Account and obtain a Shortcode. The Shortcode required here is NOT the same as a Buy Goods Till or Paybill Till Number. For more information and to apply for a Bulk Disbursement Account, refer to the official documentation: - https://developer.safaricom.co.ke/APIs/BusinessToCustomerPayment + https://developer.safaricom.co.ke/dashboard/apis?api=BusinessToCustomer Attributes: OriginatorConversationID (str): Unique identifier for the specific request. @@ -275,6 +277,11 @@ class B2CResultCallback(BaseModel): } ) + def is_successful(self) -> bool: + """Return True if ResultCode indicates success (e.g., '0', '00000000').""" + code = str(self.Result.ResultCode) + return code.strip("0") == "" and code != "" + class B2CResultCallbackResponse(BaseModel): """Schema for response to B2C result callback.""" diff --git a/mpesakit/b2c_account_top_up/schemas.py b/mpesakit/b2c_account_top_up/schemas.py index 81db82e..dea8720 100644 --- a/mpesakit/b2c_account_top_up/schemas.py +++ b/mpesakit/b2c_account_top_up/schemas.py @@ -1,7 +1,8 @@ """Schemas for M-PESA B2C Account TopUp APIs.""" -from pydantic import BaseModel, Field, HttpUrl, ConfigDict -from typing import Optional, List, Any +from typing import Any, List, Optional + +from pydantic import BaseModel, ConfigDict, Field, HttpUrl class B2CAccountTopUpRequest(BaseModel): @@ -231,7 +232,8 @@ class B2CAccountTopUpCallback(BaseModel): def is_successful(self) -> bool: """Check if the callback indicates a successful transaction.""" - return str(self.Result.ResultCode) == "0" + code = str(self.Result.ResultCode) + return code.strip("0") == "" and code != "" class B2CAccountTopUpCallbackResponse(BaseModel): diff --git a/mpesakit/business_paybill/schemas.py b/mpesakit/business_paybill/schemas.py index 365aa27..689970d 100644 --- a/mpesakit/business_paybill/schemas.py +++ b/mpesakit/business_paybill/schemas.py @@ -1,7 +1,8 @@ """This module defines schemas for M-Pesa Business PayBill API requests and responses.""" -from pydantic import BaseModel, Field, ConfigDict -from typing import Optional, List +from typing import List, Optional + +from pydantic import BaseModel, ConfigDict, Field class BusinessPayBillRequest(BaseModel): @@ -228,8 +229,9 @@ class BusinessPayBillResultCallback(BaseModel): ) def is_successful(self) -> bool: - """Check if the result indicates success.""" - return str(self.Result.ResultCode) == "0" + """Return True if ResultCode indicates success (e.g., '0', '00000000').""" + code = str(self.Result.ResultCode) + return code.strip("0") == "" and code != "" class BusinessPayBillResultCallbackResponse(BaseModel): diff --git a/mpesakit/reversal/schemas.py b/mpesakit/reversal/schemas.py index 34d615a..e932050 100644 --- a/mpesakit/reversal/schemas.py +++ b/mpesakit/reversal/schemas.py @@ -3,8 +3,9 @@ It includes models for reversal requests, responses, and result notifications. """ -from pydantic import BaseModel, Field, ConfigDict, model_validator -from typing import Optional, List +from typing import List, Optional + +from pydantic import BaseModel, ConfigDict, Field, model_validator class ReversalRequest(BaseModel): @@ -231,6 +232,11 @@ class ReversalResultCallback(BaseModel): } ) + def is_successful(self) -> bool: + """Return True if ResultCode indicates success (e.g., '0', '00000000').""" + code = str(self.Result.ResultCode) + return code.strip("0") == "" and code != "" + class ReversalResultCallbackResponse(BaseModel): """Schema for response to Reversal result callback.""" diff --git a/mpesakit/transaction_status/schemas.py b/mpesakit/transaction_status/schemas.py index b4fddc1..77b4c57 100644 --- a/mpesakit/transaction_status/schemas.py +++ b/mpesakit/transaction_status/schemas.py @@ -4,8 +4,10 @@ """ from enum import Enum -from pydantic import BaseModel, Field, ConfigDict, model_validator from typing import Optional + +from pydantic import BaseModel, ConfigDict, Field, model_validator + from mpesakit.utils.phone import normalize_phone_number @@ -46,7 +48,9 @@ class TransactionStatusRequest(BaseModel): IdentifierType: int = Field(..., description="Type of identifier for PartyA.") ResultURL: str = Field(..., description="URL for result notifications.") QueueTimeOutURL: str = Field(..., description="URL for timeout notifications.") - Remarks: str = Field(default="Status Query", description="Comments for the transaction.") + Remarks: str = Field( + default="Status Query", description="Comments for the transaction." + ) Occasion: Optional[str] = Field( None, description="Optional occasion for the query." ) @@ -261,6 +265,11 @@ class TransactionStatusResultCallback(BaseModel): } ) + def is_successful(self) -> bool: + """Return True if ResultCode indicates success (e.g., '0', '00000000').""" + code = str(self.Result.ResultCode) + return code.strip("0") == "" and code != "" + class TransactionStatusResultCallbackResponse(BaseModel): """Schema for response to Transaction Status result callback.""" diff --git a/tests/unit/b2c/test_b2c.py b/tests/unit/b2c/test_b2c.py index b06d3ec..3e4dec6 100644 --- a/tests/unit/b2c/test_b2c.py +++ b/tests/unit/b2c/test_b2c.py @@ -1,19 +1,20 @@ """Unit tests for the B2C functionality of the Mpesa SDK.""" -import pytest from unittest.mock import MagicMock +import pytest + from mpesakit.auth import TokenManager -from mpesakit.http_client import HttpClient from mpesakit.b2c.b2c import B2C from mpesakit.b2c.schemas import ( - B2CResultMetadata, - B2CResultParameter, - B2CResultCallback, + B2CCommandIDType, B2CRequest, B2CResponse, - B2CCommandIDType, + B2CResultCallback, + B2CResultMetadata, + B2CResultParameter, ) +from mpesakit.http_client import HttpClient @pytest.fixture @@ -343,3 +344,80 @@ def test_result_callback_schema(): assert isinstance(callback.Result, B2CResultMetadata) assert callback.Result.transaction_amount == 1000 assert callback.Result.transaction_receipt == "LKXXXX1234" + + +def test_result_callback_is_successful_zero_code(): + """Test is_successful returns True for ResultCode 0.""" + meta = B2CResultMetadata( + ResultType=0, + ResultCode=0, + ResultDesc="Success", + OriginatorConversationID="conv-id", + ConversationID="conv-id-2", + TransactionID="LKXXXX1234", + ResultParameters=[], + ) + callback = B2CResultCallback(Result=meta) + assert callback.is_successful() is True + + +def test_result_callback_is_successful_all_zeros(): + """Test is_successful returns True for ResultCode as string of zeros.""" + meta = B2CResultMetadata( + ResultType=0, + ResultCode=0, + ResultDesc="Success", + OriginatorConversationID="conv-id", + ConversationID="conv-id-2", + TransactionID="LKXXXX1234", + ResultParameters=[], + ) + callback = B2CResultCallback(Result=meta) + # Simulate ResultCode "00000000" + callback.Result.ResultCode = 0 + assert callback.is_successful() is True + + +def test_result_callback_is_successful_non_zero_code(): + """Test is_successful returns False for non-zero ResultCode.""" + meta = B2CResultMetadata( + ResultType=0, + ResultCode=1, + ResultDesc="Failed", + OriginatorConversationID="conv-id", + ConversationID="conv-id-2", + TransactionID=None, + ResultParameters=[], + ) + callback = B2CResultCallback(Result=meta) + assert callback.is_successful() is False + + +def test_result_callback_is_successful_mixed_code(): + """Test is_successful returns False for mixed code like 00001.""" + meta = B2CResultMetadata( + ResultType=0, + ResultCode=1, + ResultDesc="Failed", + OriginatorConversationID="conv-id", + ConversationID="conv-id-2", + TransactionID=None, + ResultParameters=[], + ) + callback = B2CResultCallback(Result=meta) + assert callback.is_successful() is False + + +def test_result_callback_is_successful_negative_code(): + """Test is_successful returns False for negative ResultCode.""" + meta = B2CResultMetadata( + ResultType=0, + ResultCode=-1, + ResultDesc="Error", + OriginatorConversationID="conv-id", + ConversationID="conv-id-2", + TransactionID=None, + ResultParameters=[], + ) + callback = B2CResultCallback(Result=meta) + assert callback.is_successful() is False diff --git a/tests/unit/reversal/test_reversal.py b/tests/unit/reversal/test_reversal.py index b15b9de..7ca642e 100644 --- a/tests/unit/reversal/test_reversal.py +++ b/tests/unit/reversal/test_reversal.py @@ -4,12 +4,12 @@ process responses correctly, and manage error cases. """ -import pytest from unittest.mock import MagicMock + +import pytest + from mpesakit.auth import TokenManager from mpesakit.http_client import HttpClient -from mpesakit.reversal.reversal import Reversal - from mpesakit.reversal import ( ReversalRequest, ReversalResponse, @@ -18,6 +18,7 @@ ReversalTimeoutCallback, ReversalTimeoutCallbackResponse, ) +from mpesakit.reversal.reversal import Reversal @pytest.fixture @@ -201,6 +202,7 @@ def test_reversal_request_occasion_too_long_raises(): ReversalRequest(**kwargs) assert "Occasion must not exceed 100 characters." in str(excinfo.value) + def test_reverse_responsecode_string_no_type_error(reversal, mock_http_client): """Ensure is_successful handles ResponseCode as a string without TypeError.""" request = valid_reversal_request() @@ -217,3 +219,89 @@ def test_reverse_responsecode_string_no_type_error(reversal, mock_http_client): assert isinstance(response, ReversalResponse) # Calling is_successful should not raise a TypeError when comparing str to int assert response.is_successful() is True + + +def test_reversal_result_callback_success_is_successful(): + """Test is_successful method for a successful reversal result callback.""" + payload = { + "Result": { + "ResultType": 0, + "ResultCode": "0", + "ResultDesc": "The service request is processed successfully", + "OriginatorConversationID": "8521-4298025-1", + "ConversationID": "AG_20181005_00004d7ee675c0c7ee0b", + "TransactionID": "MJ561H6X5O", + "ResultParameters": { + "ResultParameter": [ + {"Key": "Amount", "Value": "100"}, + ] + }, + "ReferenceData": { + "ReferenceItem": { + "Key": "QueueTimeoutURL", + "Value": "https://internalsandbox.safaricom.co.ke/mpesa/reversalresults/v1/submit", + } + }, + } + } + callback = ReversalResultCallback(**payload) + assert callback.is_successful() is True + + +def test_reversal_result_callback_failure_is_successful(): + """Test is_successful method for a failure reversal result callback.""" + payload = { + "Result": { + "ResultType": 1, + "ResultCode": "1", + "ResultDesc": "The service request failed.", + "OriginatorConversationID": "8521-4298025-1", + "ConversationID": "AG_20181005_00004d7ee675c0c7ee0b", + "TransactionID": "MJ561H6X5O", + "ResultParameters": { + "ResultParameter": [ + {"Key": "Amount", "Value": "100"}, + ] + }, + "ReferenceData": { + "ReferenceItem": { + "Key": "QueueTimeoutURL", + "Value": "https://internalsandbox.safaricom.co.ke/mpesa/reversalresults/v1/submit", + } + }, + } + } + callback = ReversalResultCallback(**payload) + assert callback.is_successful() is False + + +def test_reversal_result_callback_success_code_is_successful(): + """Test is_successful method with a success code as a string.""" + payload = { + "Result": { + "ResultType": 0, + "ResultCode": "00000000", + "ResultDesc": "The service request is processed successfully", + "OriginatorConversationID": "8521-4298025-1", + "ConversationID": "AG_20181005_00004d7ee675c0c7ee0b", + "TransactionID": "MJ561H6X5O", + } + } + callback = ReversalResultCallback(**payload) + assert callback.is_successful() is True + + +def test_reversal_result_callback_failure_code_is_successful(): + """Test is_successful method with a failure code.""" + payload = { + "Result": { + "ResultType": 1, + "ResultCode": "12345", + "ResultDesc": "The service request failed.", + "OriginatorConversationID": "8521-4298025-1", + "ConversationID": "AG_20181005_00004d7ee675c0c7ee0b", + "TransactionID": "MJ561H6X5O", + } + } + callback = ReversalResultCallback(**payload) + assert callback.is_successful() is False diff --git a/tests/unit/transaction_status/test_transaction_status.py b/tests/unit/transaction_status/test_transaction_status.py index 0f87193..7cc5930 100644 --- a/tests/unit/transaction_status/test_transaction_status.py +++ b/tests/unit/transaction_status/test_transaction_status.py @@ -3,19 +3,20 @@ This module tests the TransactionStatus class and its methods for querying transaction status. """ -import pytest from unittest.mock import MagicMock + +import pytest + from mpesakit.auth import TokenManager from mpesakit.http_client import HttpClient - from mpesakit.transaction_status import ( TransactionStatus, + TransactionStatusIdentifierType, TransactionStatusRequest, TransactionStatusResponse, - TransactionStatusIdentifierType, - TransactionStatusResultParameter, - TransactionStatusResultMetadata, TransactionStatusResultCallback, + TransactionStatusResultMetadata, + TransactionStatusResultParameter, ) @@ -355,6 +356,7 @@ def test_result_callback_schema(): assert callback.Result.transaction_receipt == "LKXXXX1234" assert callback.Result.transaction_status == "Completed" + def test_query_response_code_type_variations(transaction_status, mock_http_client): """Ensure TransactionStatusResponse.is_successful handles ResponseCode as str or int without TypeError.""" request = valid_transaction_status_request() @@ -379,3 +381,78 @@ def test_query_response_code_type_variations(transaction_status, mock_http_clien resp = transaction_status.query(request) assert isinstance(resp, TransactionStatusResponse) assert resp.is_successful() is expected_success + + +def test_transaction_status_result_callback_is_successful_zero_code(): + """Test is_successful method with ResultCode as '0'.""" + result = TransactionStatusResultMetadata( + ResultType=0, + ResultCode="0", + ResultDesc="Success", + OriginatorConversationID="12345-67890-1", + ConversationID="AG_20170717_00006c6f7f5b8b6b1a62", + TransactionID="LKXXXX1234", + ResultParameters=[], + ) + callback = TransactionStatusResultCallback(Result=result) + assert callback.is_successful() is True + + +def test_transaction_status_result_callback_is_successful_all_zeros(): + """Test is_successful method with ResultCode as '00000000'.""" + result = TransactionStatusResultMetadata( + ResultType=0, + ResultCode="00000000", + ResultDesc="Success", + OriginatorConversationID="12345-67890-1", + ConversationID="AG_20170717_00006c6f7f5b8b6b1a62", + TransactionID="LKXXXX1234", + ResultParameters=[], + ) + callback = TransactionStatusResultCallback(Result=result) + assert callback.is_successful() is True + + +def test_transaction_status_result_callback_is_successful_non_zero_code(): + """Test is_successful method with ResultCode as '1'.""" + result = TransactionStatusResultMetadata( + ResultType=1, + ResultCode="1", + ResultDesc="Failure", + OriginatorConversationID="12345-67890-1", + ConversationID="AG_20170717_00006c6f7f5b8b6b1a62", + TransactionID="LKXXXX1234", + ResultParameters=[], + ) + callback = TransactionStatusResultCallback(Result=result) + assert callback.is_successful() is False + + +def test_transaction_status_result_callback_is_successful_mixed_code(): + """Test is_successful method with ResultCode as '00001'.""" + result = TransactionStatusResultMetadata( + ResultType=1, + ResultCode="00001", + ResultDesc="Failure", + OriginatorConversationID="12345-67890-1", + ConversationID="AG_20170717_00006c6f7f5b8b6b1a62", + TransactionID="LKXXXX1234", + ResultParameters=[], + ) + callback = TransactionStatusResultCallback(Result=result) + assert callback.is_successful() is False + + +def test_transaction_status_result_callback_is_successful_empty_code(): + """Test is_successful method with an empty ResultCode.""" + result = TransactionStatusResultMetadata( + ResultType=0, + ResultCode="", + ResultDesc="Failure", + OriginatorConversationID="12345-67890-1", + ConversationID="AG_20170717_00006c6f7f5b8b6b1a62", + TransactionID="LKXXXX1234", + ResultParameters=[], + ) + callback = TransactionStatusResultCallback(Result=result) + assert callback.is_successful() is False