Develop

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):

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."""

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):

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):

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."""

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."""

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