Chore: align sync/async clients, update stk_push tests and add 100-call STK push stability tests

mpesakit/http_client/mpesa_async_http_client.py

@@ -1,10 +1,98 @@
 """MpesaAsyncHttpClient: An asynchronous client for making HTTP requests to the M-Pesa API."""
 
 from typing import Dict, Any, Optional
+from urllib import response
 import httpx
+import logging
 
 from mpesakit.errors import MpesaError, MpesaApiException
 from .http_client import AsyncHttpClient
+from urllib.parse import urljoin
+
+from tenacity import (
+    AsyncRetrying,
+    RetryCallState,
+    before_sleep_log,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_random_exponential,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def handle_request_error(response: httpx.Response):
+    """Handles non-successful HTTP responses.
+
+    This function is now responsible for converting HTTP status codes
+    and JSON parsing errors into MpesaApiException.
+    """
+    if response.is_success:
+        return
+    try:
+        response_data = response.json()
+    except ValueError:
+        response_data = {"errorMessage": response.text.strip() or ""}
+
+    error_message = response_data.get("errorMessage", "")
+    raise MpesaApiException(
+        MpesaError(
+            error_code=f"HTTP_{response.status_code}",
+            error_message=error_message,
+            status_code=response.status_code,
+            raw_response=response.text,
+        )
+    )
+
+
+def handle_retry_exception(retry_state: RetryCallState):
+    """Custom hook to handle exceptions after all retries fail.
+
+    It raises a custom MpesaApiException with the appropriate error code.
+    """
+    if retry_state.outcome:
+        exception = retry_state.outcome.exception()
+
+        if isinstance(exception, httpx.TimeoutException):
+            raise MpesaApiException(
+                MpesaError(error_code="REQUEST_TIMEOUT", error_message=str(exception))
+            ) from exception
+        elif isinstance(exception, httpx.ConnectError):
+            raise MpesaApiException(
+                MpesaError(error_code="CONNECTION_ERROR", error_message="Failed to connect to M-Pesa API.")
+            ) from exception
+
+        raise MpesaApiException(
+            MpesaError(error_code="REQUEST_FAILED", error_message=str(exception))
+        ) from exception
+
+    raise MpesaApiException(
+        MpesaError(
+            error_code="REQUEST_FAILED",
+            error_message="An unknown retry error occurred.",
+        )
+    )
+
+
+def retry_enabled(enabled: bool):
+    """Factory function to conditionally enable retries.
+
+    Args:
+        enabled (bool): Whether to enable retry logic.
+
+    Returns:
+        A retry condition function.
+    """
+    base_retry = retry_if_exception_type(
+        httpx.TimeoutException
+    ) | retry_if_exception_type(httpx.ConnectError)
+
+    def _retry(retry_state):
+        if not enabled:
+            return False
+        return base_retry(retry_state)
+
+    return _retry
 
 
 class MpesaAsyncHttpClient(AsyncHttpClient):
@@ -20,140 +108,150 @@ class MpesaAsyncHttpClient(AsyncHttpClient):
     base_url: str
     _client: httpx.AsyncClient
 
-    def __init__(self, env: str = "sandbox"):
+    def __init__(self, env: str = "sandbox", retry_enabled_flag: bool = True, wait_strategy=None, stop_strategy=None):
         """Initializes the MpesaAsyncHttpClient with the specified environment."""
         self.base_url = self._resolve_base_url(env)
         self._client = httpx.AsyncClient(base_url=self.base_url)
 
+        self._retry_enabled_flag = retry_enabled_flag
+        self._wait_strategy = wait_strategy or wait_random_exponential(multiplier=5, max=8)
+        self._stop_strategy = stop_strategy or stop_after_attempt(3)
+
     def _resolve_base_url(self, env: str) -> str:
         if env.lower() == "production":
             return "https://api.safaricom.co.ke"
         return "https://sandbox.safaricom.co.ke"
 
+    def _build_retrying(self):
+        return AsyncRetrying(
+            retry=retry_enabled(enabled=self._retry_enabled_flag),
+            wait=self._wait_strategy,
+            stop=self._stop_strategy,
+            retry_error_callback=handle_retry_exception,
+            before_sleep=before_sleep_log(logger, logging.WARNING),
+            reraise=True,
+        )
 
-    async def __aenter__(self):
-        return self
 
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        await self._client.aclose()
 
+    async def async_raw_post(
+        self, url: str, json: Dict[str, Any], headers: Dict[str, str], timeout: int = 10
+    ) -> httpx.Response:
+        """Low-level POST request - may raise httpx exceptions."""
+        full_url = urljoin(self.base_url, url)
 
-    async def post(
-        self, url: str, json: Dict[str, Any], headers: Dict[str, str]
-    ) -> Dict[str, Any]:
-        """Sends an asynchronous POST request to the M-Pesa API."""
-        try:
+        async for attempt in self._build_retrying():
+            with attempt:
+                response = await self._client.post(
+                    full_url, json=json, headers=headers, timeout=timeout
+                )
+                break
+
+        if response is None:
+            raise RuntimeError("Retry loop exited without returning a response or raising an exception.")
+
+        return response
 
-            response = await self._client.post(
-                url, json=json, headers=headers, timeout=10
-            )
 
 
-            try:
-                response_data = response.json()
-            except ValueError:
-                response_data = {"errorMessage": response.text.strip() or ""}
-
-            if not response.is_success:
-                error_message = response_data.get("errorMessage", "")
-                raise MpesaApiException(
-                    MpesaError(
-                        error_code=f"HTTP_{response.status_code}",
-                        error_message=error_message,
-                        status_code=response.status_code,
-                        raw_response=response_data,
-                    )
-                )
+    async def post(
+        self, url: str, json: Dict[str, Any], headers: Dict[str, str]
+    ) -> Dict[str, Any]:
+        """Sends a POST request to the M-Pesa API.
 
-            return response_data
+        Args:
+            url (str): The URL path for the request.
+            json (Dict[str, Any]): The JSON payload for the request body.
+            headers (Dict[str, str]): The HTTP headers for the request.
+            timeout (int): The timeout for the request in seconds.
 
-        except httpx.TimeoutException:
-            raise MpesaApiException(
-                MpesaError(
-                    error_code="REQUEST_TIMEOUT",
-                    error_message="Request to Mpesa timed out.",
-                    status_code=None,
-                )
-            )
-        except httpx.ConnectError:
-            raise MpesaApiException(
-                MpesaError(
-                    error_code="CONNECTION_ERROR",
-                    error_message="Failed to connect to Mpesa API. Check network or URL.",
-                    status_code=None,
-                )
+        Returns:
+            Dict[str, Any]: The JSON response from the API.
+        """
+        response: httpx.Response | None = None
+        try:
+            response = await self.async_raw_post(
+                url, json=json, headers=headers, timeout=10
             )
-        except httpx.HTTPError as e:
+            handle_request_error(response)
+            return response.json()
 
+        except (httpx.RequestError, ValueError) as e:
             raise MpesaApiException(
                 MpesaError(
                     error_code="REQUEST_FAILED",
-                    error_message=f"HTTP request failed: {str(e)}",
-                    status_code=None,
-                    raw_response=None,
+                    error_message="HTTP request failed.",
+                    status_code=getattr(response, "status_code", None),
+                    raw_response=getattr(response, "text", None),
                 )
-            )
+            ) from e
 
