diff --git a/docs/docs/swap.mdx b/docs/docs/swap.mdx new file mode 100644 index 0000000..0590ae9 --- /dev/null +++ b/docs/docs/swap.mdx @@ -0,0 +1,218 @@ +--- +title: SIM Swap Detection +description: How mpesakit checks SIM swap history using Safaricom's paid checkATI API, including onboarding and environment setup. +--- + +import SectionHeader from '@site/docs/components/SectionHeader'; +import CodeBlock from '@site/docs/components/CodeBlock'; +import Alert from '@site/docs/components/Alert'; +import ParametersTable from '@site/docs/components/ParametersTable'; + + + + + Unlike standard M-Pesa APIs (like STK Push or C2B), the IMSI checkATI / SIM Swap API is a paid service on the Safaricom Developer Portal (Daraja). +

+ Requirements before integration: + + + + + +To keep credentials secure, configure your consumer key, secret, and target phone numbers in your `.env` file or environment settings: + + +{`# Safaricom Daraja Credentials (Must have Swap/IMSI API product enabled) +MPESA_CONSUMER_KEY=your_consumer_key_here +MPESA_CONSUMER_SECRET=your_consumer_secret_here + +# M-Pesa Environment: "sandbox" or "production" +MPESA_ENVIRONMENT=sandbox + +# Optional: Default target subscriber for testing +TEST_CUSTOMER_NUMBER=0722000000 +`} + + + + + + + + +The `Swap` and `AsyncSwap` classes interface with Safaricom's `/imsi/v2/checkATI` endpoint to check if a subscriber's SIM card has been swapped recently. + +- **Automatic MSISDN Normalization:** Normalizes local formats (e.g. `0722...` or `+254722...`) into standard `254722...` format via `SwapRequest`. +- **Smart Fraud Helper:** Exposes `response.is_recently_swapped`, which automatically checks if `lastSwapDate` differs from the default baseline (`01-01-1900 00:00`). +- **Sync & Async Support:** Ready for synchronous web frameworks (Django/Flask) or asynchronous engines (FastAPI/Tornado). + + + + +{`import os +from dotenv import load_dotenv +from mpesakit.auth import TokenManager +from mpesakit.http_client import MpesaHttpClient +from mpesakit.swap import Swap, SwapRequest + +load_dotenv() + +env = os.getenv("MPESA_ENVIRONMENT", "sandbox") + +# 1. Initialize HTTP client and TokenManager +http_client = MpesaHttpClient(env=env) +token_mgr = TokenManager( + consumer_key=os.getenv("MPESA_CONSUMER_KEY"), + consumer_secret=os.getenv("MPESA_CONSUMER_SECRET"), + http_client=http_client, +) + +# 2. Instantiate Swap service +swap_service = Swap( + http_client=http_client, + token_manager=token_mgr, + environment=env, +) + +# 3. Create request and perform query +request = SwapRequest(customerNumber=os.getenv("TEST_CUSTOMER_NUMBER", "0722000000")) +response = swap_service.swap_request(request) + +if response.is_recently_swapped: + print(f"FRAUD WARNING: SIM swapped on {response.lastSwapDate}") +else: + print("SAFE: No recent SIM swap detected") +`} + + + + + +{`import os +import asyncio +from dotenv import load_dotenv +from mpesakit.auth import AsyncTokenManager +from mpesakit.http_client import MpesaAsyncHttpClient +from mpesakit.swap import AsyncSwap, SwapRequest + +load_dotenv() + +async def check_sim_swap(): + env = os.getenv("MPESA_ENVIRONMENT", "sandbox") + + http_client = MpesaAsyncHttpClient(env=env) + token_mgr = AsyncTokenManager( + consumer_key=os.getenv("MPESA_CONSUMER_KEY"), + consumer_secret=os.getenv("MPESA_CONSUMER_SECRET"), + http_client=http_client, + ) + + swap_service = AsyncSwap( + http_client=http_client, + token_manager=token_mgr, + environment=env, + ) + + request = SwapRequest(customerNumber=os.getenv("TEST_CUSTOMER_NUMBER", "0722000000")) + response = await swap_service.swap_request(request) + + print(f"Request Ref ID: {response.requestRefID}") + print(f"Last Swap Date: {response.lastSwapDate}") + print(f"Is Recently Swapped: {response.is_recently_swapped}") + +asyncio.run(check_sim_swap()) +`} + + + + + + + +
    +
  • Sandbox: https://sandbox.safaricom.co.ke/imsi/v2/checkATI
    • +
  • Production: https://api.safaricom.co.ke/imsi/v2/checkATI
    • +
