From fa7f2895b441ee4162430a97da101925acffce8f Mon Sep 17 00:00:00 2001 From: Bramwel Barack Date: Sun, 16 Aug 2026 20:48:12 +0300 Subject: [PATCH 1/5] feat: add swap service and tests --- mpesakit/services/swap.py | 63 ++++++ mpesakit/swap/__init__.py | 9 + mpesakit/swap/schemas.py | 46 +++++ mpesakit/swap/swap.py | 71 +++++++ stubs/mpesakit/swap/__init__.pyi | 9 + stubs/mpesakit/swap/schemas.py | 19 ++ stubs/mpesakit/swap/swap.py | 25 +++ tests/integration/swap/test_swap_e2e.py | 138 +++++++++++++ tests/unit/services/test_swap_service.py | 236 +++++++++++++++++++++++ tests/unit/swap/test_swap.py | 219 +++++++++++++++++++++ 10 files changed, 835 insertions(+) create mode 100644 mpesakit/services/swap.py create mode 100644 mpesakit/swap/__init__.py create mode 100644 mpesakit/swap/schemas.py create mode 100644 mpesakit/swap/swap.py create mode 100644 stubs/mpesakit/swap/__init__.pyi create mode 100644 stubs/mpesakit/swap/schemas.py create mode 100644 stubs/mpesakit/swap/swap.py create mode 100644 tests/integration/swap/test_swap_e2e.py create mode 100644 tests/unit/services/test_swap_service.py create mode 100644 tests/unit/swap/test_swap.py 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..6b851a4 --- /dev/null +++ b/mpesakit/swap/schemas.py @@ -0,0 +1,46 @@ +"""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).""" + 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..6004fb7 --- /dev/null +++ b/mpesakit/swap/swap.py @@ -0,0 +1,71 @@ +"""SWAP: This API returns the last date a SIM card was swapped. + +Provides functionality to initiate Swap queries using the M-Pesa API. +""" + +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): + """Represents the Swap API client for SIM card dating.""" + + http_client: HttpClient + token_manager: TokenManager + environment: Literal["sandbox", "production"] = "sandbox" + + model_config = ConfigDict(arbitrary_types_allowed=True) + + def swap_request(self, request: SwapRequest) -> SwapResponse: + """Initiates a Swap request synchronously.""" + base_domain = ( + "sandbox.safaricom.co.ke" + if self.environment == "sandbox" + else "api.safaricom.co.ke" + ) + url = f"https://{base_domain}/imsi/v2/checkATI" + + headers = { + "Authorization": f"Bearer {self.token_manager.get_token()}", + "Content-Type": "application/json", + } + + response_data = self.http_client.post( + url, 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 + environment: Literal["sandbox", "production"] = "sandbox" + + model_config = ConfigDict(arbitrary_types_allowed=True) + + async def swap_request(self, request: SwapRequest) -> SwapResponse: + """Initiates a Swap request asynchronously.""" + base_domain = ( + "sandbox.safaricom.co.ke" + if self.environment == "sandbox" + else "api.safaricom.co.ke" + ) + url = f"https://{base_domain}/imsi/v2/checkATI" + + token = await self.token_manager.get_token() + headers = { + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + } + + response_data = await self.http_client.post( + url, json=request.model_dump(by_alias=True), headers=headers + ) + return SwapResponse(**response_data) 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.py b/stubs/mpesakit/swap/schemas.py new file mode 100644 index 0000000..c8e68c1 --- /dev/null +++ b/stubs/mpesakit/swap/schemas.py @@ -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.py b/stubs/mpesakit/swap/swap.py new file mode 100644 index 0000000..a031ddf --- /dev/null +++ b/stubs/mpesakit/swap/swap.py @@ -0,0 +1,25 @@ +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 + environment: Literal["sandbox", "production"] = "sandbox" + + model_config = ConfigDict(arbitrary_types_allowed=True) + + def swap_request(self, request: SwapRequest) -> SwapResponse: ... + + +class AsyncSwap(BaseModel): + http_client: AsyncHttpClient + token_manager: AsyncTokenManager + environment: Literal["sandbox", "production"] = "sandbox" + + 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..6fbbfad --- /dev/null +++ b/tests/unit/services/test_swap_service.py @@ -0,0 +1,236 @@ +"""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 + + +@pytest.fixture(params=["sandbox", "production"]) +def env(request): + """Parametrized fixture providing both sandbox and production environments.""" + return request.param + + +@pytest.fixture +def mock_http_client(): + """Mock that passes Pydantic instance check.""" + client = MagicMock(spec=HttpClient) + client.post = MagicMock() + return client + + +@pytest.fixture +def mock_token_manager(): + """Mock that passes Pydantic instance check.""" + 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, env): + """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 = env + service._swap = Swap.model_construct( + http_client=mock_http_client, + token_manager=mock_token_manager, + environment=env, + ) + return service + + +@pytest.fixture +def async_swap_client(mock_async_http_client, mock_async_token_manager, env): + """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 = env + service._swap = AsyncSwap.model_construct( + http_client=mock_async_http_client, + token_manager=mock_async_token_manager, + environment=env, + ) + 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, env): + """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 + + expected_domain = ( + "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" + ) + mock_http_client.post.assert_called_once_with( + f"https://{expected_domain}/imsi/v2/checkATI", + 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, env +): + """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 + + expected_domain = ( + "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" + ) + mock_http_client.post.assert_called_once_with( + f"https://{expected_domain}/imsi/v2/checkATI", + 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 that passes Pydantic instance check.""" + client = MagicMock(spec=AsyncHttpClient) + client.post = AsyncMock() + return client + + +@pytest.fixture +def mock_async_token_manager(): + """Mock that passes Pydantic instance check.""" + manager = MagicMock(spec=AsyncTokenManager) + manager.get_token = AsyncMock(return_value="mocked_async_access_token") + return manager + + +@pytest.mark.asyncio +async def test_async_swap_query_success( + async_swap_client, mock_async_http_client, mock_success_response, env +): + """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 + + expected_domain = ( + "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" + ) + mock_async_http_client.post.assert_called_once_with( + f"https://{expected_domain}/imsi/v2/checkATI", + 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, env +): + """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 + + expected_domain = ( + "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" + ) + mock_async_http_client.post.assert_called_once_with( + f"https://{expected_domain}/imsi/v2/checkATI", + 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..be055cd --- /dev/null +++ b/tests/unit/swap/test_swap.py @@ -0,0 +1,219 @@ +"""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 + + +@pytest.fixture(params=["sandbox", "production"]) +def env(request): + """Parametrized fixture providing both sandbox and production environments.""" + return request.param + + +@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, env): + """Fixture providing a synchronous Swap client instance via direct constructor call.""" + return Swap( + http_client=mock_http_client, + token_manager=mock_token_manager, + environment=env, + ) + + +@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_request_success( + swap_client, + mock_http_client, + mock_token_manager, + valid_swap_request, + mock_success_response, + env, +): + """Test successful synchronous Swap request execution across environments.""" + 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 + + expected_domain = ( + "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" + ) + expected_url = f"https://{expected_domain}/imsi/v2/checkATI" + + mock_http_client.post.assert_called_once_with( + expected_url, + 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, env): + """Fixture providing an asynchronous Swap client instance via direct constructor call.""" + return AsyncSwap( + http_client=mock_async_http_client, + token_manager=mock_async_token_manager, + environment=env, + ) + + +@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, + env, +): + """Test successful asynchronous Swap request execution across environments.""" + 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 + + expected_domain = ( + "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" + ) + expected_url = f"https://{expected_domain}/imsi/v2/checkATI" + + mock_async_http_client.post.assert_called_once_with( + expected_url, + 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 From 651200e9374d2dd8684c16bf78081e248d363880 Mon Sep 17 00:00:00 2001 From: Bramwel Barack Date: Sun, 16 Aug 2026 21:27:24 +0300 Subject: [PATCH 2/5] Add documentation --- docs/docs/swap.mdx | 218 ++++++++++++++++++++++++ docs/sidebars.ts | 5 + docs/src/components/MpesaKitLanding.tsx | 6 + 3 files changed, 229 insertions(+) create mode 100644 docs/docs/swap.mdx 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: +
    +
  • You must request access and complete commercial onboarding with Safaricom to enable the API on your App credentials.
    • +
  • In production, calls to this endpoint consume API credits or billed usage per query according to your Safaricom tariff plan.
    • +
+ + + + +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"} } ] }; From 493bd688a34317ca676cbfd1a25a8a5427b33971 Mon Sep 17 00:00:00 2001 From: Bramwel Barack Date: Sun, 16 Aug 2026 21:39:57 +0300 Subject: [PATCH 3/5] fix ruff error --- mpesakit/swap/schemas.py | 2 +- stubs/mpesakit/swap/{schemas.py => schemas.pyi} | 0 stubs/mpesakit/swap/{swap.py => swap.pyi} | 0 3 files changed, 1 insertion(+), 1 deletion(-) rename stubs/mpesakit/swap/{schemas.py => schemas.pyi} (100%) rename stubs/mpesakit/swap/{swap.py => swap.pyi} (100%) diff --git a/mpesakit/swap/schemas.py b/mpesakit/swap/schemas.py index 6b851a4..7025642 100644 --- a/mpesakit/swap/schemas.py +++ b/mpesakit/swap/schemas.py @@ -20,7 +20,7 @@ class SwapRequest(BaseModel): @field_validator("customerNumber") @classmethod def validate_customer_number(cls, v: str) -> str: - """Conform the customer Number to required format""" + """Conform the customer Number to required format.""" normalized = normalize_phone_number(str(v)) if not normalized: raise ValueError(f"Invalid Kenyan MSISDN: '{v}'") diff --git a/stubs/mpesakit/swap/schemas.py b/stubs/mpesakit/swap/schemas.pyi similarity index 100% rename from stubs/mpesakit/swap/schemas.py rename to stubs/mpesakit/swap/schemas.pyi diff --git a/stubs/mpesakit/swap/swap.py b/stubs/mpesakit/swap/swap.pyi similarity index 100% rename from stubs/mpesakit/swap/swap.py rename to stubs/mpesakit/swap/swap.pyi From db4564d9fdc59ceec9f1dd032f33bd000ccd9c50 Mon Sep 17 00:00:00 2001 From: Bramwel Barack Date: Sat, 22 Aug 2026 20:46:59 +0300 Subject: [PATCH 4/5] simplify endpoint resolution,update test suite and add stubs for service/swap --- mpesakit/swap/schemas.py | 7 ++- mpesakit/swap/swap.py | 19 +----- stubs/mpesakit/services/__init__.pyi | 3 + stubs/mpesakit/services/swap.pyi | 22 +++++++ tests/unit/services/test_swap_service.py | 80 ++++++++++-------------- tests/unit/swap/test_swap.py | 52 +++++++-------- 6 files changed, 89 insertions(+), 94 deletions(-) create mode 100644 stubs/mpesakit/services/swap.pyi diff --git a/mpesakit/swap/schemas.py b/mpesakit/swap/schemas.py index 7025642..36cfc36 100644 --- a/mpesakit/swap/schemas.py +++ b/mpesakit/swap/schemas.py @@ -42,5 +42,10 @@ def is_successful(self) -> bool: @property def is_recently_swapped(self) -> bool: - """Returns False if lastSwapDate returns default non-swap date (01-01-1900).""" + """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 index 6004fb7..be71b9e 100644 --- a/mpesakit/swap/swap.py +++ b/mpesakit/swap/swap.py @@ -11,6 +11,7 @@ from .schemas import SwapRequest, SwapResponse +SWAP_ENDPOINT = "/imsi/v2/checkATI" class Swap(BaseModel): """Represents the Swap API client for SIM card dating.""" @@ -23,20 +24,13 @@ class Swap(BaseModel): def swap_request(self, request: SwapRequest) -> SwapResponse: """Initiates a Swap request synchronously.""" - base_domain = ( - "sandbox.safaricom.co.ke" - if self.environment == "sandbox" - else "api.safaricom.co.ke" - ) - url = f"https://{base_domain}/imsi/v2/checkATI" - headers = { "Authorization": f"Bearer {self.token_manager.get_token()}", "Content-Type": "application/json", } response_data = self.http_client.post( - url, json=request.model_dump(by_alias=True), headers=headers + SWAP_ENDPOINT, json=request.model_dump(by_alias=True), headers=headers ) return SwapResponse(**response_data) @@ -52,13 +46,6 @@ class AsyncSwap(BaseModel): async def swap_request(self, request: SwapRequest) -> SwapResponse: """Initiates a Swap request asynchronously.""" - base_domain = ( - "sandbox.safaricom.co.ke" - if self.environment == "sandbox" - else "api.safaricom.co.ke" - ) - url = f"https://{base_domain}/imsi/v2/checkATI" - token = await self.token_manager.get_token() headers = { "Authorization": f"Bearer {token}", @@ -66,6 +53,6 @@ async def swap_request(self, request: SwapRequest) -> SwapResponse: } response_data = await self.http_client.post( - url, json=request.model_dump(by_alias=True), headers=headers + 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..0eaace8 --- /dev/null +++ b/stubs/mpesakit/services/swap.pyi @@ -0,0 +1,22 @@ +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 + environment: 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 + environment: 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/tests/unit/services/test_swap_service.py b/tests/unit/services/test_swap_service.py index 6fbbfad..60e4024 100644 --- a/tests/unit/services/test_swap_service.py +++ b/tests/unit/services/test_swap_service.py @@ -8,17 +8,12 @@ from mpesakit.http_client import AsyncHttpClient, HttpClient from mpesakit.services.swap import AsyncSwapService, SwapService from mpesakit.swap import AsyncSwap, Swap, SwapRequest, SwapResponse - - -@pytest.fixture(params=["sandbox", "production"]) -def env(request): - """Parametrized fixture providing both sandbox and production environments.""" - return request.param +from mpesakit.swap.swap import SWAP_ENDPOINT @pytest.fixture def mock_http_client(): - """Mock that passes Pydantic instance check.""" + """Mock synchronous HTTP client with spec to satisfy Pydantic type checks.""" client = MagicMock(spec=HttpClient) client.post = MagicMock() return client @@ -26,38 +21,23 @@ def mock_http_client(): @pytest.fixture def mock_token_manager(): - """Mock that passes Pydantic instance check.""" + """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, env): +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 = env + service.environment = "sandbox" service._swap = Swap.model_construct( http_client=mock_http_client, token_manager=mock_token_manager, - environment=env, - ) - return service - - -@pytest.fixture -def async_swap_client(mock_async_http_client, mock_async_token_manager, env): - """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 = env - service._swap = AsyncSwap.model_construct( - http_client=mock_async_http_client, - token_manager=mock_async_token_manager, - environment=env, + environment="sandbox", ) return service @@ -82,7 +62,8 @@ def make_mock_exception(code: str, message: str) -> MpesaApiException: return MpesaApiException(mock_error) -def test_swap_query_success(swap_client, mock_http_client, mock_success_response, env): + +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 @@ -91,11 +72,8 @@ def test_swap_query_success(swap_client, mock_http_client, mock_success_response assert isinstance(response, SwapResponse) assert response.is_successful is True - expected_domain = ( - "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" - ) mock_http_client.post.assert_called_once_with( - f"https://{expected_domain}/imsi/v2/checkATI", + SWAP_ENDPOINT, json={"customerNumber": "254722000000"}, headers={ "Authorization": "Bearer mocked_access_token", @@ -105,7 +83,7 @@ def test_swap_query_success(swap_client, mock_http_client, mock_success_response def test_swap_request_direct_model_success( - swap_client, mock_http_client, mock_success_response, env + 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 @@ -116,11 +94,8 @@ def test_swap_request_direct_model_success( assert isinstance(response, SwapResponse) assert response.is_successful is True - expected_domain = ( - "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" - ) mock_http_client.post.assert_called_once_with( - f"https://{expected_domain}/imsi/v2/checkATI", + SWAP_ENDPOINT, json={"customerNumber": "254711000000"}, headers={ "Authorization": "Bearer mocked_access_token", @@ -148,7 +123,7 @@ def test_swap_query_http_error(swap_client, mock_http_client): @pytest.fixture def mock_async_http_client(): - """Mock that passes Pydantic instance check.""" + """Mock synchronous HTTP client with spec to satisfy Pydantic type checks.""" client = MagicMock(spec=AsyncHttpClient) client.post = AsyncMock() return client @@ -156,15 +131,30 @@ def mock_async_http_client(): @pytest.fixture def mock_async_token_manager(): - """Mock that passes Pydantic instance check.""" + """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, env + 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 @@ -174,11 +164,8 @@ async def test_async_swap_query_success( assert isinstance(response, SwapResponse) assert response.is_successful is True - expected_domain = ( - "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" - ) mock_async_http_client.post.assert_called_once_with( - f"https://{expected_domain}/imsi/v2/checkATI", + SWAP_ENDPOINT, json={"customerNumber": "254722000000"}, headers={ "Authorization": "Bearer mocked_async_access_token", @@ -189,7 +176,7 @@ async def test_async_swap_query_success( @pytest.mark.asyncio async def test_async_swap_request_direct_model_success( - async_swap_client, mock_async_http_client, mock_success_response, env + 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 @@ -200,11 +187,8 @@ async def test_async_swap_request_direct_model_success( assert isinstance(response, SwapResponse) assert response.is_successful is True - expected_domain = ( - "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" - ) mock_async_http_client.post.assert_called_once_with( - f"https://{expected_domain}/imsi/v2/checkATI", + SWAP_ENDPOINT, json={"customerNumber": "254711000000"}, headers={ "Authorization": "Bearer mocked_async_access_token", diff --git a/tests/unit/swap/test_swap.py b/tests/unit/swap/test_swap.py index be055cd..bc96a40 100644 --- a/tests/unit/swap/test_swap.py +++ b/tests/unit/swap/test_swap.py @@ -1,4 +1,4 @@ -"""Unit tests for the Swap class in the Mpesakit.Swap module.""" +"""Unit tests for the Swap class in the mpesakit.swap module.""" from unittest.mock import AsyncMock, MagicMock import pytest @@ -7,12 +7,7 @@ from mpesakit.errors import MpesaApiException from mpesakit.http_client import AsyncHttpClient, HttpClient from mpesakit.swap import AsyncSwap, Swap, SwapRequest, SwapResponse - - -@pytest.fixture(params=["sandbox", "production"]) -def env(request): - """Parametrized fixture providing both sandbox and production environments.""" - return request.param +from mpesakit.swap.swap import SWAP_ENDPOINT @pytest.fixture @@ -32,12 +27,11 @@ def mock_token_manager(): @pytest.fixture -def swap_client(mock_http_client, mock_token_manager, env): - """Fixture providing a synchronous Swap client instance via direct constructor call.""" +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, - environment=env, ) @@ -96,20 +90,32 @@ def test_swap_response_helper_properties(mock_success_response, mock_swapped_res 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, - env, ): - """Test successful synchronous Swap request execution across environments.""" + """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) @@ -118,13 +124,8 @@ def test_swap_request_success( assert response.responseCode == "200" assert response.is_successful is True - expected_domain = ( - "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" - ) - expected_url = f"https://{expected_domain}/imsi/v2/checkATI" - mock_http_client.post.assert_called_once_with( - expected_url, + SWAP_ENDPOINT, json={"customerNumber": "254722000000"}, headers={ "Authorization": "Bearer mocked_access_token", @@ -162,12 +163,11 @@ def mock_async_http_client(): @pytest.fixture -def async_swap_client(mock_async_http_client, mock_async_token_manager, env): - """Fixture providing an asynchronous Swap client instance via direct constructor call.""" +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, - environment=env, ) @@ -178,9 +178,8 @@ async def test_async_swap_request_success( mock_async_token_manager, valid_swap_request, mock_success_response, - env, ): - """Test successful asynchronous Swap request execution across environments.""" + """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) @@ -189,13 +188,8 @@ async def test_async_swap_request_success( assert response.responseCode == "200" assert response.is_successful is True - expected_domain = ( - "sandbox.safaricom.co.ke" if env == "sandbox" else "api.safaricom.co.ke" - ) - expected_url = f"https://{expected_domain}/imsi/v2/checkATI" - mock_async_http_client.post.assert_called_once_with( - expected_url, + SWAP_ENDPOINT, json={"customerNumber": "254722000000"}, headers={ "Authorization": "Bearer mocked_async_access_token", From cd89039112d0bae05470346f4dff3ac0d2e76aab Mon Sep 17 00:00:00 2001 From: Bramwel Barack Date: Tue, 1 Sep 2026 15:35:03 +0300 Subject: [PATCH 5/5] deprecate environment parameter and wire Swap to clients --- mpesakit/mpesa_client.py | 13 +++++++++++++ mpesakit/services/__init__.py | 3 +++ mpesakit/swap/swap.py | 3 --- stubs/mpesakit/services/swap.pyi | 2 -- stubs/mpesakit/swap/swap.pyi | 5 ----- tests/unit/test_mpesa_client.py | 7 ++++++- 6 files changed, 22 insertions(+), 11 deletions(-) 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/swap/swap.py b/mpesakit/swap/swap.py index be71b9e..6084599 100644 --- a/mpesakit/swap/swap.py +++ b/mpesakit/swap/swap.py @@ -3,7 +3,6 @@ Provides functionality to initiate Swap queries using the M-Pesa API. """ -from typing import Literal from pydantic import BaseModel, ConfigDict from mpesakit.auth import AsyncTokenManager, TokenManager @@ -18,7 +17,6 @@ class Swap(BaseModel): http_client: HttpClient token_manager: TokenManager - environment: Literal["sandbox", "production"] = "sandbox" model_config = ConfigDict(arbitrary_types_allowed=True) @@ -40,7 +38,6 @@ class AsyncSwap(BaseModel): http_client: AsyncHttpClient token_manager: AsyncTokenManager - environment: Literal["sandbox", "production"] = "sandbox" model_config = ConfigDict(arbitrary_types_allowed=True) diff --git a/stubs/mpesakit/services/swap.pyi b/stubs/mpesakit/services/swap.pyi index 0eaace8..9a92af6 100644 --- a/stubs/mpesakit/services/swap.pyi +++ b/stubs/mpesakit/services/swap.pyi @@ -6,7 +6,6 @@ from mpesakit.swap import SwapRequest, SwapResponse class SwapService: http_client: Incomplete token_manager: Incomplete - environment: 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: ... @@ -15,7 +14,6 @@ class SwapService: class AsyncSwapService: http_client: Incomplete token_manager: Incomplete - environment: 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/swap.pyi b/stubs/mpesakit/swap/swap.pyi index a031ddf..ed8e576 100644 --- a/stubs/mpesakit/swap/swap.pyi +++ b/stubs/mpesakit/swap/swap.pyi @@ -8,18 +8,13 @@ from .schemas import SwapRequest, SwapResponse class Swap(BaseModel): http_client: HttpClient token_manager: TokenManager - environment: Literal["sandbox", "production"] = "sandbox" model_config = ConfigDict(arbitrary_types_allowed=True) - def swap_request(self, request: SwapRequest) -> SwapResponse: ... class AsyncSwap(BaseModel): http_client: AsyncHttpClient token_manager: AsyncTokenManager - environment: Literal["sandbox", "production"] = "sandbox" - model_config = ConfigDict(arbitrary_types_allowed=True) - async def swap_request(self, request: SwapRequest) -> SwapResponse: ... 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."""