-    async def get(
+
+    async def async_raw_get(
         self,
         url: str,
         params: Optional[Dict[str, Any]] = None,
         headers: Optional[Dict[str, str]] = None,
-    ) -> Dict[str, Any]:
-        """Sends an asynchronous GET request to the M-Pesa API."""
-        try:
-            if headers is None:
-                headers = {}
-
-            response = await self._client.get(
-                url, params=params, headers=headers, timeout=10
-            )
+        timeout: int = 10,
+    ) -> httpx.Response:
+        """Low-level GET request - may raise httpx exceptions."""
+        if headers is None:
+            headers = {}
+        full_url = urljoin(self.base_url, url)
 
-            try:
-                response_data = response.json()
-            except ValueError:
-                response_data = {"errorMessage": response.text.strip() or ""}
-
-            if not response.is_success:
-                error_message = response_data.get("errorMessage", "")
-                raise MpesaApiException(
-                    MpesaError(
-                        error_code=f"HTTP_{response.status_code}",
-                        error_message=error_message,
-                        status_code=response.status_code,
-                        raw_response=response_data,
-                    )
+        async for attempt in self._build_retrying():
+            with attempt:
+                response = await self._client.get(
+                    full_url, params=params, headers=headers, timeout=timeout
                 )
+
+        if response is None:
+            raise RuntimeError("Retry loop exited without returning a response or raising an exception.")
+
+        return response
 
-            return response_data
 
 
-        except httpx.TimeoutException:
-            raise MpesaApiException(
-                MpesaError(
-                    error_code="REQUEST_TIMEOUT",
-                    error_message="Request to Mpesa timed out.",
-                    status_code=None,
-                )
-            )
-        except httpx.ConnectError:
-            raise MpesaApiException(
-                MpesaError(
-                    error_code="CONNECTION_ERROR",
-                    error_message="Failed to connect to Mpesa API. Check network or URL.",
-                    status_code=None,
-                )
-            )
-        except httpx.HTTPError as e:
+    async def get(
+        self,
+        url: str,
+        params: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Dict[str, Any]:
+        """Sends a GET request to the M-Pesa API.
+
+        Args:
+            url (str): The URL path for the request.
+            params (Optional[Dict[str, Any]]): The URL parameters.
+            headers (Optional[Dict[str, str]]): The HTTP headers.
+            timeout (int): The timeout for the request in seconds.
+
+        Returns:
+            Dict[str, Any]: The JSON response from the API.
+        """
+        response: httpx.Response | None = None
+        try:
+            response = await self.async_raw_get(url, params, headers, timeout = 10)
+            handle_request_error(response)
+            return response.json()
+        except (httpx.RequestError, ValueError) as e:
             raise MpesaApiException(
                 MpesaError(
                     error_code="REQUEST_FAILED",
-                    error_message=f"HTTP request failed: {str(e)}",
-                    status_code=None,
-                    raw_response=None,
+                    error_message="HTTP request failed.",
+                    status_code=getattr(response, "status_code", None),
+                    raw_response=getattr(response, "text", None),
                 )
-            )
+            ) from e
 
     async def aclose(self):
         """Manually close the underlying httpx client connection pool."""
         await self._client.aclose()
+
+    async def __aenter__(self) -> "MpesaAsyncHttpClient":
+        """Context manager entry point."""
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+        """Context manager exit point. Closes the client."""
+        await self._client.aclose()

mpesakit/mpesa_express/schemas.py

@@ -89,7 +89,7 @@ class StkPushSimulateRequest(BaseModel):
         json_schema_extra={
             "example": {
                 "BusinessShortCode": 654321,
-                "Password": "bXlwYXNzd29yZA==",
+                "Password": "bXlwYXNzd29yZA==",  # nosec B105 
                 "Timestamp": "20240607123045",
                 "TransactionType": "CustomerPayBillOnline",
                 "Amount": 10,
@@ -499,7 +499,7 @@ class StkPushQueryRequest(BaseModel):
         json_schema_extra={
             "example": {
                 "BusinessShortCode": 654321,
-                "Password": "bXlwYXNzd29yZA==",
+                "Password": "bXlwYXNzd29yZA==",  # nosec B105 
                 "Timestamp": "20240607123045",
                 "CheckoutRequestID": "ws_CO_DMZ_123212312_2342347678234",
             }

tests/unit/http_client/test_mpesa_async_http_client.py

@@ -48,7 +48,7 @@ async def test_post_success(async_client):
 
         assert result == {"foo": "bar"}
         mock_post.assert_called_once()
-        mock_post.assert_called_with("/test", json={"a": 1}, headers={"h": "v"}, timeout=10)
+        assert mock_post.call_args[0][0] == "https://sandbox.safaricom.co.ke/test"
 
 
 @pytest.mark.asyncio
@@ -126,7 +126,7 @@ async def test_post_generic_httpx_error(async_client):
             await async_client.post("/error", json={}, headers={})
 
         assert exc.value.error.error_code == "REQUEST_FAILED"
-        assert "protocol error" in exc.value.error.error_message
+        assert "HTTP request failed" in exc.value.error.error_message
 
 
 @pytest.mark.asyncio
@@ -141,7 +141,7 @@ async def test_get_success(async_client):
 
         assert result == {"foo": "bar"}
         mock_get.assert_called_once()
-        mock_get.assert_called_with("/test", params={"a": 1}, headers={"h": "v"}, timeout=10)
+        assert mock_get.call_args[0][0] == "https://sandbox.safaricom.co.ke/test"
 
 
 @pytest.mark.asyncio
@@ -171,7 +171,7 @@ async def test_get_timeout(async_client):
             await async_client.get("/timeout")
 
         assert exc.value.error.error_code == "REQUEST_TIMEOUT"
-        assert "timed out" in exc.value.error.error_message
+        assert "Test Timeout" in exc.value.error.error_message
 
 
 @pytest.mark.asyncio