+ + +## Related Documentation + +- [Auth & Token Management](/auth) - Managing credentials and OAuth tokens +- [Getting Credentials](/getting-credentials) - How to set up Daraja portal apps +- [Production Checklist](/production) - Going live on Safaricom M-Pesa APIs \ No newline at end of file diff --git a/docs/sidebars.ts b/docs/sidebars.ts index 57929ff..7732461 100644 --- a/docs/sidebars.ts +++ b/docs/sidebars.ts @@ -103,6 +103,11 @@ const sidebars: SidebarsConfig = { id: 'b2c-account-top-up', label: 'B2C Account Top Up', }, + { + type: 'doc', + id: 'swap', + label: 'SIM Swap Detection', + }, ], }, { diff --git a/docs/src/components/MpesaKitLanding.tsx b/docs/src/components/MpesaKitLanding.tsx index 70a9bfd..7006435 100644 --- a/docs/src/components/MpesaKitLanding.tsx +++ b/docs/src/components/MpesaKitLanding.tsx @@ -307,6 +307,12 @@ export const MpesaKitLanding: React.FC = () => { status: "maintenance" as "working" | "down" | "maintenance", note: "On hold per Safaricom API support", link: { text: "Ratiba (notes)", href: "https://developer.safaricom.co.ke/APIs/MpesaRatiba" } + }, + { + name: "SIM Swap Detection", + status: "working" as "working" | "down" | "maintenance", + note: "IMSI checkATI query endpoint operating normally (Paid API product)", + link: { text: "SIM Swap Docs", href: "/swap"} } ] }; diff --git a/mpesakit/mpesa_client.py b/mpesakit/mpesa_client.py index 38344b7..ed16adf 100644 --- a/mpesakit/mpesa_client.py +++ b/mpesakit/mpesa_client.py @@ -22,6 +22,8 @@ AsyncRatibaService, ReversalService, AsyncReversalService, + SwapService, + AsyncSwapService, TaxService, AsyncTaxService, TransactionService, @@ -223,6 +225,12 @@ def __init__( ) + # swap => M-PESA Swap services + self.swap = SwapService( + http_client=self.http_client, token_manager=self.token_manager + ) + + class AsyncMpesaClient(MpesaCallbackMixin): """Unified async client for all M-PESA services.""" @@ -298,6 +306,11 @@ def __init__( http_client=self.http_client, token_manager=self.token_manager ) + # swap => M-PESA Swap services + self.swap = AsyncSwapService( + http_client=self.http_client, token_manager=self.token_manager + ) + async def __aenter__(self): await self.http_client.__aenter__() return self diff --git a/mpesakit/services/__init__.py b/mpesakit/services/__init__.py index ea7bdea..7294e68 100644 --- a/mpesakit/services/__init__.py +++ b/mpesakit/services/__init__.py @@ -7,6 +7,7 @@ from .express import StkPushService, AsyncStkPushService from .ratiba import RatibaService, AsyncRatibaService from .reversal import ReversalService, AsyncReversalService +from .swap import SwapService, AsyncSwapService from .tax import TaxService, AsyncTaxService from .transaction import TransactionService, AsyncTransactionService @@ -29,6 +30,8 @@ "AsyncRatibaService", "ReversalService", "AsyncReversalService", + "SwapService", + "AsyncSwapService", "TaxService", "AsyncTaxService", "TransactionService", diff --git a/mpesakit/services/swap.py b/mpesakit/services/swap.py new file mode 100644 index 0000000..452af39 --- /dev/null +++ b/mpesakit/services/swap.py @@ -0,0 +1,63 @@ +"""Facade for M-Pesa Swap APIs.""" + +from mpesakit.auth import AsyncTokenManager, TokenManager +from mpesakit.http_client import AsyncHttpClient, HttpClient +from mpesakit.swap import AsyncSwap, Swap, SwapRequest, SwapResponse + + +class SwapService: + """Facade for all M-Pesa Swap APIs.""" + + def __init__( + self, + http_client: HttpClient, + token_manager: TokenManager, + environment: str = "sandbox", + ) -> None: + """Initialize the Swap service facade.""" + self.http_client = http_client + self.token_manager = token_manager + self.environment = environment + self._swap = Swap( + http_client=self.http_client, + token_manager=self.token_manager, + environment=self.environment, + ) + + def swap_query(self, customer_number: str) -> SwapResponse: + """Initiate a Swap query using a phone number string.""" + request = SwapRequest(customerNumber=customer_number) + return self._swap.swap_request(request) + + def swap_request(self, request: SwapRequest) -> SwapResponse: + """Execute a Swap query using a SwapRequest model.""" + return self._swap.swap_request(request) + + +class AsyncSwapService: + """Async facade for all M-Pesa Swap APIs.""" + + def __init__( + self, + http_client: AsyncHttpClient, + token_manager: AsyncTokenManager, + environment: str = "sandbox", + ) -> None: + """Initialize the Swap service facade.""" + self.http_client = http_client + self.token_manager = token_manager + self.environment = environment + self._swap = AsyncSwap( + http_client=self.http_client, + token_manager=self.token_manager, + environment=self.environment, + ) + + async def swap_query(self, customer_number: str) -> SwapResponse: + """Initiate an async Swap query using a phone number string.""" + request = SwapRequest(customerNumber=customer_number) + return await self._swap.swap_request(request) + + async def swap_request(self, request: SwapRequest) -> SwapResponse: + """Execute an async Swap query using a SwapRequest model.""" + return await self._swap.swap_request(request) diff --git a/mpesakit/swap/__init__.py b/mpesakit/swap/__init__.py new file mode 100644 index 0000000..fd84c44 --- /dev/null +++ b/mpesakit/swap/__init__.py @@ -0,0 +1,9 @@ +from .schemas import SwapRequest, SwapResponse +from .swap import AsyncSwap, Swap + +__all__ = [ + "AsyncSwap", + "Swap", + "SwapRequest", + "SwapResponse", +] diff --git a/mpesakit/swap/schemas.py b/mpesakit/swap/schemas.py new file mode 100644 index 0000000..36cfc36 --- /dev/null +++ b/mpesakit/swap/schemas.py @@ -0,0 +1,51 @@ +"""This module defines schemas for M-Pesa Swap API requests and responses. + +It includes models for swap requests and responses. +""" +from pydantic import BaseModel, ConfigDict, Field, field_validator +from mpesakit.utils.phone import normalize_phone_number + + +class SwapRequest(BaseModel): + """Request schema for Safaricom Daraja SWAP API (/imsi/v2/checkATI).""" + + customerNumber: str = Field( + ..., description="The customer MSISDN in '254XXXXXXXXX' format." + ) + + model_config = ConfigDict( + json_schema_extra={"example": {"customerNumber": "254722000000"}} + ) + + @field_validator("customerNumber") + @classmethod + def validate_customer_number(cls, v: str) -> str: + """Conform the customer Number to required format.""" + normalized = normalize_phone_number(str(v)) + if not normalized: + raise ValueError(f"Invalid Kenyan MSISDN: '{v}'") + return str(normalized) + + +class SwapResponse(BaseModel): + """Response schema returned by Safaricom SWAP API.""" + + requestRefID: str = Field(..., description="Unique transaction ID.") + responseCode: str = Field(..., description="API result status code (e.g. '200').") + responseDesc: str = Field(..., description="Human-readable response message.") + lastSwapDate: str = Field(..., description="SIM last swap timestamp string.") + + @property + def is_successful(self) -> bool: + """Returns True if response code indicates success ('200').""" + return str(self.responseCode).strip() == "200" + + @property + def is_recently_swapped(self) -> bool: + """Returns False if lastSwapDate returns default non-swap date (01-01-1900). + + If the specified SIM was swapped more than 3 months ago, + the API returns a default date of 01-01-1900. + Please refer to:https://developer.safaricom.co.ke/apis/Swap. + """ + return self.is_successful and not self.lastSwapDate.startswith("01-01-1900") diff --git a/mpesakit/swap/swap.py b/mpesakit/swap/swap.py new file mode 100644 index 0000000..6084599 --- /dev/null +++ b/mpesakit/swap/swap.py @@ -0,0 +1,55 @@ +"""SWAP: This API returns the last date a SIM card was swapped. + +Provides functionality to initiate Swap queries using the M-Pesa API. +""" + +from pydantic import BaseModel, ConfigDict + +from mpesakit.auth import AsyncTokenManager, TokenManager +from mpesakit.http_client import AsyncHttpClient, HttpClient + +from .schemas import SwapRequest, SwapResponse + +SWAP_ENDPOINT = "/imsi/v2/checkATI" + +class Swap(BaseModel): + """Represents the Swap API client for SIM card dating.""" + + http_client: HttpClient + token_manager: TokenManager + + model_config = ConfigDict(arbitrary_types_allowed=True) + + def swap_request(self, request: SwapRequest) -> SwapResponse: + """Initiates a Swap request synchronously.""" + headers = { + "Authorization": f"Bearer {self.token_manager.get_token()}", + "Content-Type": "application/json", + } + + response_data = self.http_client.post( + SWAP_ENDPOINT, json=request.model_dump(by_alias=True), headers=headers + ) + return SwapResponse(**response_data) + + +class AsyncSwap(BaseModel): + """Represents the async Swap API client for SIM card dating.""" + + http_client: AsyncHttpClient + token_manager: AsyncTokenManager + + model_config = ConfigDict(arbitrary_types_allowed=True) + + async def swap_request(self, request: SwapRequest) -> SwapResponse: + """Initiates a Swap request asynchronously.""" + token = await self.token_manager.get_token() + headers = { + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + } + + response_data = await self.http_client.post( + SWAP_ENDPOINT, json=request.model_dump(by_alias=True), headers=headers + ) + return SwapResponse(**response_data) diff --git a/stubs/mpesakit/services/__init__.pyi b/stubs/mpesakit/services/__init__.pyi index f92f52b..b430cd1 100644 --- a/stubs/mpesakit/services/__init__.pyi +++ b/stubs/mpesakit/services/__init__.pyi @@ -7,6 +7,7 @@ from .dynamic_qr import DynamicQRCodeService as DynamicQRCodeService, AsyncDynam from .express import StkPushService as StkPushService, AsyncStkPushService as AsyncStkPushService from .ratiba import RatibaService as RatibaService, AsyncRatibaService as AsyncRatibaService from .reversal import ReversalService as ReversalService, AsyncReversalService as AsyncReversalService +from .swap import SwapService as SwapService, AsyncSwapService as AsyncSwapService from .tax import TaxService as TaxService, AsyncTaxService as AsyncTaxService from .transaction import TransactionService as TransactionService, AsyncTransactionService as AsyncTransactionService @@ -29,6 +30,8 @@ __all__ = [ 'AsyncRatibaService', 'ReversalService', 'AsyncReversalService', + 'SwapService', + 'AsyncSwapService', 'TaxService', 'AsyncTaxService', 'TransactionService', diff --git a/stubs/mpesakit/services/swap.pyi b/stubs/mpesakit/services/swap.pyi new file mode 100644 index 0000000..9a92af6 --- /dev/null +++ b/stubs/mpesakit/services/swap.pyi @@ -0,0 +1,20 @@ +from _typeshed import Incomplete +from mpesakit.auth import AsyncTokenManager, TokenManager +from mpesakit.http_client import AsyncHttpClient, HttpClient +from mpesakit.swap import SwapRequest, SwapResponse + +class SwapService: + http_client: Incomplete + token_manager: Incomplete + def __init__(self,http_client:HttpClient,token_manager:TokenManager,environment:str) -> None:... + def swap_query(self,customer_number:str) -> SwapResponse: ... + def swap_request(self,request: SwapRequest) -> SwapResponse: ... + + +class AsyncSwapService: + http_client: Incomplete + token_manager: Incomplete + def __init__(self,http_client:AsyncHttpClient,token_manager:AsyncTokenManager,environment:str) -> None:... + async def swap_query(self,customer_number:str) -> SwapResponse: ... + async def swap_request(self,request: SwapRequest) -> SwapResponse: ... + diff --git a/stubs/mpesakit/swap/__init__.pyi b/stubs/mpesakit/swap/__init__.pyi new file mode 100644 index 0000000..fd84c44 --- /dev/null +++ b/stubs/mpesakit/swap/__init__.pyi @@ -0,0 +1,9 @@ +from .schemas import SwapRequest, SwapResponse +from .swap import AsyncSwap, Swap + +__all__ = [ + "AsyncSwap", + "Swap", + "SwapRequest", + "SwapResponse", +] diff --git a/stubs/mpesakit/swap/schemas.pyi b/stubs/mpesakit/swap/schemas.pyi new file mode 100644 index 0000000..c8e68c1 --- /dev/null +++ b/stubs/mpesakit/swap/schemas.pyi @@ -0,0 +1,19 @@ +from typing import ClassVar +from pydantic import BaseModel, ConfigDict + + +class SwapRequest(BaseModel): + customerNumber: str + model_config = ClassVar[ConfigDict] + + def validate_customer_number(cls, v: str) -> str: ... + + +class SwapResponse(BaseModel): + responseCode: str + requestRefID: str + responseDesc: str + lastSwapDate: str + + def is_successful(self) -> bool: ... + def is_recently_swapped(self) -> bool: ... diff --git a/stubs/mpesakit/swap/swap.pyi b/stubs/mpesakit/swap/swap.pyi new file mode 100644 index 0000000..ed8e576 --- /dev/null +++ b/stubs/mpesakit/swap/swap.pyi @@ -0,0 +1,20 @@ +from typing import Literal +from pydantic import BaseModel, ConfigDict +from mpesakit.auth import AsyncTokenManager, TokenManager +from mpesakit.http_client import AsyncHttpClient, HttpClient +from .schemas import SwapRequest, SwapResponse + + +class Swap(BaseModel): + http_client: HttpClient + token_manager: TokenManager + + model_config = ConfigDict(arbitrary_types_allowed=True) + def swap_request(self, request: SwapRequest) -> SwapResponse: ... + + +class AsyncSwap(BaseModel): + http_client: AsyncHttpClient + token_manager: AsyncTokenManager + model_config = ConfigDict(arbitrary_types_allowed=True) + async def swap_request(self, request: SwapRequest) -> SwapResponse: ... diff --git a/tests/integration/swap/test_swap_e2e.py b/tests/integration/swap/test_swap_e2e.py new file mode 100644 index 0000000..e12f53f --- /dev/null +++ b/tests/integration/swap/test_swap_e2e.py @@ -0,0 +1,138 @@ +"""End-to-End Tests for M-Pesa Swap Request. + +These tests simulate sending live Swap queries to the M-Pesa Daraja API. +They require valid Daraja credentials configured in your environment or .env file. +""" + +import os +import pytest +from dotenv import load_dotenv + +from mpesakit.auth import AsyncTokenManager, TokenManager +from mpesakit.errors import MpesaApiException +from mpesakit.http_client import MpesaAsyncHttpClient, MpesaHttpClient +from mpesakit.swap import AsyncSwap, Swap, SwapRequest, SwapResponse + +load_dotenv() + +pytestmark = pytest.mark.live + + +@pytest.fixture +def env(): + """Return configured MPESA environment (default: sandbox).""" + return os.getenv("MPESA_ENV", "sandbox") + + +@pytest.fixture +def recipient_phone(): + """Return configured test phone number.""" + return os.getenv("MPESA_RECIPIENT_PHONE", "254722000000") + + +@pytest.fixture +def swap_service(env): + """Initialize synchronous M-Pesa Swap service.""" + consumer_key = os.getenv("MPESA_CONSUMER_KEY") + consumer_secret = os.getenv("MPESA_CONSUMER_SECRET") + + if not consumer_key or not consumer_secret: + pytest.skip( + "Missing MPESA_CONSUMER_KEY or MPESA_CONSUMER_SECRET in environment." + ) + + http_client = MpesaHttpClient(env=env) + token_manager = TokenManager( + consumer_key=consumer_key, + consumer_secret=consumer_secret, + http_client=http_client, + ) + + return Swap( + http_client=http_client, + token_manager=token_manager, + environment=env, + ) + + +@pytest.fixture +def async_swap_service(env): + """Initialize asynchronous M-Pesa Swap service.""" + consumer_key = os.getenv("MPESA_CONSUMER_KEY") + consumer_secret = os.getenv("MPESA_CONSUMER_SECRET") + + if not consumer_key or not consumer_secret: + pytest.skip( + "Missing MPESA_CONSUMER_KEY or MPESA_CONSUMER_SECRET in environment." + ) + + http_client = MpesaAsyncHttpClient(env=env) + token_manager = AsyncTokenManager( + consumer_key=consumer_key, + consumer_secret=consumer_secret, + http_client=http_client, + ) + + return AsyncSwap( + http_client=http_client, + token_manager=token_manager, + environment=env, + ) + + +def test_swap_e2e_success(swap_service, recipient_phone): + """Test live synchronous Swap query against M-Pesa Daraja API.""" + print(f"\n [Sync E2E] Initiating Swap query for {recipient_phone}") + + request = SwapRequest(customerNumber=recipient_phone) + print(f" Payload: {request.model_dump()}") + + try: + response = swap_service.swap_request(request) + print(f" Response: {response.model_dump()}") + + assert isinstance(response, SwapResponse) + assert response.requestRefID is not None + assert response.responseCode == "200" + assert response.is_successful is True, f"Swap failed: {response.responseDesc}" + + if response.is_recently_swapped: + print(f" SIM was swapped recently on: {response.lastSwapDate}") + else: + print(" SIM has not been swapped within the last 3 months.") + + except MpesaApiException as e: + pytest.fail(f"API Request failed with status code {e.status_code}: {e.message}") + + +def test_swap_e2e_invalid_phone_format(swap_service): + """Test client-side phone validation before sending E2E network call.""" + print("\n[Sync E2E] Testing client-side validation with invalid phone") + + with pytest.raises(ValueError, match="Invalid Kenyan MSISDN"): + SwapRequest(customerNumber="000000") + + +@pytest.mark.asyncio +async def test_async_swap_e2e_success(async_swap_service, recipient_phone): + """Test live asynchronous Swap query against M-Pesa Daraja API.""" + print(f"\n [Async E2E] Initiating Swap query for {recipient_phone}") + + request = SwapRequest(customerNumber=recipient_phone) + print(f" Async Payload: {request.model_dump()}") + + try: + response = await async_swap_service.swap_request(request) + print(f" Async Response: {response.model_dump()}") + + assert isinstance(response, SwapResponse) + assert response.requestRefID is not None + assert response.responseCode == "200" + assert response.is_successful is True, ( + f"Async Swap failed: {response.responseDesc}" + ) + + except MpesaApiException as e: + pytest.fail( + f"Async API Request failed with status code {e.status_code}: {e.message}" + ) diff --git a/tests/unit/services/test_swap_service.py b/tests/unit/services/test_swap_service.py new file mode 100644 index 0000000..60e4024 --- /dev/null +++ b/tests/unit/services/test_swap_service.py @@ -0,0 +1,220 @@ +"""Unit tests for SwapService and AsyncSwapService Facades.""" + +from unittest.mock import AsyncMock, MagicMock +import pytest + +from mpesakit.auth import AsyncTokenManager, TokenManager +from mpesakit.errors import MpesaApiException +from mpesakit.http_client import AsyncHttpClient, HttpClient +from mpesakit.services.swap import AsyncSwapService, SwapService +from mpesakit.swap import AsyncSwap, Swap, SwapRequest, SwapResponse +from mpesakit.swap.swap import SWAP_ENDPOINT + + +@pytest.fixture +def mock_http_client(): + """Mock synchronous HTTP client with spec to satisfy Pydantic type checks.""" + client = MagicMock(spec=HttpClient) + client.post = MagicMock() + return client + + +@pytest.fixture +def mock_token_manager(): + """Mock synchronous token manager with spec to satisfy Pydantic type checks.""" + manager = MagicMock(spec=TokenManager) + manager.get_token.return_value = "mocked_access_token" + return manager + + +@pytest.fixture +def swap_client(mock_http_client, mock_token_manager): + """Constructs SwapService bypassing Pydantic checks strictly in test setup.""" + service = SwapService.__new__(SwapService) + service.http_client = mock_http_client + service.token_manager = mock_token_manager + service.environment = "sandbox" + service._swap = Swap.model_construct( + http_client=mock_http_client, + token_manager=mock_token_manager, + environment="sandbox", + ) + return service + + +@pytest.fixture +def mock_success_response(): + """Sample successful API response dictionary.""" + return { + "requestRefID": "4277-415525-1", + "responseCode": "200", + "responseDesc": "Success", + "lastSwapDate": "01-01-1900 00:00", + } + + +def make_mock_exception(code: str, message: str) -> MpesaApiException: + """Helper to construct MpesaApiException with mock payload.""" + mock_error = MagicMock() + mock_error.error_code = code + mock_error.error_message = message + mock_error.__str__.return_value = f"[{code}] {message}" + return MpesaApiException(mock_error) + + + +def test_swap_query_success(swap_client, mock_http_client, mock_success_response): + """Test facade swap_query method normalizes string phone and calls HTTP client.""" + mock_http_client.post.return_value = mock_success_response + + response = swap_client.swap_query("0722000000") + + assert isinstance(response, SwapResponse) + assert response.is_successful is True + + mock_http_client.post.assert_called_once_with( + SWAP_ENDPOINT, + json={"customerNumber": "254722000000"}, + headers={ + "Authorization": "Bearer mocked_access_token", + "Content-Type": "application/json", + }, + ) + + +def test_swap_request_direct_model_success( + swap_client, mock_http_client, mock_success_response +): + """Test facade swap_request method accepts SwapRequest model directly.""" + mock_http_client.post.return_value = mock_success_response + request_model = SwapRequest(customerNumber="254711000000") + + response = swap_client.swap_request(request_model) + + assert isinstance(response, SwapResponse) + assert response.is_successful is True + + mock_http_client.post.assert_called_once_with( + SWAP_ENDPOINT, + json={"customerNumber": "254711000000"}, + headers={ + "Authorization": "Bearer mocked_access_token", + "Content-Type": "application/json", + }, + ) + + +def test_swap_query_invalid_phone_raises_validation_error(swap_client): + """Test client-side validation fails before network call on invalid phone format.""" + with pytest.raises(ValueError, match="Invalid Kenyan MSISDN"): + swap_client.swap_query("12345") + + +def test_swap_query_http_error(swap_client, mock_http_client): + """Test error propagation through synchronous facade layer.""" + mock_http_client.post.side_effect = make_mock_exception("HTTP_400", "Bad Request") + + with pytest.raises(MpesaApiException) as exc: + swap_client.swap_query("0722000000") + + assert exc.value.error_code == "HTTP_400" + assert "Bad Request" in exc.value.error.error_message + + +@pytest.fixture +def mock_async_http_client(): + """Mock synchronous HTTP client with spec to satisfy Pydantic type checks.""" + client = MagicMock(spec=AsyncHttpClient) + client.post = AsyncMock() + return client + + +@pytest.fixture +def mock_async_token_manager(): + """Mock asynchronous token manager with spec to satisfy Pydantic type checks.""" + manager = MagicMock(spec=AsyncTokenManager) + manager.get_token = AsyncMock(return_value="mocked_async_access_token") + return manager + + +@pytest.fixture +def async_swap_client(mock_async_http_client, mock_async_token_manager): + """Constructs AsyncSwapService bypassing Pydantic checks strictly in test setup.""" + service = AsyncSwapService.__new__(AsyncSwapService) + service.http_client = mock_async_http_client + service.token_manager = mock_async_token_manager + service.environment = "sandbox" + service._swap = AsyncSwap.model_construct( + http_client=mock_async_http_client, + token_manager=mock_async_token_manager, + environment="sandbox", + ) + return service + + +@pytest.mark.asyncio +async def test_async_swap_query_success( + async_swap_client, mock_async_http_client, mock_success_response +): + """Test async facade swap_query method normalizes string phone and calls async client.""" + mock_async_http_client.post.return_value = mock_success_response + + response = await async_swap_client.swap_query("0722000000") + + assert isinstance(response, SwapResponse) + assert response.is_successful is True + + mock_async_http_client.post.assert_called_once_with( + SWAP_ENDPOINT, + json={"customerNumber": "254722000000"}, + headers={ + "Authorization": "Bearer mocked_async_access_token", + "Content-Type": "application/json", + }, + ) + + +@pytest.mark.asyncio +async def test_async_swap_request_direct_model_success( + async_swap_client, mock_async_http_client, mock_success_response +): + """Test async facade swap_request method accepts SwapRequest model directly.""" + mock_async_http_client.post.return_value = mock_success_response + request_model = SwapRequest(customerNumber="254711000000") + + response = await async_swap_client.swap_request(request_model) + + assert isinstance(response, SwapResponse) + assert response.is_successful is True + + mock_async_http_client.post.assert_called_once_with( + SWAP_ENDPOINT, + json={"customerNumber": "254711000000"}, + headers={ + "Authorization": "Bearer mocked_async_access_token", + "Content-Type": "application/json", + }, + ) + + +@pytest.mark.asyncio +async def test_async_swap_query_invalid_phone_raises_validation_error( + async_swap_client, +): + """Test async client-side validation fails before network call on invalid phone format.""" + with pytest.raises(ValueError, match="Invalid Kenyan MSISDN"): + await async_swap_client.swap_query("invalid_phone") + + +@pytest.mark.asyncio +async def test_async_swap_query_http_error(async_swap_client, mock_async_http_client): + """Test error propagation through asynchronous facade layer.""" + mock_async_http_client.post.side_effect = make_mock_exception( + "HTTP_401", "Unauthorized" + ) + + with pytest.raises(MpesaApiException) as exc: + await async_swap_client.swap_query("0722000000") + + assert exc.value.error_code == "HTTP_401" + assert "Unauthorized" in exc.value.error.error_message diff --git a/tests/unit/swap/test_swap.py b/tests/unit/swap/test_swap.py new file mode 100644 index 0000000..bc96a40 --- /dev/null +++ b/tests/unit/swap/test_swap.py @@ -0,0 +1,213 @@ +"""Unit tests for the Swap class in the mpesakit.swap module.""" + +from unittest.mock import AsyncMock, MagicMock +import pytest + +from mpesakit.auth import AsyncTokenManager, TokenManager +from mpesakit.errors import MpesaApiException +from mpesakit.http_client import AsyncHttpClient, HttpClient +from mpesakit.swap import AsyncSwap, Swap, SwapRequest, SwapResponse +from mpesakit.swap.swap import SWAP_ENDPOINT + + +@pytest.fixture +def mock_http_client(): + """Mock synchronous HTTP client with spec to satisfy Pydantic type checks.""" + client = MagicMock(spec=HttpClient) + client.post = MagicMock() + return client + + +@pytest.fixture +def mock_token_manager(): + """Mock synchronous token manager with spec to satisfy Pydantic type checks.""" + manager = MagicMock(spec=TokenManager) + manager.get_token.return_value = "mocked_access_token" + return manager + + +@pytest.fixture +def swap_client(mock_http_client, mock_token_manager): + """Fixture providing a synchronous Swap client instance.""" + return Swap( + http_client=mock_http_client, + token_manager=mock_token_manager, + ) + + +@pytest.fixture +def valid_swap_request(): + """Return a valid SwapRequest instance.""" + return SwapRequest(customerNumber="254722000000") + + +@pytest.fixture +def mock_success_response(): + """Sample successful API response dictionary (default 1900 date).""" + return { + "requestRefID": "4277-415525-1", + "responseCode": "200", + "responseDesc": "Success", + "lastSwapDate": "01-01-1900 00:00", + } + + +@pytest.fixture +def mock_swapped_response(): + """Sample successful API response dictionary for a recently swapped SIM.""" + return { + "requestRefID": "4277-415525-2", + "responseCode": "200", + "responseDesc": "Success", + "lastSwapDate": "15-08-2026 10:30", + } + + +def make_mock_exception(code: str, message: str) -> MpesaApiException: + """Helper to construct MpesaApiException with an MpesaError mock payload.""" + mock_error = MagicMock() + mock_error.error_code = code + mock_error.error_message = message + mock_error.__str__.return_value = f"[{code}] {message}" + return MpesaApiException(mock_error) + + +def test_swap_request_valid_phone(): + """Test that a valid phone number normalizes properly to 254 format string.""" + req = SwapRequest(customerNumber="0722000000") + assert req.customerNumber == "254722000000" + + +def test_swap_request_invalid_phone(): + """Test that an invalid phone number raises ValueError during validation.""" + with pytest.raises(ValueError, match="Invalid Kenyan MSISDN"): + SwapRequest(customerNumber="12345") + + +def test_swap_response_helper_properties(mock_success_response, mock_swapped_response): + """Test SwapResponse helper properties like is_successful and is_recently_swapped.""" + response = SwapResponse(**mock_success_response) + assert response.is_successful is True + assert response.is_recently_swapped is False + + + swapped_response = SwapResponse(**mock_swapped_response) + assert swapped_response.is_successful is True + assert swapped_response.is_recently_swapped is True + + +def test_swap_response_failed_status(): + """Test that is_recently_swapped evaluates to False on unsuccessful response codes.""" + failed_response = SwapResponse( + requestRefID="4277-415525-3", + responseCode="500", + responseDesc="Internal Server Error", + lastSwapDate="15-08-2026 10:30", + ) + assert failed_response.is_successful is False + assert failed_response.is_recently_swapped is False + + +def test_swap_request_success( + swap_client, + mock_http_client, + mock_token_manager, + valid_swap_request, + mock_success_response, +): + """Test successful synchronous Swap request execution using endpoint path.""" + mock_http_client.post.return_value = mock_success_response + + response = swap_client.swap_request(valid_swap_request) + + assert isinstance(response, SwapResponse) + assert response.responseCode == "200" + assert response.is_successful is True + + mock_http_client.post.assert_called_once_with( + SWAP_ENDPOINT, + json={"customerNumber": "254722000000"}, + headers={ + "Authorization": "Bearer mocked_access_token", + "Content-Type": "application/json", + }, + ) + + +def test_swap_request_http_error(swap_client, mock_http_client, valid_swap_request): + """Test synchronous POST request propagates MpesaApiException on HTTP failure.""" + api_error = make_mock_exception("HTTP_400", "Bad Request") + mock_http_client.post.side_effect = api_error + + with pytest.raises(MpesaApiException) as exc: + swap_client.swap_request(valid_swap_request) + + assert exc.value.error_code == "HTTP_400" + assert "Bad Request" in exc.value.error.error_message + + +@pytest.fixture +def mock_async_token_manager(): + """Mock asynchronous token manager with spec to satisfy Pydantic type checks.""" + manager = MagicMock(spec=AsyncTokenManager) + manager.get_token = AsyncMock(return_value="mocked_async_access_token") + return manager + + +@pytest.fixture +def mock_async_http_client(): + """Mock asynchronous HTTP client with spec to satisfy Pydantic type checks.""" + client = MagicMock(spec=AsyncHttpClient) + client.post = AsyncMock() + return client + + +@pytest.fixture +def async_swap_client(mock_async_http_client, mock_async_token_manager): + """Fixture providing an asynchronous Swap client instance.""" + return AsyncSwap( + http_client=mock_async_http_client, + token_manager=mock_async_token_manager, + ) + + +@pytest.mark.asyncio +async def test_async_swap_request_success( + async_swap_client, + mock_async_http_client, + mock_async_token_manager, + valid_swap_request, + mock_success_response, +): + """Test successful asynchronous Swap request execution using endpoint path.""" + mock_async_http_client.post.return_value = mock_success_response + + response = await async_swap_client.swap_request(valid_swap_request) + + assert isinstance(response, SwapResponse) + assert response.responseCode == "200" + assert response.is_successful is True + + mock_async_http_client.post.assert_called_once_with( + SWAP_ENDPOINT, + json={"customerNumber": "254722000000"}, + headers={ + "Authorization": "Bearer mocked_async_access_token", + "Content-Type": "application/json", + }, + ) + + +@pytest.mark.asyncio +async def test_async_swap_request_http_error( + async_swap_client, mock_async_http_client, valid_swap_request +): + """Test asynchronous POST request propagates MpesaApiException on HTTP failure.""" + api_error = make_mock_exception("HTTP_401", "Unauthorized") + mock_async_http_client.post.side_effect = api_error + + with pytest.raises(MpesaApiException) as exc: + await async_swap_client.swap_request(valid_swap_request) + + assert exc.value.error_code == "HTTP_401" + assert "Unauthorized" in exc.value.error.error_message diff --git a/tests/unit/test_mpesa_client.py b/tests/unit/test_mpesa_client.py index 6bc0c15..8565791 100644 --- a/tests/unit/test_mpesa_client.py +++ b/tests/unit/test_mpesa_client.py @@ -15,6 +15,7 @@ StkPushService, RatibaService, ReversalService, + SwapService, TaxService, TransactionService, ) @@ -114,7 +115,11 @@ def test_ratiba_service_instance(client): assert isinstance(client.ratiba, RatibaService) -# Tests for callback processing methods +def test_swap_service_instance(client): + """Test that the swap service is an instance of SwapService.""" + assert isinstance(client.swap, SwapService) + + class TestCallbackProcessing: """Tests for MpesaClient callback processing methods."""