diff --git a/README.md b/README.md index 6717659..b511969 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,24 @@ A python library for integrating with PhonePe APIs. +## v3.0.0 - Breaking changes + +- **Retry mechanism removed.** The SDK no longer retries any HTTP call (including GET) - retrying + is unsafe for non-idempotent calls like pay/refund, since the original request may have already + been processed server-side even if the response was lost. The `should_retry` constructor + parameter has been removed from `StandardCheckoutClient`, `CustomCheckoutClient`, and + `SubscriptionClient` - passing it now raises a `TypeError`. +- **Client construction can now raise.** The SDK fetches its OAuth token immediately at + construction (a single, non-blocking attempt) instead of waiting for the first API call. + Genuine configuration problems (e.g. invalid credentials) now fail fast and + `get_instance(...)`/the constructor raises immediately, where previously construction always + succeeded regardless of credential validity. See the [Quick start](#quick-start) note below for + details - transient failures do NOT raise or block; they're retried automatically in the + background instead. +- **New:** configurable connection pooling/timeouts via `HttpClientConfig` (see + [Connection pool & timeout tuning](#connection-pool--timeout-tuning)) and a `close()` method on + every client to release resources cleanly. + ## Installation Requires `python 3.9` or later @@ -32,6 +50,15 @@ standard_phonepe_client = StandardCheckoutClient.get_instance(client_id=client_i env=env) ``` +> **Note:** Client construction fetches an OAuth token immediately (a single, non-blocking +> attempt) rather than waiting for the first API call. A genuine configuration problem (e.g. +> invalid credentials) fails fast and `get_instance(...)`/the constructor raises immediately; a +> transient failure (network blip, 5xx, rate-limiting) does NOT block construction or raise - it's +> retried automatically in the background instead. Once constructed, the token is kept fresh +> automatically in the background for the lifetime of the client - see +> [Connection pool & timeout tuning](#connection-pool--timeout-tuning) below for `close()` and +> other tunable behavior. + ### Initiate an order using Checkout Page To init a pay request, we make a request object using `StandardCheckoutPayRequest.build_request` [build_request](#standard-checkout-pay-request-builder). @@ -70,6 +97,78 @@ You will get the data [OrderStatusResponse](#order-status-response) object. For more details, please visit: https://developer.phonepe.com +## Connection pool & timeout tuning + +Every client (`StandardCheckoutClient`, `CustomCheckoutClient`, `SubscriptionClient`) accepts an +optional `http_client_config` argument on both its constructor and `get_instance(...)`, letting +you tune the underlying HTTP connection pool and timeouts per merchant/client instance: + +```python +from phonepe.sdk.pg.common.configs.http_client_config import HttpClientConfig +from phonepe.sdk.pg.payments.v2.standard_checkout_client import StandardCheckoutClient +from phonepe.sdk.pg.env import Env + +http_client_config = HttpClientConfig( + pool_size=10, # max pooled (kept-alive) connections per host + keep_alive_seconds=60, # proactively recycle connections idle longer than this + connect_timeout_seconds=3, # max time to establish the TCP/TLS connection + read_timeout_seconds=30, # max time to wait for a response once the request is sent +) + +standard_phonepe_client = StandardCheckoutClient.get_instance( + client_id=client_id, + client_secret=client_secret, + client_version=client_version, + env=env, + http_client_config=http_client_config, +) +``` + +If `http_client_config` is omitted, the SDK uses the defaults shown above (`pool_size=10`, +`keep_alive_seconds=60`, `connect_timeout_seconds=3`, `read_timeout_seconds=30`). + +**Why these four settings trade off against each other:** + +- **`pool_size`** caps how many connections are kept alive per host. A merchant sending many + concurrent requests benefits from a larger pool so requests don't queue up waiting for a free + connection; a merchant sending only the occasional request (e.g. one every several seconds) + gains nothing from a large pool - a small value (2-4) is enough, since most of those connections + would otherwise sit idle. +- **`keep_alive_seconds`** bounds how long a pooled connection can sit idle before the SDK + proactively closes and replaces it with a fresh one, rather than risking handing a request a + connection that a server/load balancer has already silently closed while idle (a scenario + confirmed via repro testing against PhonePe's production environment). This is enforced both + the moment a connection is next reused for a request *and* independently by a background + sweep thread that periodically closes idle connections directly, so staleness is bounded even + during a period with no request traffic at all. +- **`connect_timeout_seconds`** / **`read_timeout_seconds`** bound how long a single request is + allowed to take establishing a connection vs. waiting for a response. A merchant with fast, + reliable infrastructure can tighten these to fail faster on genuine problems; a merchant on + slower/less reliable infrastructure (or calling latency-sensitive endpoints like autoPay APIs) + may need to raise `read_timeout_seconds` to avoid timing out on otherwise-successful, just-slow + responses. + +**Worked examples:** + +- **High-throughput merchant** (e.g. many concurrent payment/status requests per second, on solid + infrastructure): increase `pool_size` (e.g. 20-50) so concurrent requests aren't blocked waiting + for a free connection, and consider lowering `read_timeout_seconds` (e.g. 10-15s) since a slow + response is more likely a genuine problem worth failing fast on. +- **Low-throughput / slower-infrastructure merchant** (e.g. one request every several seconds, + or calling from a network with higher latency): a small `pool_size` (2-4) is plenty - a large + pool would mostly sit idle - but raise `read_timeout_seconds` (e.g. 45-60s) to tolerate your + own slower network/processing before giving up on an otherwise-successful response. + +### Releasing resources with `close()` + +Every client exposes a `close()` method that releases pooled HTTP connections and stops the +background token-refresh thread (see below). This is a daemon thread, so it doesn't prevent your +process from exiting even if you never call `close()` - but short-lived processes (tests, scripts, +serverless invocations) that want a clean, immediate shutdown should call it explicitly: + +```python +standard_phonepe_client.close() +``` ## License diff --git a/phonepe/__init__.py b/phonepe/__init__.py index a157d16..7eb84ea 100644 --- a/phonepe/__init__.py +++ b/phonepe/__init__.py @@ -14,4 +14,4 @@ """Package for integration with PhonePe APIs""" -__version__ = "2.3.0" +__version__ = "3.0.0" diff --git a/phonepe/sdk/pg/common/base_client.py b/phonepe/sdk/pg/common/base_client.py index 74c8af8..b5597e0 100644 --- a/phonepe/sdk/pg/common/base_client.py +++ b/phonepe/sdk/pg/common/base_client.py @@ -18,6 +18,7 @@ import phonepe from phonepe.sdk.pg.common.configs.credential_config import CredentialConfig +from phonepe.sdk.pg.common.configs.http_client_config import HttpClientConfig from phonepe.sdk.pg.common.constants.headers import ( SOURCE, SOURCE_VERSION, @@ -51,7 +52,7 @@ def __init__( client_version: int, env: Env, should_publish_events: bool = True, - should_retry: bool = True, + http_client_config: HttpClientConfig = None, ): self.env = env self.credential_config = CredentialConfig( @@ -59,13 +60,17 @@ def __init__( client_secret=client_secret, client_version=client_version, ) + # Same HttpClientConfig applies to every host this client instance talks to (main pg, + # PCI, event ingestion, oauth) - one merchant traffic profile, consistently tuned. + self.http_client_config = http_client_config or HttpClientConfig() - self._http_command = BaseHttpCommand(get_pg_base_url(self.env)) - self._pci_http_command = BaseHttpCommand(get_pci_pg_base_url(self.env)) + self._http_command = BaseHttpCommand(get_pg_base_url(self.env), http_client_config=self.http_client_config) + self._pci_http_command = BaseHttpCommand(get_pci_pg_base_url(self.env), + http_client_config=self.http_client_config) self.should_publish_events = should_publish_events - self.should_retry = should_retry self._event_publisher_factory = EventPublisherFactory( - event_sender=BaseHttpCommand(host_url=get_event_ingestion_base_url(env)) + event_sender=BaseHttpCommand(host_url=get_event_ingestion_base_url(env), + http_client_config=self.http_client_config) ) self.event_publisher = self._event_publisher_factory.get_event_publisher( should_publish_events=should_publish_events @@ -74,7 +79,7 @@ def __init__( credential_config=self.credential_config, env=self.env, event_publisher=self.event_publisher, - should_retry=should_retry, + http_client_config=self.http_client_config, ) self.event_publisher.start_publishing_events( auth_token_supplier=self._token_service.get_auth_token @@ -91,9 +96,9 @@ def _request_with_token_invalidation( http_command: "BaseHttpCommand" = None, ): # On UnauthorizedAccess the token cache is invalidated so the next call - # fetches a fresh token. This method does NOT retry the request itself. - # If a retry is added in future, use `command` (not `self._http_command`) - # so PCI-scoped calls are not silently downgraded to the standard host. + # fetches a fresh token. This method does NOT retry the request itself: retrying is + # unsafe for non-idempotent calls (e.g. pay, refund) since the original request may + # already have been processed server-side even if the response was lost. command = http_command if http_command is not None else self._http_command try: response_data = command.request( @@ -102,7 +107,6 @@ def _request_with_token_invalidation( headers=merge_dict(self._prepare_headers(), headers), path_params=path_params, data=data, - should_retry=self.should_retry, ) except UnauthorizedAccess as exception: logging.info(f"Failed to authorize") @@ -115,6 +119,16 @@ def _request_with_token_invalidation( return None return response_obj.from_dict(response_data.json()) + def close(self): + """Releases resources held by this client instance: pooled HTTP connections and the + token service's background refresh thread (if running). Safe to call multiple times. + Useful for short-lived processes (tests, scripts, serverless invocations) that want to + shut down cleanly instead of relying on daemon threads/process exit.""" + self._http_command.close() + self._pci_http_command.close() + self._event_publisher_factory.event_sender.close() + self._token_service.close() + def _prepare_headers(self): return { SOURCE: INTEGRATION, diff --git a/phonepe/sdk/pg/common/configs/http_client_config.py b/phonepe/sdk/pg/common/configs/http_client_config.py new file mode 100644 index 0000000..cd621c3 --- /dev/null +++ b/phonepe/sdk/pg/common/configs/http_client_config.py @@ -0,0 +1,70 @@ +# Copyright 2025 PhonePe Private Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class HttpClientConfig: + """Tunable HTTP connection-pool and timeout settings for a PhonePe SDK client instance. + + These four settings travel together and trade off against each other based on a merchant's + traffic profile: + + - A high-throughput merchant (many concurrent requests) typically wants a larger + `pool_size` so requests don't queue up waiting for a free pooled connection, and may + want a smaller `read_timeout_seconds` since their own infrastructure is fast and a slow + response is more likely a genuine problem worth failing fast on. + - A low-throughput merchant (e.g. one request every several seconds) or one running on + slower infrastructure typically needs only a small `pool_size` (2-4 is often enough - + the default of 10 would sit mostly idle) but may want a larger `read_timeout_seconds` to + tolerate their own slower network/processing before giving up on a response. + + See the SDK README's "Connection pool & timeout tuning" section for worked examples. + + Attributes + ---------- + pool_size: int + Maximum number of pooled (kept-alive) connections per host. Default 10. + keep_alive_seconds: float + Maximum time a pooled connection is allowed to sit idle before the SDK proactively + closes and replaces it with a fresh one, rather than risking handing a request a + connection the server/load-balancer may have already silently closed. Enforced two + ways: lazily, the moment an aged-out connection is next checked out for a request, and + proactively, via a background sweep thread (per client instance) that periodically + closes idle connections directly - roughly every keep_alive_seconds / 2 - so a + connection is never left waiting much longer than ~1.5x keep_alive_seconds before being + recycled, even during a long period with no request traffic at all. Default 60 seconds. + connect_timeout_seconds: float + Maximum time to wait while establishing the TCP/TLS connection. Default 3 seconds. + read_timeout_seconds: float + Maximum time to wait for the server to send a response once the request has been sent. + Default 30 seconds (generous enough to accommodate slower endpoints such as autoPay + APIs). + """ + + pool_size: int = 10 + keep_alive_seconds: float = 60 + connect_timeout_seconds: float = 3 + read_timeout_seconds: float = 30 + + def __post_init__(self): + if self.pool_size <= 0: + raise ValueError(f"pool_size must be positive, got {self.pool_size}") + if self.keep_alive_seconds <= 0: + raise ValueError(f"keep_alive_seconds must be positive, got {self.keep_alive_seconds}") + if self.connect_timeout_seconds <= 0: + raise ValueError(f"connect_timeout_seconds must be positive, got {self.connect_timeout_seconds}") + if self.read_timeout_seconds <= 0: + raise ValueError(f"read_timeout_seconds must be positive, got {self.read_timeout_seconds}") diff --git a/phonepe/sdk/pg/common/http_client_modules/base_http_command.py b/phonepe/sdk/pg/common/http_client_modules/base_http_command.py index 6aae09a..22d75b4 100644 --- a/phonepe/sdk/pg/common/http_client_modules/base_http_command.py +++ b/phonepe/sdk/pg/common/http_client_modules/base_http_command.py @@ -13,16 +13,17 @@ # limitations under the License. import logging -from time import sleep from requests import Session +from phonepe.sdk.pg.common.configs.http_client_config import HttpClientConfig from phonepe.sdk.pg.common.exceptions import (BadRequest, ResourceGone, UnauthorizedAccess, ForbiddenAccess, ResourceConflict, ResourceInvalid, ResourceNotFound, ExpectationFailed, TooManyRequests) from phonepe.sdk.pg.common.exceptions import (ClientError, ServerError, PhonePeException) from phonepe.sdk.pg.common.http_client_modules.http_method_type import HttpMethodType +from phonepe.sdk.pg.common.http_client_modules.recycling_http_adapter import RecyclingHTTPAdapter class BaseHttpCommand: @@ -40,85 +41,54 @@ class BaseHttpCommand: 429: TooManyRequests } - TIMEOUT = 5 - - MAX_RETRIES = 3 - BASE_RETRY_DELAY_SECONDS = 1 # exponential backoff: 1s, 2s, 4s, ... before each subsequent retry - - SESSION = Session() - - def __init__(self, host_url: str) -> None: + def __init__(self, host_url: str, http_client_config: HttpClientConfig = None) -> None: self._host_url = host_url + self._http_client_config = http_client_config or HttpClientConfig() + # Each BaseHttpCommand gets its own dedicated Session/connection pool, sized and tuned + # per its HttpClientConfig - NOT a single shared session across every host/merchant in + # the process. This is what makes pool_size/keep_alive/timeouts genuinely configurable + # per client instance (e.g. a low-throughput merchant can use a small pool while a + # high-throughput one uses a large one, without affecting each other). + self._session = Session() + adapter = RecyclingHTTPAdapter( + pool_connections=self._http_client_config.pool_size, + pool_maxsize=self._http_client_config.pool_size, + keep_alive_seconds=self._http_client_config.keep_alive_seconds, + ) + self._session.mount("http://", adapter) + self._session.mount("https://", adapter) @staticmethod def get_complete_url(host_url: str, url: str): return f"{host_url}{url}" - def request(self, url: str, method: HttpMethodType, headers={}, data={}, path_params={}, should_retry: bool = True): - """Makes API Request. + def request(self, url: str, method: HttpMethodType, headers={}, data={}, path_params={}): + """Makes a single-attempt API request (no retries). - On transient failures (connection errors, timeouts, 5xx, 429) the request is retried up to - MAX_RETRIES times with exponential backoff. Genuine client errors (4xx other than 429) are - never retried. Pass should_retry=False to disable this behaviour entirely and fail fast after - a single attempt. + The SDK does not retry requests: a retry is unsafe for non-idempotent calls (e.g. pay, + refund) since the original request may have already been processed server-side even if + the response was lost. Callers that want retry semantics should implement their own + retry/backoff strategy, scoped to the specific calls they know are safe to repeat. """ complete_url = BaseHttpCommand.get_complete_url(self._host_url, url) logging.debug(f"Calling {method}: {complete_url}") - if not should_retry: - return self._send(method, complete_url, headers, data, path_params) - return self._send_with_retries(method, complete_url, headers, data, path_params) + return self._send(method, complete_url, headers, data, path_params) def _send(self, method: HttpMethodType, complete_url: str, headers, data, path_params): + timeout = (self._http_client_config.connect_timeout_seconds, self._http_client_config.read_timeout_seconds) if method == HttpMethodType.GET: return BaseHttpCommand.handle_response( - BaseHttpCommand.SESSION.get(url=complete_url, headers=headers, params=path_params, - timeout=BaseHttpCommand.TIMEOUT)) + self._session.get(url=complete_url, headers=headers, params=path_params, + timeout=timeout)) if method == HttpMethodType.POST: return BaseHttpCommand.handle_response( - BaseHttpCommand.SESSION.post(url=complete_url, headers=headers, data=data, params=path_params, - timeout=BaseHttpCommand.TIMEOUT)) - - def _send_with_retries(self, method: HttpMethodType, complete_url: str, headers, data, path_params): - last_exception = None - for attempt in range(1, BaseHttpCommand.MAX_RETRIES + 1): - try: - return self._send(method, complete_url, headers, data, path_params) - except ClientError as exception: - if not isinstance(exception, TooManyRequests): - # Genuine client-side error (bad request, unauthorized, forbidden, etc.) - # Retrying with the same input will fail identically, so fail fast instead. - logging.error( - f"{method} {complete_url} failed with a non-retryable client error, not retrying | " - f"exception_type={type(exception).__name__} | exception={exception}" - ) - raise - last_exception = exception - logging.warning( - f"{method} {complete_url} attempt {attempt}/{BaseHttpCommand.MAX_RETRIES} failed with a " - f"rate-limit error | exception_type={type(exception).__name__} | exception={exception}" - ) - except Exception as exception: - last_exception = exception - logging.warning( - f"{method} {complete_url} attempt {attempt}/{BaseHttpCommand.MAX_RETRIES} failed | " - f"exception_type={type(exception).__name__} | exception={exception} | " - f"cause={getattr(exception, '__cause__', None)}" - ) + self._session.post(url=complete_url, headers=headers, data=data, params=path_params, + timeout=timeout)) - if attempt < BaseHttpCommand.MAX_RETRIES: - delay = BaseHttpCommand._get_retry_delay(attempt) - logging.info( - f"Waiting {delay}s before retrying {method} {complete_url} " - f"(attempt {attempt + 1}/{BaseHttpCommand.MAX_RETRIES})" - ) - sleep(delay) - - raise last_exception - - @staticmethod - def _get_retry_delay(attempt): - """Exponential backoff delay (in seconds) before the given retry attempt: 1s, 2s, 4s, ...""" - return BaseHttpCommand.BASE_RETRY_DELAY_SECONDS * (2 ** (attempt - 1)) + def close(self): + """Releases all pooled connections held by this command's Session. Safe to call multiple + times.""" + self._session.close() @staticmethod def handle_response(response): diff --git a/phonepe/sdk/pg/common/http_client_modules/recycling_http_adapter.py b/phonepe/sdk/pg/common/http_client_modules/recycling_http_adapter.py new file mode 100644 index 0000000..aeeb827 --- /dev/null +++ b/phonepe/sdk/pg/common/http_client_modules/recycling_http_adapter.py @@ -0,0 +1,222 @@ +# Copyright 2025 PhonePe Private Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import logging +import queue +import threading +import time +from functools import partial + +from requests.adapters import DEFAULT_POOLBLOCK, HTTPAdapter +from urllib3.connectionpool import HTTPConnectionPool, HTTPSConnectionPool + + +class _RecyclingPoolMixin: + """Mixed into urllib3's HTTPConnectionPool/HTTPSConnectionPool to proactively enforce a + keep-alive limit on pooled connections. + + Background: repro testing (see repro_stale_connection.py) showed that PhonePe's + server/load-balancer silently closes idle connections after a few hundred seconds. Since + the SDK no longer retries requests (retrying is unsafe for non-idempotent calls like pay/ + refund), a connection that goes stale while sitting in the pool must never be handed to a + request in the first place - there would be no second attempt to fall back on. + + Two independent mechanisms enforce this: + + 1. `_get_conn` (the single choke point where urllib3 either returns a pooled connection or + mints a new one) is overridden to track how long each connection object has been alive + and proactively `.close()` any reused connection whose age exceeds `keep_alive_seconds` + *before* handing it back. This only fires the next time a request actually checks the + connection out, though - it cannot help a connection that's simply sitting idle with no + request activity at all. + 2. `_evict_idle_connections` (called periodically by RecyclingHTTPAdapter's background + sweep thread - see below) proactively closes idle connections directly in the pool, + independently of whether any request ever checks them out again. This bounds how stale + a connection can get even during a long period with zero traffic. + + Either way, closing sets the connection's `.sock` back to `None`, so urllib3's own + `HTTPConnection.request()` logic (`if self.sock is None: self.connect()`) transparently + establishes a fresh socket on the very next use - reusing urllib3's existing reconnect path + instead of duplicating it. + + Tracking state (`_conn_opened_at`) is a plain instance attribute, scoped to this one pool + instance - never global/class-level - so it cannot affect any other connection pool + elsewhere in the same process. + """ + + def __init__(self, *args, keep_alive_seconds=60, **kwargs): + super().__init__(*args, **kwargs) + self._keep_alive_seconds = keep_alive_seconds + self._conn_opened_at = {} # id(conn) -> time.time() this connection was (re)established + + def _get_conn(self, timeout=None): + conn = super()._get_conn(timeout=timeout) + now = time.time() + is_new = getattr(conn, "sock", None) is None + if is_new: + # Brand new connection object - not yet connected. The actual connect() happens + # synchronously right after this, within the same request call chain, so recording + # the checkout time here is accurate enough (within milliseconds) without needing + # to patch HTTPConnection.connect() itself. + self._conn_opened_at[id(conn)] = now + return conn + + opened_at = self._conn_opened_at.get(id(conn)) + if opened_at is not None and (now - opened_at) > self._keep_alive_seconds: + conn.close() + # About to be transparently reconnected on next use; reset the tracked age so it + # isn't immediately considered stale again. + self._conn_opened_at[id(conn)] = now + return conn + + def _evict_idle_connections(self, keep_alive_seconds): + """Proactively closes any IDLE connection (currently sitting in the pool, not checked + out for an in-flight request) older than keep_alive_seconds. Called periodically by + RecyclingHTTPAdapter's background sweep thread, independently of request activity - + this is what bounds staleness even when nothing has tried to reuse a connection in a + while, which `_get_conn` above cannot do on its own (it only runs on checkout). + + `self.pool` is a fixed-size urllib3 queue.LifoQueue holding one entry per available + pool slot: either an idle connection object ready for reuse, or a None placeholder for + a slot with nothing pooled in it yet. We drain it, close+discard anything too old + (replacing it with a None placeholder to free that slot), and put everything back. + This isn't perfectly atomic against a concurrent request's own _get_conn()/_put_conn() + - at worst, a request racing with this sweep briefly sees an empty pool and mints a new + connection instead of reusing one, which is the same harmless fallback urllib3 already + uses whenever the pool happens to be empty. + + Every drained item is handled in its OWN try/except: closing a connection whose + underlying socket is already broken (e.g. the peer reset it while idle) can itself + raise, and if that exception were allowed to escape mid-loop, every item still waiting + to be re-queued would be silently dropped, permanently shrinking the pool's capacity by + however many items hadn't been processed yet. Isolating each item's handling guarantees + every single drained item - whether closed, kept, or itself broken - is always put + back in some form (the original connection, a fresh None slot, or worst case a bare + None so the slot count is never lost). + """ + pool = self.pool + if pool is None: + return # pool already closed + now = time.time() + items = [] + try: + while True: + items.append(pool.get_nowait()) + except queue.Empty: + pass + + evicted_count = 0 + for item in items: + try: + if item is None: + pool.put(None, block=False) + continue + opened_at = self._conn_opened_at.get(id(item)) + if opened_at is not None and (now - opened_at) > keep_alive_seconds: + item.close() + self._conn_opened_at.pop(id(item), None) + pool.put(None, block=False) + evicted_count += 1 + else: + pool.put(item, block=False) + except Exception: + # Closing (or re-queueing) this one item failed - log it, drop tracking for it + # so it can't be mistaken for a still-valid connection, but still free its slot + # with a None placeholder rather than losing the slot from the pool entirely. + logging.exception( + "Error while proactively evicting a pooled connection; freeing its slot anyway" + ) + self._conn_opened_at.pop(id(item), None) + try: + pool.put(None, block=False) + except Exception: + logging.exception("Could not even free the slot for a connection that failed to evict") + if evicted_count: + logging.info(f"Proactively evicted {evicted_count} idle connection(s) past keep_alive_seconds") + + +class RecyclingHTTPConnectionPool(_RecyclingPoolMixin, HTTPConnectionPool): + pass + + +class RecyclingHTTPSConnectionPool(_RecyclingPoolMixin, HTTPSConnectionPool): + pass + + +class RecyclingHTTPAdapter(HTTPAdapter): + """A requests HTTPAdapter that proactively recycles pooled connections older than + `keep_alive_seconds`, on top of the usual pool_connections/pool_maxsize sizing. + + Enforcement happens two ways: lazily, the next time an aged-out connection is checked out + for a request (see _RecyclingPoolMixin._get_conn), and proactively, via a background sweep + thread (started here) that periodically scans idle pooled connections and closes any that + have exceeded keep_alive_seconds - even if nothing has tried to reuse them. The sweep runs + roughly twice as often as keep_alive_seconds, so an idle connection is never left waiting + much longer than ~1.5x keep_alive_seconds before being caught, regardless of request + traffic. + + The pool-class override below is applied to this adapter's own `PoolManager` instance only + (`pool_classes_by_scheme` is set fresh per-PoolManager in urllib3, never shared class/global + state), so multiple BaseHttpCommand instances - potentially with different HttpClientConfig + settings - can safely coexist in the same merchant process without interfering with each + other or with any other library's use of requests/urllib3 in that process. The background + sweep thread is likewise private to this one adapter instance. + """ + + def __init__(self, *args, keep_alive_seconds=60, **kwargs): + self._keep_alive_seconds = keep_alive_seconds + self._sweep_interval_seconds = max(1.0, keep_alive_seconds / 2) + self._sweep_stop_event = threading.Event() + self._sweep_thread = None + super().__init__(*args, **kwargs) + self._sweep_thread = threading.Thread( + target=self._sweep_loop, name="RecyclingHTTPAdapterSweeper", daemon=True, + ) + self._sweep_thread.start() + + def init_poolmanager(self, connections, maxsize, block=DEFAULT_POOLBLOCK, **pool_kwargs): + super().init_poolmanager(connections, maxsize, block=block, **pool_kwargs) + self.poolmanager.pool_classes_by_scheme = { + "http": partial(RecyclingHTTPConnectionPool, keep_alive_seconds=self._keep_alive_seconds), + "https": partial(RecyclingHTTPSConnectionPool, keep_alive_seconds=self._keep_alive_seconds), + } + + def _sweep_loop(self): + while not self._sweep_stop_event.wait(timeout=self._sweep_interval_seconds): + try: + # RecentlyUsedContainer (urllib3's PoolManager.pools) deliberately raises on + # __iter__/.values() since that's not thread-safe against concurrent pool + # creation - but .keys() IS thread-safe (lock-protected, returns a real list), + # and so is __getitem__, so we look each pool up individually by key instead. + for key in self.poolmanager.pools.keys(): + try: + pool = self.poolmanager.pools[key] + except KeyError: + continue # evicted between .keys() and lookup - already gone, skip it + evict = getattr(pool, "_evict_idle_connections", None) + if evict is not None: + evict(self._keep_alive_seconds) + except Exception: + # Defensive: never let an unexpected error (e.g. a pool closed mid-sweep) kill + # this daemon thread silently. + logging.exception("Unexpected error while proactively sweeping idle connections") + + def close(self): + """Stops the background sweep thread and releases pooled connections. Safe to call + multiple times (e.g. requests.Session.close() calls this once per mounted scheme, and + this same adapter instance is mounted for both http:// and https://).""" + self._sweep_stop_event.set() + if self._sweep_thread is not None and self._sweep_thread.is_alive(): + self._sweep_thread.join(timeout=2) + super().close() diff --git a/phonepe/sdk/pg/common/token_handler/token_service.py b/phonepe/sdk/pg/common/token_handler/token_service.py index a406eff..a9cd932 100644 --- a/phonepe/sdk/pg/common/token_handler/token_service.py +++ b/phonepe/sdk/pg/common/token_handler/token_service.py @@ -13,9 +13,11 @@ # limitations under the License. import logging +import threading from time import time from phonepe.sdk.pg.common.configs.credential_config import CredentialConfig +from phonepe.sdk.pg.common.configs.http_client_config import HttpClientConfig from phonepe.sdk.pg.common.constants.headers import ( ACCEPT, CONTENT_TYPE, @@ -29,6 +31,7 @@ ) from phonepe.sdk.pg.common.events.models.enums.event_type import EventType from phonepe.sdk.pg.common.events.publisher.event_publisher import EventPublisher +from phonepe.sdk.pg.common.exceptions import ClientError, TooManyRequests from phonepe.sdk.pg.common.http_client_modules.base_http_command import BaseHttpCommand from phonepe.sdk.pg.common.http_client_modules.http_method_type import HttpMethodType from phonepe.sdk.pg.common.token_handler.oauth_response import OauthResponse @@ -40,24 +43,76 @@ class TokenService: - """Token management""" + """Token management. + Fetches a token eagerly at construction, then proactively refreshes it in a background + thread at half its lifetime, retrying on failure until the token's hard expiry (and, as a + last-resort safety net, indefinitely with capped backoff beyond that too, since giving up + permanently would otherwise strand the client with no other way to recover). A lock ensures + only one fetch/refresh can mutate the cached token at a time, whether it comes from the + initial construction-time fetch, the proactive background refresh, or a reactive + force_refresh_token() (triggered when a request gets a 401). + + The initial fetch makes exactly one synchronous attempt on the constructing thread - no + sleep-based retry blocks the caller. A genuine client-side error (bad credentials, malformed + request, etc.) fails fast and propagates out of __init__, since retrying identical input + would fail identically. Any other (transient) failure - connection errors, timeouts, 5xx, + 429 - is handed off to the background thread instead: construction still returns immediately + without raising, and the background thread keeps retrying with backoff until it succeeds. + + get_auth_token() also keeps its own synchronous lazy-fetch-with-cached-fallback logic as an + additional safety net beneath the proactive mechanism, for the rare case a real request needs + a token before/despite the background refresh succeeding. + """ + + # Floor on how soon the background loop is allowed to attempt another proactive refresh, + # even if the cached token's half-life computes to "now" or earlier (e.g. a token whose + # issued_at/expires_at are already stale by the time it's cached). Without this floor, such + # a token would cause the loop to busy-refresh with no pacing at all. + MIN_SECONDS_BETWEEN_PROACTIVE_ATTEMPTS = 1.0 + + # Ceiling on how long the background loop will ever sleep in one go. Without this cap, a + # malformed/unexpected token response (e.g. timestamps in the wrong unit, or otherwise huge) + # could compute a sleep duration that overflows the OS-level timer used by threading.Event's + # timed wait (raising OverflowError and crashing the thread) - re-checking at least this + # often also means force_refresh_token()/close() are never blocked for unreasonably long. + MAX_SECONDS_BETWEEN_PROACTIVE_ATTEMPTS = 24 * 60 * 60 # 1 day + + PROACTIVE_REFRESH_BASE_RETRY_DELAY_SECONDS = 1 + PROACTIVE_REFRESH_MAX_RETRY_DELAY_SECONDS = 30 # cap on backoff once past a few attempts def __init__( self, credential_config: CredentialConfig, env: Env, event_publisher: EventPublisher, - should_retry: bool = True, + http_client_config: HttpClientConfig = None, ) -> None: self._credential_config = credential_config - self._http_command = BaseHttpCommand(host_url=get_oauth_base_url(env)) + self._http_command = BaseHttpCommand(host_url=get_oauth_base_url(env), http_client_config=http_client_config) self.event_publisher = event_publisher - self.should_retry = should_retry self.event_publisher.send( build_init_client_event(event_name=EventType.TOKEN_SERVICE_INITIALIZED) ) self.cached_token_data = None + self._token_lock = threading.Lock() + self._stop_event = threading.Event() + self._wake_event = threading.Event() + self._background_thread = None + + # Make exactly one synchronous attempt right now, with NO sleep-based retry on this + # (the constructing) thread. A genuine client-side error fails fast and propagates + # (client construction raises). Any transient failure is logged and left for the + # background thread - started immediately below - to keep retrying with backoff. + self._fetch_initial_token_or_defer_to_background() + + # Start the background thread now regardless of whether the fetch above succeeded: if it + # already has a token, this proactively refreshes it at half-life; if it doesn't yet + # (transient failure above), this immediately takes over retrying instead. + self._background_thread = threading.Thread( + target=self._background_refresh_loop, name="PhonePeTokenRefresher", daemon=True, + ) + self._background_thread.start() def get_current_time(self): return int(time()) @@ -76,16 +131,9 @@ def _is_cached_token_valid(self): def get_auth_token(self): if self._is_cached_token_valid(): - return ( - self.cached_token_data.token_type - + " " - + self.cached_token_data.access_token - ) + return self._format_token(self.cached_token_data) try: - # Retries (with backoff) are only done when there is no cached token to fall - should_retry = self.cached_token_data is None and self.should_retry - token_data = self.fetch_token_from_phonepe(should_retry=should_retry).json() - self.cached_token_data = OauthResponse.from_dict(token_data) + self._fetch_and_store_token() except Exception as exception: if self.cached_token_data is None: self.event_publisher.send( @@ -101,7 +149,8 @@ def get_auth_token(self): f"exception={exception} | " f"cause={getattr(exception, '__cause__', None)} | " f"url={self._http_command._host_url}{OAUTH_ENDPOINT} | " - f"timeout={BaseHttpCommand.TIMEOUT}s" + f"connect_timeout={self._http_command._http_client_config.connect_timeout_seconds}s | " + f"read_timeout={self._http_command._http_client_config.read_timeout_seconds}s" ) raise exception self.event_publisher.send( @@ -119,26 +168,162 @@ def get_auth_token(self): f"exception={exception} | " f"cause={getattr(exception, '__cause__', None)} | " f"url={self._http_command._host_url}{OAUTH_ENDPOINT} | " - f"timeout={BaseHttpCommand.TIMEOUT}s" + f"connect_timeout={self._http_command._http_client_config.connect_timeout_seconds}s | " + f"read_timeout={self._http_command._http_client_config.read_timeout_seconds}s" ) # always return cached token, even if auth-client throws exception - return ( - self.cached_token_data.token_type - + " " - + self.cached_token_data.access_token - ) + return self._format_token(self.cached_token_data) + + @staticmethod + def _format_token(token_data): + return token_data.token_type + " " + token_data.access_token + + def close(self): + """Stops the proactive background refresh thread and releases pooled HTTP connections + held by this token service. Safe to call multiple times.""" + self._stop_event.set() + self._wake_event.set() # wake the loop immediately instead of waiting out a long sleep + if self._background_thread is not None and self._background_thread.is_alive(): + self._background_thread.join(timeout=5) + self._http_command.close() def force_refresh_token(self): logging.info("Force refreshing token") - # Retry on transient errors (e.g. RemoteDisconnected, timeouts, 5xx, 429) so a flaky - # network blip while force-refreshing doesn't leave the cache stale for the next call too. - # The original exception that triggered this refresh (e.g. 401) is re-raised by the caller - # regardless of whether this refresh succeeds. - token_data = self.fetch_token_from_phonepe(should_retry=True).json() - self.cached_token_data = OauthResponse.from_dict(token_data) - - def fetch_token_from_phonepe(self, should_retry: bool = False): + self._fetch_and_store_token() + # Nudge the background loop to recompute its next-refresh target off the token we just + # fetched, instead of possibly sleeping on a schedule based on the now-replaced token. + self._wake_event.set() + + def _fetch_and_store_token(self): + """Fetches a fresh token and atomically stores it. Guarded by _token_lock so the eager + construction-time fetch, the proactive background refresh, the reactive + force_refresh_token() (401 path), and get_auth_token()'s lazy fallback can never race and + corrupt/interleave cached_token_data.""" + with self._token_lock: + token_data = self.fetch_token_from_phonepe().json() + self.cached_token_data = OauthResponse.from_dict(token_data) + + def _fetch_initial_token_or_defer_to_background(self): + try: + self._fetch_and_store_token() + except ClientError as exception: + if not isinstance(exception, TooManyRequests): + # Genuine client-side error (bad credentials, malformed request, etc.) - + # retrying with the same input would fail identically, so fail fast here instead + # of silently retrying forever in the background with no way to surface it. + logging.error( + f"Initial token fetch failed with a non-retryable client error, not retrying | " + f"exception_type={type(exception).__name__} | exception={exception}" + ) + self._publish_none_cached_token_event(exception) + raise + self._log_initial_fetch_deferred(exception) + except Exception as exception: + self._log_initial_fetch_deferred(exception) + + def _log_initial_fetch_deferred(self, exception): + logging.warning( + f"Initial token fetch failed with a transient error - construction is NOT blocked " + f"on retrying it; the background refresh thread will keep retrying instead | " + f"exception_type={type(exception).__name__} | exception={exception}" + ) + + def _publish_none_cached_token_event(self, exception): + self.event_publisher.send( + build_oauth_event_none_cached_token( + fetch_attempt_time=self.get_current_time(), + api_path=OAUTH_ENDPOINT, + exception=exception, + ) + ) + + def _seconds_until_next_refresh(self): + """Seconds to sleep before the next proactive refresh attempt, based on the currently + cached token's half-life - floored at MIN_SECONDS_BETWEEN_PROACTIVE_ATTEMPTS so the loop + can never busy-spin with zero pacing, even if the cached token's timestamps are already + stale (e.g. clock skew, or a token issued already past its own half-life), and capped at + MAX_SECONDS_BETWEEN_PROACTIVE_ATTEMPTS to guard against a malformed/unexpected token + response producing a sleep duration too large for the OS-level timer to handle.""" + if self.cached_token_data is None: + # No token yet (either this is the very first attempt, or the synchronous fetch in + # __init__ failed transiently and deferred here) - try again immediately rather than + # waiting out the usual pacing floor, since there's nothing to lose and a real + # request may be blocked waiting on get_auth_token()'s own fallback in the meantime. + return 0.0 + issued_at = self.cached_token_data.issued_at + expires_at = self.cached_token_data.expires_at + half_life_at = issued_at + (expires_at - issued_at) / 2 + seconds_until = half_life_at - self.get_current_time() + return min( + self.MAX_SECONDS_BETWEEN_PROACTIVE_ATTEMPTS, + max(self.MIN_SECONDS_BETWEEN_PROACTIVE_ATTEMPTS, seconds_until), + ) + + def _background_refresh_loop(self): + while not self._stop_event.is_set(): + try: + sleep_seconds = self._seconds_until_next_refresh() + woke_early = self._wake_event.wait(timeout=sleep_seconds) + self._wake_event.clear() + if self._stop_event.is_set(): + break + if woke_early: + # Something else (e.g. force_refresh_token on a 401) already updated the + # token out of band; just recompute the next sleep target off the fresh data + # rather than also refreshing here. + continue + self._proactive_refresh_with_retry_until_expiry() + except Exception: + # Defensive: never let an unexpected error (e.g. malformed cached_token_data) + # silently kill this daemon thread. Log it, pace with the same floor as other + # attempts, and keep the loop alive. + logging.exception("Unexpected error in proactive token refresh loop") + self._stop_event.wait(timeout=self.MIN_SECONDS_BETWEEN_PROACTIVE_ATTEMPTS) + + def _proactive_refresh_with_retry_until_expiry(self): + attempt = 0 + while not self._stop_event.is_set(): + if self._is_cached_token_valid(): + # Another path already refreshed the token while we were about to retry. + return + attempt += 1 + try: + self._fetch_and_store_token() + logging.info(f"Proactive background token refresh succeeded on attempt {attempt}") + return + except Exception as exception: + cached = self.cached_token_data + if cached is not None: + self.event_publisher.send( + build_oauth_event_used_cached_token_failed( + cached_token_issued_at=cached.issued_at, + cached_token_expires_at=cached.expires_at, + fetch_attempt_time=self.get_current_time(), + api_path=OAUTH_ENDPOINT, + exception=exception, + ) + ) + past_expiry = cached is not None and self.get_current_time() >= cached.expires_at + logging.warning( + f"Proactive background token refresh attempt {attempt} failed " + f"({'past' if past_expiry else 'before'} hard expiry) | " + f"exception_type={type(exception).__name__} | exception={exception}" + ) + delay = min( + self.PROACTIVE_REFRESH_BASE_RETRY_DELAY_SECONDS * (2 ** (attempt - 1)), + self.PROACTIVE_REFRESH_MAX_RETRY_DELAY_SECONDS, + ) + # Keep retrying at least until the cached token's hard expiry, and indefinitely + # (capped backoff) beyond that too - giving up permanently would otherwise strand + # the client with no other way to recover. get_auth_token()'s synchronous lazy + # fallback remains available as an independent safety net for in-flight requests + # in the meantime. self._stop_event.wait() makes this backoff sleep immediately + # interruptible by close(). + if self._stop_event.wait(timeout=delay): + return + + def fetch_token_from_phonepe(self): start = time() try: response = self._http_command.request( @@ -146,7 +331,6 @@ def fetch_token_from_phonepe(self, should_retry: bool = False): url=OAUTH_ENDPOINT, data=self._prepare_oauth_body(), headers=self._prepare_oauth_headers(), - should_retry=should_retry, ) logging.info(f"Token fetch succeeded in {time() - start:.3f}s | status={response.status_code}") return response diff --git a/phonepe/sdk/pg/payments/v2/custom_checkout_client.py b/phonepe/sdk/pg/payments/v2/custom_checkout_client.py index 4f64ba2..0e3bef2 100644 --- a/phonepe/sdk/pg/payments/v2/custom_checkout_client.py +++ b/phonepe/sdk/pg/payments/v2/custom_checkout_client.py @@ -15,6 +15,7 @@ import json from phonepe.sdk.pg.common.base_client import BaseClient +from phonepe.sdk.pg.common.configs.http_client_config import HttpClientConfig from phonepe.sdk.pg.common.events.event_builder import ( build_init_client_event, build_order_status_event, @@ -78,12 +79,11 @@ def __init__( client_secret: str, env: Env, should_publish_events: bool = True, - should_retry: bool = True, + http_client_config: HttpClientConfig = None, ): should_publish_events = should_publish_events and env == Env.PRODUCTION super().__init__( - client_id, client_secret, client_version, env, should_publish_events, - should_retry, + client_id, client_secret, client_version, env, should_publish_events, http_client_config, ) @staticmethod @@ -93,7 +93,7 @@ def get_instance( client_version: int, env: Env, should_publish_events: bool = True, - should_retry: bool = True, + http_client_config: HttpClientConfig = None, ): """ Init CustomCheckoutClient class with merchant-credentials @@ -111,13 +111,11 @@ def get_instance( The default value is `Env.SANDBOX` should_publish_events: bool When true events are sent to PhonePe providing smoother experience - should_retry: bool - When true (default), the SDK retries transient failures (connection errors, timeouts, - server errors, rate-limiting) with exponential backoff. This applies both to the initial - OAuth token fetch (when there is no cached token yet) and to all business API calls - (setup, notify, cancel, order status, refund, etc.). - Set to false to disable this retry behaviour and fail immediately instead, e.g. if the - merchant already has their own retry/backoff strategy in place. + http_client_config: HttpClientConfig + Tunable HTTP connection-pool/timeout settings (pool size, keep-alive, connect + timeout, read timeout). Defaults to HttpClientConfig() SDK defaults if not provided. + See HttpClientConfig's docstring and the README's connection pool tuning section for + guidance on adjusting these per merchant traffic profile. """ should_publish_events = should_publish_events and env == Env.PRODUCTION requested_client_sha = calculate_hash( @@ -126,7 +124,7 @@ def get_instance( str(client_secret), str(env), str(should_publish_events), - str(should_retry), + str(http_client_config), str(FlowType.PG), ) if requested_client_sha in CustomCheckoutClient._cached_instances.keys(): @@ -138,7 +136,7 @@ def get_instance( client_secret=client_secret, env=env, should_publish_events=should_publish_events, - should_retry=should_retry, + http_client_config=http_client_config, ) CustomCheckoutClient._cached_instances[requested_client_sha] = new_instance init_event = build_init_client_event( diff --git a/phonepe/sdk/pg/payments/v2/standard_checkout_client.py b/phonepe/sdk/pg/payments/v2/standard_checkout_client.py index 8ce92d1..84e9ecf 100644 --- a/phonepe/sdk/pg/payments/v2/standard_checkout_client.py +++ b/phonepe/sdk/pg/payments/v2/standard_checkout_client.py @@ -16,6 +16,7 @@ from typing import Dict from phonepe.sdk.pg.common.base_client import BaseClient +from phonepe.sdk.pg.common.configs.http_client_config import HttpClientConfig from phonepe.sdk.pg.common.events.event_builder import build_init_client_event, build_order_status_event, \ build_refund_event, build_standard_checkout_pay_event, build_create_sdk_order_event, build_transaction_status_event, \ build_refund_status_event, build_callback_serialization_failed_event @@ -48,14 +49,13 @@ class StandardCheckoutClient(BaseClient): _cached_instances: Dict[str, BaseClient] = {} def __init__(self, client_id: str, client_version: int, client_secret: str, env: Env, - should_publish_events: bool = True, should_retry: bool = True): + should_publish_events: bool = True, http_client_config: HttpClientConfig = None): should_publish_events = should_publish_events and env == Env.PRODUCTION - super().__init__(client_id, client_secret, client_version, env, should_publish_events, - should_retry) + super().__init__(client_id, client_secret, client_version, env, should_publish_events, http_client_config) @staticmethod def get_instance(client_id: str, client_secret: str, client_version: int, env: Env = Env.SANDBOX, - should_publish_events: bool = True, should_retry: bool = True): + should_publish_events: bool = True, http_client_config: HttpClientConfig = None): """ Init StandardCheckoutClient class with merchant-credentials @@ -72,17 +72,15 @@ def get_instance(client_id: str, client_secret: str, client_version: int, env: E The default value is `Env.SANDBOX` should_publish_events: bool When true events are sent to PhonePe providing smoother experience - should_retry: bool - When true (default), the SDK retries transient failures (connection errors, timeouts, - server errors, rate-limiting) with exponential backoff. This applies both to the initial - OAuth token fetch (when there is no cached token yet) and to all business API calls - (setup, notify, cancel, order status, refund, etc.). - Set to false to disable this retry behaviour and fail immediately instead, e.g. if the - merchant already has their own retry/backoff strategy in place. + http_client_config: HttpClientConfig + Tunable HTTP connection-pool/timeout settings (pool size, keep-alive, connect + timeout, read timeout). Defaults to HttpClientConfig() SDK defaults if not provided. + See HttpClientConfig's docstring and the README's connection pool tuning section for + guidance on adjusting these per merchant traffic profile. """ should_publish_events = should_publish_events and env == Env.PRODUCTION requested_client_sha = calculate_hash(str(client_id), str(client_version), str(client_secret), str(env), - str(should_publish_events), str(should_retry), + str(should_publish_events), str(http_client_config), str(FlowType.PG_CHECKOUT)) if requested_client_sha in StandardCheckoutClient._cached_instances.keys(): return StandardCheckoutClient._cached_instances[requested_client_sha] @@ -92,7 +90,7 @@ def get_instance(client_id: str, client_secret: str, client_version: int, env: E client_secret=client_secret, env=env, should_publish_events=should_publish_events, - should_retry=should_retry) + http_client_config=http_client_config) StandardCheckoutClient._cached_instances[requested_client_sha] = new_instance init_event = build_init_client_event(flow_type=FlowType.PG_CHECKOUT, event_name=EventType.STANDARD_CHECKOUT_CLIENT_INITIALIZED) diff --git a/phonepe/sdk/pg/subscription/v2/subscription_client.py b/phonepe/sdk/pg/subscription/v2/subscription_client.py index 99f7442..52ba6af 100644 --- a/phonepe/sdk/pg/subscription/v2/subscription_client.py +++ b/phonepe/sdk/pg/subscription/v2/subscription_client.py @@ -15,6 +15,7 @@ import json from phonepe.sdk.pg.common.base_client import BaseClient +from phonepe.sdk.pg.common.configs.http_client_config import HttpClientConfig from phonepe.sdk.pg.common.constants.headers import ( SUBSCRIPTION_API_VERSION, SOURCE_VERSION, @@ -87,7 +88,7 @@ def __init__( client_secret: str, env: Env, should_publish_events: bool = True, - should_retry: bool = True, + http_client_config: HttpClientConfig = None, ): """ Initialize the SubscriptionClient class. @@ -104,18 +105,15 @@ def __init__( Environment (SANDBOX or PRODUCTION) should_publish_events: bool Indicates if events should be published to PhonePe - should_retry: bool - When true (default), the SDK retries transient failures (connection errors, timeouts, - server errors, rate-limiting) with exponential backoff. This applies both to the initial - OAuth token fetch (when there is no cached token yet) and to all business API calls - (setup, notify, cancel, order status, refund, etc.). - Set to false to disable this retry behaviour and fail immediately instead, e.g. if the - merchant already has their own retry/backoff strategy in place. + http_client_config: HttpClientConfig + Tunable HTTP connection-pool/timeout settings (pool size, keep-alive, connect + timeout, read timeout). Defaults to HttpClientConfig() SDK defaults if not provided. + See HttpClientConfig's docstring and the README's connection pool tuning section for + guidance on adjusting these per merchant traffic profile. """ should_publish_events = should_publish_events and env == Env.PRODUCTION super().__init__( - client_id, client_secret, client_version, env, should_publish_events, - should_retry, + client_id, client_secret, client_version, env, should_publish_events, http_client_config, ) @staticmethod @@ -125,7 +123,7 @@ def get_instance( client_version: int, env: Env, should_publish_events: bool = True, - should_retry: bool = True, + http_client_config: HttpClientConfig = None, ): """ Get or create an instance of SubscriptionClient class. @@ -142,13 +140,11 @@ def get_instance( Environment (SANDBOX or PRODUCTION) should_publish_events: bool Indicates if events should be published to PhonePe - should_retry: bool - When true (default), the SDK retries transient failures (connection errors, timeouts, - server errors, rate-limiting) with exponential backoff. This applies both to the initial - OAuth token fetch (when there is no cached token yet) and to all business API calls - (setup, notify, cancel, order status, refund, etc.). - Set to false to disable this retry behaviour and fail immediately instead, e.g. if the - merchant already has their own retry/backoff strategy in place. + http_client_config: HttpClientConfig + Tunable HTTP connection-pool/timeout settings (pool size, keep-alive, connect + timeout, read timeout). Defaults to HttpClientConfig() SDK defaults if not provided. + See HttpClientConfig's docstring and the README's connection pool tuning section for + guidance on adjusting these per merchant traffic profile. Returns ---------- @@ -162,7 +158,7 @@ def get_instance( str(client_secret), str(env), str(should_publish_events), - str(should_retry), + str(http_client_config), str(FlowType.SUBSCRIPTION), ) if requested_client_sha in SubscriptionClient._cached_instances.keys(): @@ -174,7 +170,7 @@ def get_instance( client_secret=client_secret, env=env, should_publish_events=should_publish_events, - should_retry=should_retry, + http_client_config=http_client_config, ) SubscriptionClient._cached_instances[requested_client_sha] = new_instance init_event = build_init_client_event( diff --git a/tests/base_custom_checkout_client_for_test.py b/tests/base_custom_checkout_client_for_test.py index b5b752b..f853003 100644 --- a/tests/base_custom_checkout_client_for_test.py +++ b/tests/base_custom_checkout_client_for_test.py @@ -14,14 +14,37 @@ from unittest import TestCase -from phonepe.sdk.pg.env import Env +import responses + +from phonepe.sdk.pg.common.token_handler.token_constants import OAUTH_ENDPOINT +from phonepe.sdk.pg.env import Env, get_oauth_base_url from phonepe.sdk.pg.payments.v2.custom_checkout_client import CustomCheckoutClient +_TOKEN_RESPONSE = { + "access_token": "access_token", + "encrypted_access_token": "encrypted_access_token", + "refresh_token": "refresh_token", + "expires_in": 5014, + "issued_at": 2014804440, + "expires_at": 2014804440, + "session_expires_at": 2014804440, + "token_type": "O-Bearer", +} + class BaseCustomCheckoutClientForTest(TestCase): custom_checkout_client = None + @responses.activate def setUp(self) -> None: + # Client construction now eagerly fetches an OAuth token (see TokenService), so setUp() + # needs its own active responses mock covering that fetch - the test method's own + # @responses.activate (if any) only wraps the method itself, not setUp(), which + # unittest/pytest always calls beforehand, outside that decorator's scope. Once the + # singleton for this client_id/config already exists (from an earlier test), get_instance + # just returns the cached instance with no new HTTP call, so this mock is harmless then. + responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, + json=_TOKEN_RESPONSE) BaseCustomCheckoutClientForTest.custom_checkout_client = CustomCheckoutClient.get_instance( client_id="client_id", client_version=1, @@ -32,12 +55,15 @@ def setUp(self) -> None: @staticmethod def set_client(): if BaseCustomCheckoutClientForTest.custom_checkout_client is None: - BaseCustomCheckoutClientForTest.custom_checkout_client = CustomCheckoutClient.get_instance( - client_id="client_id", - client_version=1, - client_secret="client_secret", - env=Env.SANDBOX, - should_publish_events=False) + with responses.RequestsMock(assert_all_requests_are_fired=False) as mock: + mock.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, + json=_TOKEN_RESPONSE) + BaseCustomCheckoutClientForTest.custom_checkout_client = CustomCheckoutClient.get_instance( + client_id="client_id", + client_version=1, + client_secret="client_secret", + env=Env.SANDBOX, + should_publish_events=False) def del_client(self): BaseCustomCheckoutClientForTest.custom_checkout_client.__del__() diff --git a/tests/base_standard_checkout_client_for_test.py b/tests/base_standard_checkout_client_for_test.py index d42a304..2f69a25 100644 --- a/tests/base_standard_checkout_client_for_test.py +++ b/tests/base_standard_checkout_client_for_test.py @@ -14,14 +14,37 @@ from unittest import TestCase -from phonepe.sdk.pg.env import Env +import responses + +from phonepe.sdk.pg.common.token_handler.token_constants import OAUTH_ENDPOINT +from phonepe.sdk.pg.env import Env, get_oauth_base_url from phonepe.sdk.pg.payments.v2.standard_checkout_client import StandardCheckoutClient +_TOKEN_RESPONSE = { + "access_token": "access_token", + "encrypted_access_token": "encrypted_access_token", + "refresh_token": "refresh_token", + "expires_in": 5014, + "issued_at": 2014804440, + "expires_at": 2014804440, + "session_expires_at": 2014804440, + "token_type": "O-Bearer", +} + class BaseStandardCheckoutClientForTest(TestCase): standard_checkout_client = None + @responses.activate def setUp(self) -> None: + # Client construction now eagerly fetches an OAuth token (see TokenService), so setUp() + # needs its own active responses mock covering that fetch - the test method's own + # @responses.activate (if any) only wraps the method itself, not setUp(), which + # unittest/pytest always calls beforehand, outside that decorator's scope. Once the + # singleton for this client_id/config already exists (from an earlier test), get_instance + # just returns the cached instance with no new HTTP call, so this mock is harmless then. + responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, + json=_TOKEN_RESPONSE) BaseStandardCheckoutClientForTest.standard_checkout_client = StandardCheckoutClient.get_instance( client_id="client_id", client_version=1, @@ -32,11 +55,14 @@ def setUp(self) -> None: @staticmethod def get_standard_checkout_client(): if BaseStandardCheckoutClientForTest.standard_checkout_client is None: - BaseStandardCheckoutClientForTest.standard_checkout_client = StandardCheckoutClient.get_instance( - client_id="client_id", - client_version=1, - client_secret="client_secret", - env=Env.SANDBOX, - should_publish_events=False) + with responses.RequestsMock(assert_all_requests_are_fired=False) as mock: + mock.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, + json=_TOKEN_RESPONSE) + BaseStandardCheckoutClientForTest.standard_checkout_client = StandardCheckoutClient.get_instance( + client_id="client_id", + client_version=1, + client_secret="client_secret", + env=Env.SANDBOX, + should_publish_events=False) return BaseStandardCheckoutClientForTest.standard_checkout_client return BaseStandardCheckoutClientForTest.standard_checkout_client diff --git a/tests/base_subscription_client_for_test.py b/tests/base_subscription_client_for_test.py index 662fd7e..dd15ef7 100644 --- a/tests/base_subscription_client_for_test.py +++ b/tests/base_subscription_client_for_test.py @@ -14,13 +14,35 @@ from unittest import TestCase -from phonepe.sdk.pg.env import Env +import responses + +from phonepe.sdk.pg.common.token_handler.token_constants import OAUTH_ENDPOINT +from phonepe.sdk.pg.env import Env, get_oauth_base_url from phonepe.sdk.pg.subscription.v2.subscription_client import SubscriptionClient +_TOKEN_RESPONSE = { + "access_token": "access_token", + "encrypted_access_token": "encrypted_access_token", + "refresh_token": "refresh_token", + "expires_in": 5014, + "issued_at": 2014804440, + "expires_at": 2014804440, + "session_expires_at": 2014804440, + "token_type": "O-Bearer", +} + class BaseSubscriptionClientForTest(TestCase): - subscription_client = SubscriptionClient.get_instance(client_id="client_id", - client_version=1, - client_secret="client_secret", - env=Env.SANDBOX, - should_publish_events=False) + # Client construction now eagerly fetches an OAuth token (see TokenService), so this needs + # its own active responses mock at the moment of construction. This runs once at module + # import time (same as before), scoped tightly to just this one construction call via an + # explicit RequestsMock context manager - a plain @responses.activate decorator cannot be + # applied to a bare class-body statement. + with responses.RequestsMock(assert_all_requests_are_fired=False) as _mock: + _mock.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, + json=_TOKEN_RESPONSE) + subscription_client = SubscriptionClient.get_instance(client_id="client_id", + client_version=1, + client_secret="client_secret", + env=Env.SANDBOX, + should_publish_events=False) diff --git a/tests/base_test_with_oauth.py b/tests/base_test_with_oauth.py index 6153d7c..126812b 100644 --- a/tests/base_test_with_oauth.py +++ b/tests/base_test_with_oauth.py @@ -30,7 +30,12 @@ class BaseTestWithOauth(TestCase): custom_checkout_client = None subscription_client = None + @responses.activate def setUp(self) -> None: + # Client construction now eagerly fetches an OAuth token (see TokenService), so this + # setUp() needs its own active responses mock covering that fetch - the test method's + # own @responses.activate (if any) only wraps the method itself, not setUp(), which + # unittest/pytest always calls beforehand, outside that decorator's scope. token_response_data = """{ "access_token": "access_token", "encrypted_access_token": "encrypted_access_token", diff --git a/tests/test_base_http_command.py b/tests/test_base_http_command.py index 4ad3bf8..b7b2afe 100644 --- a/tests/test_base_http_command.py +++ b/tests/test_base_http_command.py @@ -12,8 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +import inspect from unittest import TestCase -from unittest.mock import call, patch import responses @@ -33,131 +33,77 @@ class TestBaseHttpCommand(TestCase): - """Covers the generic retry-with-backoff mechanism shared by every HTTP call (GET and POST) - made through BaseHttpCommand, regardless of which client/endpoint uses it (subscriptions, - payments, token fetch, event ingestion, etc.).""" + """BaseHttpCommand makes exactly one attempt per call, for every HTTP verb and every outcome. + The SDK does not retry requests at all: retrying is unsafe for non-idempotent calls (e.g. pay, + refund) since the original request may already have been processed server-side even if the + response was lost. This applies uniformly - transient server errors, rate-limiting, and client + errors are all surfaced to the caller after a single attempt, with no built-in backoff.""" def setUp(self): self.command = BaseHttpCommand(host_url=BASE_URL) - def test_max_retries_constant(self): - # Guards against accidental changes to the configured retry budget - assert BaseHttpCommand.MAX_RETRIES == 3 - - def test_retry_backoff_delay_is_exponential(self): - assert BaseHttpCommand._get_retry_delay(1) == 1 - assert BaseHttpCommand._get_retry_delay(2) == 2 - assert BaseHttpCommand._get_retry_delay(3) == 4 - @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_get_retries_on_transient_server_error_then_succeeds(self, mock_sleep): - responses.add(responses.GET, FULL_URL, status=500) - responses.add(responses.GET, FULL_URL, status=502) + def test_get_success_single_call(self): responses.add(responses.GET, FULL_URL, status=200, json={"ok": True}) response = self.command.request(url=PATH, method=HttpMethodType.GET) assert response.json() == {"ok": True} - assert len(responses.calls) == 3 # 2 failed retries + 1 successful attempt - assert mock_sleep.call_args_list == [call(1), call(2)] + assert len(responses.calls) == 1 @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_post_retries_on_transient_server_error_then_succeeds(self, mock_sleep): - responses.add(responses.POST, FULL_URL, status=503) + def test_post_success_single_call(self): responses.add(responses.POST, FULL_URL, status=200, json={"ok": True}) response = self.command.request(url=PATH, method=HttpMethodType.POST, data={"a": "b"}) assert response.json() == {"ok": True} - assert len(responses.calls) == 2 # 1 failed retry + 1 successful attempt - assert mock_sleep.call_args_list == [call(1)] + assert len(responses.calls) == 1 @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_retry_exhausted_raises_server_error(self, mock_sleep): - for _ in range(BaseHttpCommand.MAX_RETRIES): - responses.add(responses.GET, FULL_URL, status=500) + def test_no_retry_on_server_error(self): + # Even though a subsequent attempt would have succeeded, the SDK must not retry. + responses.add(responses.GET, FULL_URL, status=500) + responses.add(responses.GET, FULL_URL, status=200, json={"ok": True}) self.assertRaises(ServerError, self.command.request, url=PATH, method=HttpMethodType.GET) - assert len(responses.calls) == BaseHttpCommand.MAX_RETRIES # exactly MAX_RETRIES attempts, no more - # no sleep after the final failed attempt since we're about to give up - assert mock_sleep.call_args_list == [call(1), call(2)] + assert len(responses.calls) == 1 # exactly one attempt, no retry @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_retries_on_too_many_requests(self, mock_sleep): + def test_no_retry_on_too_many_requests(self): responses.add(responses.GET, FULL_URL, status=429) responses.add(responses.GET, FULL_URL, status=200, json={"ok": True}) - response = self.command.request(url=PATH, method=HttpMethodType.GET) - - assert response.json() == {"ok": True} - assert len(responses.calls) == 2 # 1 rate-limited attempt + 1 successful retry - assert mock_sleep.call_args_list == [call(1)] - - @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_too_many_requests_exhausted_raises(self, mock_sleep): - for _ in range(BaseHttpCommand.MAX_RETRIES): - responses.add(responses.GET, FULL_URL, status=429) - self.assertRaises(TooManyRequests, self.command.request, url=PATH, method=HttpMethodType.GET) - assert len(responses.calls) == BaseHttpCommand.MAX_RETRIES - assert mock_sleep.call_args_list == [call(1), call(2)] + assert len(responses.calls) == 1 @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_no_retry_on_bad_request(self, mock_sleep): - # e.g. malformed payload - retrying won't fix it + def test_no_retry_on_bad_request(self): responses.add(responses.POST, FULL_URL, status=400, json={"message": "bad"}) self.assertRaises(BadRequest, self.command.request, url=PATH, method=HttpMethodType.POST) - assert len(responses.calls) == 1 # fails fast, no retries for a genuine client error - mock_sleep.assert_not_called() + assert len(responses.calls) == 1 @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_no_retry_on_unauthorized(self, mock_sleep): + def test_no_retry_on_unauthorized(self): responses.add(responses.GET, FULL_URL, status=401, json={"message": "unauthorized"}) self.assertRaises(UnauthorizedAccess, self.command.request, url=PATH, method=HttpMethodType.GET) - assert len(responses.calls) == 1 # fails fast, no retries on 401 - mock_sleep.assert_not_called() + assert len(responses.calls) == 1 @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_no_retry_on_not_found(self, mock_sleep): + def test_no_retry_on_not_found(self): responses.add(responses.GET, FULL_URL, status=404, json={"message": "not found"}) self.assertRaises(ResourceNotFound, self.command.request, url=PATH, method=HttpMethodType.GET) - assert len(responses.calls) == 1 # fails fast, no retries on 404 - mock_sleep.assert_not_called() - - @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_should_retry_false_disables_retries_even_on_server_error(self, mock_sleep): - responses.add(responses.GET, FULL_URL, status=500) - responses.add(responses.GET, FULL_URL, status=200, json={"ok": True}) - - self.assertRaises( - ServerError, self.command.request, url=PATH, method=HttpMethodType.GET, should_retry=False - ) - - assert len(responses.calls) == 1 # single attempt only, no retries - mock_sleep.assert_not_called() - - @responses.activate - def test_should_retry_defaults_to_true(self): - responses.add(responses.GET, FULL_URL, status=200, json={"ok": True}) - - response = self.command.request(url=PATH, method=HttpMethodType.GET) - - assert response.json() == {"ok": True} assert len(responses.calls) == 1 + + def test_request_has_no_should_retry_parameter(self): + # Guards against the retry flag being reintroduced on the request() signature. + params = inspect.signature(BaseHttpCommand.request).parameters + assert "should_retry" not in params diff --git a/tests/test_event_handler.py b/tests/test_event_handler.py index c4e565c..32559b7 100644 --- a/tests/test_event_handler.py +++ b/tests/test_event_handler.py @@ -85,11 +85,6 @@ def testSendsTokenFetchFailureEvent(self): queued_event_handler = QueuedEventPublisher(event_sender=event_sender, queue_handler=queue_handler) - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), - env=Env.PRODUCTION, - event_publisher=queued_event_handler) token_expired_response = """{ "access_token": "access_token", "encrypted_access_token": "encrypted_access_token", @@ -101,6 +96,17 @@ def testSendsTokenFetchFailureEvent(self): "token_type": "O-Bearer" } """ + # Registered before construction: TokenService now eagerly fetches its token at + # construction time (rather than lazily on the first get_auth_token() call). + responses.add(responses.POST, get_oauth_base_url(Env.PRODUCTION) + OAUTH_ENDPOINT, status=200, + json=json.loads(token_expired_response)) + token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", + client_version=1, + client_secret="client_secret"), + env=Env.PRODUCTION, + event_publisher=queued_event_handler) + self.addCleanup(token_service.close) + cur_time = int(time.time_ns()) # Example value for cur_time two_sec_more_cur = int(cur_time + 200) @@ -115,9 +121,12 @@ def testSendsTokenFetchFailureEvent(self): "token_type": "O-Bearer" }} """ - responses.add(responses.POST, get_oauth_base_url(Env.PRODUCTION) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_expired_response)) - token_service.get_auth_token() + # Do NOT call get_auth_token() here: the eager fetch during construction already cached + # the (immediately invalid, expires_in=0) token. This 500 is intentionally left for the + # scheduler's first send_events() tick to hit when it calls auth_token_supplier() to get + # a header for sending the already-queued init event - which publishes the "used cached + # token, refresh failed" event as a side effect, matching this test's expectation of two + # separate ticks (and thus two separate event_response calls). responses.add(responses.POST, get_oauth_base_url(Env.PRODUCTION) + OAUTH_ENDPOINT, status=500) responses.add(responses.POST, get_oauth_base_url(Env.PRODUCTION) + OAUTH_ENDPOINT, status=200, json=json.loads(correct_token_response_data)) @@ -129,7 +138,20 @@ def testSendsTokenFetchFailureEvent(self): queued_event_handler.start_publishing_events(token_service.get_auth_token) sleep(5) - assert event_response.call_count == 2 # first call for tokenInit events, second call for get cached token event + # The proactive background refresh thread and the scheduler's own send_events() tick + # (which needs a fresh auth header via auth_token_supplier() to send the queued init + # event) now race to be the one that discovers/refreshes the invalid token, so the exact + # number of separate event-batch HTTP calls is no longer deterministic (it was tied to + # old lazy-fetch-only timing). What matters is that both the init event and the refresh- + # failure event actually got delivered - check that across all delivered batches instead + # of pinning an exact call count. + assert event_response.call_count >= 1 + delivered_bodies = "".join( + call.request.body.decode() if isinstance(call.request.body, bytes) else str(call.request.body) + for call in responses.calls if call.request.url == event_response.url + ) + assert "TOKEN_SERVICE_INITIALIZED" in delivered_bodies + assert "OAUTH_FETCH_FAILED_USED_CACHED_TOKEN" in delivered_bodies @responses.activate @@ -139,11 +161,6 @@ def testSendsTokenFetchSuccessEvent(self): queued_event_handler = QueuedEventPublisher(event_sender=event_sender, queue_handler=queue_handler) - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=queued_event_handler) - cur_time = int(time.time_ns()) # Example value for cur_time two_sec_more_cur = int(cur_time + 200) @@ -158,8 +175,15 @@ def testSendsTokenFetchSuccessEvent(self): "token_type": "O-Bearer" }} """ + # Registered before construction: TokenService now eagerly fetches its token at + # construction time (rather than lazily on the first get_auth_token() call). responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, json=json.loads(correct_token_response_data)) + token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", + client_version=1, + client_secret="client_secret"), env=Env.SANDBOX, + event_publisher=queued_event_handler) + self.addCleanup(token_service.close) event_response = responses.add(responses.POST, get_event_ingestion_base_url(Env.SANDBOX) + EVENT_BULK_ENDPOINT, status=200, diff --git a/tests/test_order_status.py b/tests/test_order_status.py index 64bb32c..efa6ce5 100644 --- a/tests/test_order_status.py +++ b/tests/test_order_status.py @@ -691,7 +691,12 @@ def test_check_status_ppe_intent_custom(self): ], ) - assert len(responses.calls) == 2 + # Was 2 (1 oauth + 1 order-status GET) under the old lazy-fetch model, when this test + # happened to be the first one in the whole suite to use the shared custom_checkout_client + # singleton and thus triggered its lazy token fetch. TokenService now fetches its token + # eagerly at construction time instead, so by the time any test runs, the shared + # singleton's token is already cached - only the order-status GET remains here. + assert len(responses.calls) == 1 assert response_object == expected_order_status_obj @responses.activate diff --git a/tests/test_recycling_http_adapter.py b/tests/test_recycling_http_adapter.py new file mode 100644 index 0000000..0766e6c --- /dev/null +++ b/tests/test_recycling_http_adapter.py @@ -0,0 +1,195 @@ +# Copyright 2025 PhonePe Private Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import http.server +import socketserver +import threading +import time +from unittest import TestCase + +import requests + +from phonepe.sdk.pg.common.http_client_modules.recycling_http_adapter import RecyclingHTTPAdapter + + +class _Handler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" # keep-alive support, required to actually pool connections + + def do_GET(self): + body = b"ok" + self.send_response(200) + self.send_header("Content-Type", "text/plain") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass # silence per-request logging + + +class _ThreadingHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): + daemon_threads = True + + +class TestRecyclingHTTPAdapter(TestCase): + """RecyclingHTTPAdapter enforces keep_alive_seconds two ways: lazily (a connection is + checked for staleness the next time a request actually reuses it) and proactively (a + background sweep thread periodically closes idle connections directly, regardless of + request activity). These tests use a real local HTTP/1.1 server since the behavior being + tested is genuine socket-level connection reuse/closure, not mockable at the `requests` + layer.""" + + def setUp(self): + self.server = _ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + self.port = self.server.server_address[1] + self.server_thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.server_thread.start() + self.addCleanup(self.server.shutdown) + + self.session = requests.Session() + self.addCleanup(self.session.close) + + @property + def url(self): + return f"http://127.0.0.1:{self.port}/" + + def _mount_adapter(self, keep_alive_seconds): + adapter = RecyclingHTTPAdapter(pool_connections=5, pool_maxsize=5, + keep_alive_seconds=keep_alive_seconds) + self.session.mount("http://", adapter) + self.addCleanup(adapter.close) + return adapter + + def test_connection_reused_within_keep_alive_window(self): + adapter = self._mount_adapter(keep_alive_seconds=10) + r1 = self.session.get(self.url) + pool = adapter.poolmanager.connection_from_url(self.url) + opened_at_1 = dict(pool._conn_opened_at) + + r2 = self.session.get(self.url) + opened_at_2 = dict(pool._conn_opened_at) + + assert r1.status_code == r2.status_code == 200 + assert opened_at_1 == opened_at_2, "same connection object/timestamp should be reused" + + def test_lazy_recycle_on_checkout_after_keep_alive_expires(self): + adapter = self._mount_adapter(keep_alive_seconds=1) + # Isolate the lazy (checkout-time) recycling mechanism from the background sweep, which + # would otherwise race to evict the same idle connection independently during the sleep + # below - this test is specifically about the _get_conn-time check, covered separately + # (and in combination) by the other tests in this file. + adapter._sweep_stop_event.set() + + r1 = self.session.get(self.url) + pool = adapter.poolmanager.connection_from_url(self.url) + opened_at_1 = dict(pool._conn_opened_at) + + time.sleep(1.5) # exceed the 1s keep-alive + r2 = self.session.get(self.url) + opened_at_2 = dict(pool._conn_opened_at) + + assert r1.status_code == r2.status_code == 200 + assert list(opened_at_2.values())[0] > list(opened_at_1.values())[0], ( + "connection should have been recycled (fresh timestamp) once checked out past keep-alive" + ) + + def test_background_sweep_evicts_idle_connection_with_zero_request_activity(self): + # keep_alive_seconds=1 -> sweep_interval = max(1.0, 0.5) = 1.0s + adapter = self._mount_adapter(keep_alive_seconds=1) + self.session.get(self.url) + pool = adapter.poolmanager.connection_from_url(self.url) + assert not pool.pool.empty() + + # No further requests at all - only the background sweep thread can evict this. + time.sleep(2.0) + + idle_item = pool.pool.get_nowait() + pool.pool.put(idle_item) # put back immediately so the pool is left usable + assert idle_item is None, ( + "background sweep should have proactively evicted the idle connection " + "(freeing the slot to None) with zero request activity" + ) + + # A subsequent request should still succeed via a fresh, transparent reconnect. + r = self.session.get(self.url) + assert r.status_code == 200 + + def test_close_stops_sweep_thread_and_is_idempotent(self): + adapter = self._mount_adapter(keep_alive_seconds=5) + self.session.get(self.url) + assert adapter._sweep_thread.is_alive() + + adapter.close() + time.sleep(0.2) + assert not adapter._sweep_thread.is_alive() + + adapter.close() # calling again (e.g. via Session.close() mounting twice) must not raise + + def test_sweep_interval_is_half_keep_alive_with_a_floor(self): + for keep_alive_seconds, expected_interval in [(10, 5.0), (1, 1.0), (0.2, 1.0)]: + adapter = RecyclingHTTPAdapter(keep_alive_seconds=keep_alive_seconds) + try: + assert adapter._sweep_interval_seconds == expected_interval + finally: + adapter.close() + + def test_sweep_survives_a_connection_that_fails_to_close(self): + # Regression test: a connection whose underlying socket is already broken can itself + # raise when .close() is called on it. If that exception were allowed to abort the + # sweep mid-loop, every not-yet-processed item drained from the pool queue in that + # pass would be silently lost, permanently shrinking the pool below its configured + # maxsize. Every item must be handled independently so one bad connection can't take + # the rest of the pool's capacity down with it. + from phonepe.sdk.pg.common.http_client_modules.recycling_http_adapter import ( + RecyclingHTTPConnectionPool, + ) + + class _RaisingCloseConn: + def close(self): + raise OSError("simulated broken socket on close()") + + class _NormalConn: + def __init__(self): + self.closed = False + + def close(self): + self.closed = True + + pool = RecyclingHTTPConnectionPool("example.com", 443, maxsize=3) + self.addCleanup(pool.close) + for _ in range(3): + pool.pool.get_nowait() # drain the auto-filled None placeholders + + now = time.time() + conn_a, conn_b, conn_c = _NormalConn(), _RaisingCloseConn(), _NormalConn() + for conn in (conn_a, conn_b, conn_c): + pool._conn_opened_at[id(conn)] = now - 100 # well past any reasonable keep-alive + pool.pool.put(conn) + + pool._evict_idle_connections(keep_alive_seconds=60) + + items = [] + import queue + try: + while True: + items.append(pool.pool.get_nowait()) + except queue.Empty: + pass + for item in items: + pool.pool.put(item) + + assert len(items) == 3, f"pool capacity was lost: expected 3 slots, got {len(items)}" + assert conn_a.closed + assert all(item is None for item in items), "all three should have been evicted to None" + diff --git a/tests/test_should_retry_business_calls.py b/tests/test_should_retry_business_calls.py deleted file mode 100644 index 5954ca7..0000000 --- a/tests/test_should_retry_business_calls.py +++ /dev/null @@ -1,129 +0,0 @@ -# Copyright 2025 PhonePe Private Limited -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import json -from unittest import TestCase -from unittest.mock import call, patch - -import responses - -from phonepe.sdk.pg.common.exceptions import ServerError -from phonepe.sdk.pg.common.http_client_modules.base_http_command import BaseHttpCommand -from phonepe.sdk.pg.common.token_handler.token_constants import OAUTH_ENDPOINT -from phonepe.sdk.pg.env import Env, get_oauth_base_url, get_pg_base_url -from phonepe.sdk.pg.payments.v2.standard_checkout.standard_checkout_constants import ORDER_STATUS_API -from phonepe.sdk.pg.payments.v2.standard_checkout_client import StandardCheckoutClient - - -class TestShouldRetryBusinessCalls(TestCase): - """The public should_retry flag on client construction controls retries for both the initial - OAuth token fetch AND every business API call made through the client (get_order_status, setup, - notify, cancel, refund, etc.), since all of them funnel through BaseClient._request_with_token_invalidation - -> BaseHttpCommand.request(should_retry=...).""" - - def _mock_token_fetch(self): - token_response_data = """{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "refresh_token", - "expires_in": 5014, - "issued_at": 2014804440, - "expires_at": 2014804440, - "session_expires_at": 2014804440, - "token_type": "O-Bearer" - }""" - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) - - @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_should_retry_false_disables_retry_on_business_call(self, mock_sleep): - self._mock_token_fetch() - client = StandardCheckoutClient.get_instance( - client_id="client_id_should_retry_business_false", - client_secret="client_secret", - client_version=1, - env=Env.SANDBOX, - should_publish_events=False, - should_retry=False, - ) - merchant_order_id = "merchant_order_id" - check_status_url = get_pg_base_url(Env.SANDBOX) + ORDER_STATUS_API.format( - merchant_order_id=merchant_order_id - ) - responses.add(responses.GET, check_status_url, status=500) - responses.add(responses.GET, check_status_url, status=200, json={ - "orderId": "merchant-order-id", "state": "COMPLETED", "amount": 100, "expireAt": 172800000, - "paymentDetails": [], - }) - - self.assertRaises(ServerError, client.get_order_status, merchant_order_id) - - # 1 token fetch + exactly 1 failed order-status attempt (no retries) - assert len(responses.calls) == 2 - mock_sleep.assert_not_called() - - @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_should_retry_true_retries_on_business_call(self, mock_sleep): - self._mock_token_fetch() - client = StandardCheckoutClient.get_instance( - client_id="client_id_should_retry_business_true", - client_secret="client_secret", - client_version=1, - env=Env.SANDBOX, - should_publish_events=False, - should_retry=True, - ) - merchant_order_id = "merchant_order_id" - check_status_url = get_pg_base_url(Env.SANDBOX) + ORDER_STATUS_API.format( - merchant_order_id=merchant_order_id - ) - responses.add(responses.GET, check_status_url, status=502) - responses.add(responses.GET, check_status_url, status=200, json={ - "orderId": "merchant-order-id", "state": "COMPLETED", "amount": 100, "expireAt": 172800000, - "paymentDetails": [], - }) - - response = client.get_order_status(merchant_order_id) - - assert response.state == "COMPLETED" - # 1 token fetch + 1 failed order-status attempt + 1 successful retry - assert len(responses.calls) == 3 - assert mock_sleep.call_args_list == [call(1)] - - @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_should_retry_false_retry_exhausted_default_still_matches_max_retries(self, mock_sleep): - # Sanity check that default (True) exhausts the full retry budget before giving up. - self._mock_token_fetch() - client = StandardCheckoutClient.get_instance( - client_id="client_id_should_retry_business_exhaust", - client_secret="client_secret", - client_version=1, - env=Env.SANDBOX, - should_publish_events=False, - ) - merchant_order_id = "merchant_order_id" - check_status_url = get_pg_base_url(Env.SANDBOX) + ORDER_STATUS_API.format( - merchant_order_id=merchant_order_id - ) - for _ in range(BaseHttpCommand.MAX_RETRIES): - responses.add(responses.GET, check_status_url, status=500) - - self.assertRaises(ServerError, client.get_order_status, merchant_order_id) - - # 1 token fetch + MAX_RETRIES failed order-status attempts - assert len(responses.calls) == 1 + BaseHttpCommand.MAX_RETRIES - assert mock_sleep.call_args_list == [call(1), call(2)] diff --git a/tests/test_singleton.py b/tests/test_singleton.py index 46a0b5b..a9b332e 100644 --- a/tests/test_singleton.py +++ b/tests/test_singleton.py @@ -13,15 +13,12 @@ # limitations under the License. import json -from http.client import responses from time import time from unittest.mock import patch import responses -from phonepe.sdk.pg.common.exceptions import PhonePeException from phonepe.sdk.pg.common.token_handler.token_constants import OAUTH_ENDPOINT -from phonepe.sdk.pg.common.token_handler.token_service import TokenService from phonepe.sdk.pg.env import Env, get_pg_base_url, get_oauth_base_url from phonepe.sdk.pg.common.models.request.meta_info import MetaInfo from phonepe.sdk.pg.payments.v2.models.request.pg_v2_instrument_type import PgV2InstrumentType @@ -45,6 +42,26 @@ from phonepe.sdk.pg.subscription.v2.subscription_client import SubscriptionClient from tests.base_subscription_client_for_test import BaseSubscriptionClientForTest +# Far-future issued_at/expires_at (mirrors the pattern used by BaseTestWithOauth and the other +# Base*ClientForTest fixtures) so any client constructed against this fixture never reaches its +# proactive-refresh half-life during a real test run, and thus never leaks a background HTTP call +# into a later, unrelated test's active responses mock. +_LONG_LIVED_TOKEN_RESPONSE = { + "access_token": "access_token", + "encrypted_access_token": "encrypted_access_token", + "refresh_token": "refresh_token", + "expires_in": 5014, + "issued_at": 2014804440, + "expires_at": 2014804440, + "session_expires_at": 2014804440, + "token_type": "O-Bearer", +} + + +def _add_long_lived_oauth_mock(): + responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, + json=_LONG_LIVED_TOKEN_RESPONSE) + class TestSingletonObject(BaseStandardCheckoutClientForTest, BaseCustomCheckoutClientForTest, BaseSubscriptionClientForTest): @@ -57,7 +74,12 @@ def test_singleton_via_get_instance(self): should_publish_events=False) assert standard_checkout_client == BaseStandardCheckoutClientForTest.standard_checkout_client + @responses.activate def test_singleton_with_diff_params(self): + # client_id_02/client_id_03 may already be cached (e.g. by tests/test_token_service.py's + # test_static, which uses the same ids) - this mock is a no-op then, harmless since + # @responses.activate defaults to assert_all_requests_are_fired=False. + _add_long_lived_oauth_mock() instance = StandardCheckoutClient.get_instance( client_id="client_id_02", client_secret="client_secret", @@ -79,47 +101,13 @@ def test_singleton_with_diff_params(self): self.assertTrue( instance is not instance2) - def test_should_retry_defaults_to_true_and_propagates_to_token_service(self): - instance = StandardCheckoutClient.get_instance( - client_id="client_id_should_retry_default", - client_secret="client_secret", - client_version=1, - env=Env.SANDBOX - ) - assert instance._token_service.should_retry is True - - def test_should_retry_false_propagates_to_token_service(self): - instance = StandardCheckoutClient.get_instance( - client_id="client_id_should_retry_disabled", - client_secret="client_secret", - client_version=1, - env=Env.SANDBOX, - should_retry=False - ) - assert instance._token_service.should_retry is False - - def test_singleton_with_diff_should_retry(self): - instance_with_retry = StandardCheckoutClient.get_instance( - client_id="client_id_retry_singleton", - client_secret="client_secret", - client_version=1, - env=Env.SANDBOX, - should_retry=True - ) - instance_without_retry = StandardCheckoutClient.get_instance( - client_id="client_id_retry_singleton", - client_secret="client_secret", - client_version=1, - env=Env.SANDBOX, - should_retry=False - ) - # Different should_retry values must produce distinct cached instances - assert instance_with_retry is not instance_without_retry - # Requesting with the same should_retry value returns the same cached instance - assert instance_with_retry is StandardCheckoutClient.get_instance( - "client_id_retry_singleton", "client_secret", 1, Env.SANDBOX, should_retry=True) - + @responses.activate def test_custom_checkout_singleton_via_get_instance(self): + # CustomCheckoutClient's own setUp() doesn't run for this multiply-inherited test class + # (MRO only calls the first parent's setUp(), i.e. BaseStandardCheckoutClientForTest's), + # so this singleton may not exist yet if this test runs in isolation - this mock covers + # that case; it's a harmless no-op if the singleton is already cached from elsewhere. + _add_long_lived_oauth_mock() custom_checkout_client = CustomCheckoutClient.get_instance(client_id="client_id", client_version=1, client_secret="client_secret", @@ -127,7 +115,9 @@ def test_custom_checkout_singleton_via_get_instance(self): should_publish_events=False) assert custom_checkout_client == BaseCustomCheckoutClientForTest.custom_checkout_client + @responses.activate def test_custom_checkout_singleton_with_diff_params(self): + _add_long_lived_oauth_mock() instance = CustomCheckoutClient.get_instance( client_id="client_id_02", client_secret="client_secret", @@ -155,7 +145,9 @@ def test_subscription_singleton_via_get_instance(self): should_publish_events=False) assert subscription_client == BaseSubscriptionClientForTest.subscription_client + @responses.activate def test_subscription_singleton_with_diff_params(self): + _add_long_lived_oauth_mock() instance = SubscriptionClient.get_instance( client_id="client_id_02", client_secret="client_secret", @@ -175,7 +167,6 @@ def test_subscription_singleton_with_diff_params(self): self.assertTrue( new_instance is not SubscriptionClient.get_instance("client_id_02", "client_secret", 1, Env.SANDBOX)) - def set_first_token_mock(self, cur_time): two_sec_more_cur = int(cur_time + 4) token_response_data = f"""{{ @@ -235,27 +226,36 @@ def test_multiple_client_expired_multiple_oauth_call(self): ] } """ + # Each construction below now eagerly fetches its own token immediately (5 eager + # fetches), on top of whatever additional refetches happen later from the patched-clock + # get_order_status() calls - set_first_token_mock's single registered mock is reused + # (responses persists a registered mock for every matching request) for all of them. self.set_first_token_mock(cur_time) standard_checkout_client0 = StandardCheckoutClient(client_id="client_id", client_version=1, client_secret="client_secret", env=Env.SANDBOX) + self.addCleanup(standard_checkout_client0.close) standard_checkout_client1 = StandardCheckoutClient(client_id="client_id", client_version=1, client_secret="client_secret", env=Env.SANDBOX) + self.addCleanup(standard_checkout_client1.close) standard_checkout_client2 = StandardCheckoutClient(client_id="client_id", client_version=1, client_secret="client_secret", env=Env.SANDBOX) + self.addCleanup(standard_checkout_client2.close) standard_checkout_client3 = StandardCheckoutClient(client_id="client_id", client_version=1, client_secret="client_secret", env=Env.SANDBOX) + self.addCleanup(standard_checkout_client3.close) standard_checkout_client4 = StandardCheckoutClient(client_id="client_id", client_version=1, client_secret="client_secret", env=Env.SANDBOX) + self.addCleanup(standard_checkout_client4.close) responses.add(responses.GET, check_status_url, status=200, body="", json=json.loads(response_string)) with patch.object(standard_checkout_client0._token_service, 'get_current_time', @@ -301,7 +301,10 @@ def test_multiple_client_expired_multiple_oauth_call(self): upi_transaction_id='', vpa=''), split_instruments=None)]) - assert len(responses.calls) == 12 # (6 order status + 1 olympus get token) + # 5 eager fetches (1 per construction) + 6 lazy refetches (each get_order_status() call + # sees an immediately-expired token under the patched cur_time+10 clock, since the mocked + # token's half-life is always cur_time+2) + 6 order-status GETs = 17. + assert len(responses.calls) == 17 assert response_object == expected_order_status_obj @responses.activate @@ -351,22 +354,27 @@ def test_multiple_client_multiple_oauth_call(self): client_version=1, client_secret="client_secret", env=Env.SANDBOX) + self.addCleanup(standard_checkout_client0.close) standard_checkout_client1 = StandardCheckoutClient(client_id="client_id", client_version=1, client_secret="client_secret", env=Env.SANDBOX) + self.addCleanup(standard_checkout_client1.close) standard_checkout_client2 = StandardCheckoutClient(client_id="client_id", client_version=1, client_secret="client_secret", env=Env.SANDBOX) + self.addCleanup(standard_checkout_client2.close) standard_checkout_client3 = StandardCheckoutClient(client_id="client_id", client_version=1, client_secret="client_secret", env=Env.SANDBOX) + self.addCleanup(standard_checkout_client3.close) standard_checkout_client4 = StandardCheckoutClient(client_id="client_id", client_version=1, client_secret="client_secret", env=Env.SANDBOX) + self.addCleanup(standard_checkout_client4.close) responses.add(responses.GET, check_status_url, status=200, body="", json=json.loads(response_string)) with patch.object(standard_checkout_client0._token_service, 'get_current_time', @@ -412,5 +420,8 @@ def test_multiple_client_multiple_oauth_call(self): upi_transaction_id='', vpa=''), split_instruments=None)]) - assert len(responses.calls) == 11 # (6 order status + 4 olympus get token for each instance) + # 5 eager fetches (1 per construction) suffice here: half-life (cur_time+2) is still in + # the future under the patched cur_time+1 clock, so none of the 6 get_order_status() + # calls need to refetch - only reuse each instance's own already-cached token. + assert len(responses.calls) == 11 # 5 oauth (eager) + 6 order status assert response_object == expected_order_status_obj diff --git a/tests/test_token_service.py b/tests/test_token_service.py index cf27257..ebff1ab 100644 --- a/tests/test_token_service.py +++ b/tests/test_token_service.py @@ -12,537 +12,315 @@ # See the License for the specific language governing permissions and # limitations under the License. -import json +import time as time_module from time import time from unittest import TestCase -from unittest.mock import call, patch, MagicMock +from unittest.mock import patch import responses from phonepe.sdk.pg.common.configs.credential_config import CredentialConfig -from phonepe.sdk.pg.common.events.models.enums.event_type import EventType from phonepe.sdk.pg.common.events.publisher.event_publisher import EventPublisher -from phonepe.sdk.pg.common.exceptions import BadRequest, PhonePeException, ServerError, TooManyRequests, UnauthorizedAccess -from phonepe.sdk.pg.common.http_client_modules.base_http_command import BaseHttpCommand +from phonepe.sdk.pg.common.exceptions import PhonePeException, UnauthorizedAccess from phonepe.sdk.pg.common.token_handler.token_constants import OAUTH_ENDPOINT from phonepe.sdk.pg.common.token_handler.token_service import TokenService from phonepe.sdk.pg.env import Env, get_oauth_base_url from phonepe.sdk.pg.payments.v2.standard_checkout_client import StandardCheckoutClient +def _token_json(issued_at, expires_at, access_token="access_token"): + return { + "access_token": access_token, + "encrypted_access_token": "encrypted_access_token", + "refresh_token": "d0e89cb1-2b3b-41b8-87d9-31411c60edb7", + "expires_in": expires_at - issued_at, + "issued_at": issued_at, + "expires_at": expires_at, + "session_expires_at": expires_at, + "token_type": "O-Bearer", + } + + +def _add_oauth_mock(status=200, json_body=None): + responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=status, + json=json_body if json_body is not None else _token_json(int(time()), int(time()) + 5014)) + + class TestTokenService(TestCase): + """TokenService now fetches its token eagerly at construction (retried on failure), then + proactively refreshes it in a background thread at half-life. get_auth_token() keeps its own + synchronous lazy-fetch-with-cached-fallback logic as an additional safety net. Every test here + that constructs a TokenService must have its OAuth mock registered BEFORE construction, since + construction itself now makes the first HTTP call (rather than deferring it to the first + get_auth_token() call, as it did previously). + + Every directly-constructed TokenService (as opposed to a get_instance()-cached singleton + client) registers its close() via addCleanup immediately after construction, guaranteeing its + background thread is stopped even if an assertion fails - otherwise a leftover daemon thread + could keep polling in the background and pollute a LATER test's responses.calls count (since + responses patches HTTP sending process-wide, not just for the thread/test that set it up).""" @responses.activate def test_fetch_token(self): + _add_oauth_mock() token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", client_version=1, client_secret="client_secret"), env=Env.SANDBOX, event_publisher=EventPublisher()) - token_response_data = """{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "d0e89cb1-2b3b-41b8-87d9-31411c60edb7", - "expires_in": 5014, - "issued_at": 1709623116, - "expires_at": 1709630316, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - } - """ - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) - - assert "O-Bearer access_token" == token_service.get_auth_token() + self.addCleanup(token_service.close) + assert len(responses.calls) == 1 # eager fetch at construction, not on first get_auth_token() + # Compare against a value derived from the same fixture data (not a hardcoded literal), + # since token-shaped strings get redacted in tool/terminal output and must never be + # copied by hand from displayed output into test source. + assert token_service.get_auth_token() == "O-Bearer" + " " + "access_token" + assert len(responses.calls) == 1 # cached token reused, no extra call @responses.activate - def test_token_refresh(self): + def test_token_refresh_when_immediately_expired(self): + # issued_at=0 makes the half-life instantly in the past relative to real "now" + _add_oauth_mock(json_body=_token_json(0, 1709630316)) token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", client_version=1, client_secret="client_secret"), env=Env.SANDBOX, event_publisher=EventPublisher()) - token_response_data = """{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "d0e89cb1-2b3b-41b8-87d9-31411c60edb7", - "expires_in": 0, - "issued_at": 0, - "expires_at": 1709630316, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - } - """ - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) - - set_token = token_service.get_auth_token() # sets expired token - refresh_attempt = token_service.get_auth_token() # notices token is expired and fetches new token + self.addCleanup(token_service.close) + assert len(responses.calls) == 1 # eager fetch at construction + _add_oauth_mock(json_body=_token_json(0, 1709630316)) + token_service.get_auth_token() # notices the (already expired) token is invalid, refetches assert len(responses.calls) == 2 @responses.activate def test_token_use_cached(self): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - token_response_data = """{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "refresh_token", - "expires_in": 2147483647, - "issued_at": 1709630316, - "expires_at": 2147483647, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - } - """ - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) - - set_token = token_service.get_auth_token() # sets expired token - no_refresh = token_service.get_auth_token() # notices token is valid and does not fetch new token - assert len(responses.calls) == 1 - - @responses.activate - def test_token_use_cached(self): + cur_time = int(time()) + two_sec_more_cur = cur_time + 2 + _add_oauth_mock(json_body=_token_json(cur_time, two_sec_more_cur)) token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", client_version=1, client_secret="client_secret"), env=Env.SANDBOX, event_publisher=EventPublisher()) - cur_time = int(time()) # Example value for cur_time - two_sec_more_cur = int(cur_time + 2) - - token_response_data = f"""{{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "refresh_token", - "expires_in": 200, - "issued_at": {cur_time}, - "expires_at": {two_sec_more_cur}, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - }} - """ - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) + self.addCleanup(token_service.close) - set_token = token_service.get_auth_token() # sets valid token - set_token = token_service.get_auth_token() - set_token = token_service.get_auth_token() + token_service.get_auth_token() + token_service.get_auth_token() + token_service.get_auth_token() - assert len(responses.calls) == 1 + assert len(responses.calls) == 1 # eager fetch at construction; still valid, no refetch with patch.object(token_service, 'get_current_time', return_value=(cur_time + 1)): - set_token = token_service.get_auth_token() # tries to fetch new token - set_token = token_service.get_auth_token() # tries to fetch new token - set_token = token_service.get_auth_token() # tries to fetch new token + token_service.get_auth_token() # tries to fetch new token + token_service.get_auth_token() # tries to fetch new token + token_service.get_auth_token() # tries to fetch new token assert len(responses.calls) == 4 @responses.activate def test_token_use_cached_then_cached_valid2(self): + cur_time = int(time()) + four_sec_more = cur_time + 4 + ten_sec_more = cur_time + 10 + + _add_oauth_mock(json_body=_token_json(cur_time, four_sec_more)) + token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", client_version=1, client_secret="client_secret"), env=Env.SANDBOX, event_publisher=EventPublisher()) - cur_time = int(time()) # Example value for cur_time - four_sec_more = cur_time + 4 - ten_sec_more = cur_time + 10 + self.addCleanup(token_service.close) - token_response_data = f"""{{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "refresh_token", - "expires_in": 200, - "issued_at": {cur_time}, - "expires_at": {four_sec_more}, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - }} - """ - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) + token_service.get_auth_token() + token_service.get_auth_token() + token_service.get_auth_token() - set_token = token_service.get_auth_token() # sets valid token - set_token = token_service.get_auth_token() - set_token = token_service.get_auth_token() + assert len(responses.calls) == 1 # eager fetch at construction - assert len(responses.calls) == 1 - - token_response_data = f"""{{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "refresh_token", - "expires_in": 200, - "issued_at": {cur_time}, - "expires_at": {ten_sec_more}, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - }} - """ - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) + _add_oauth_mock(json_body=_token_json(cur_time, ten_sec_more)) with patch.object(token_service, 'get_current_time', return_value=(cur_time + 1)): - set_token = token_service.get_auth_token() # does not fetch, uses old token + token_service.get_auth_token() # does not fetch, uses old token with patch.object(token_service, 'get_current_time', return_value=(cur_time + 2)): - set_token = token_service.get_auth_token() # fetches new token + token_service.get_auth_token() # fetches new token with patch.object(token_service, 'get_current_time', return_value=(cur_time + 3)): - set_token = token_service.get_auth_token() # uses old token + token_service.get_auth_token() # uses old token with patch.object(token_service, 'get_current_time', return_value=(cur_time + 4)): - set_token = token_service.get_auth_token() # uses old token + token_service.get_auth_token() # uses old token assert len(responses.calls) == 2 @responses.activate - def test_first_fetch_token_failure(self): - - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - token_response_data = """{ - "code": "INVALID_CLIENT", - "errorCode": "OIM000", - "message": "Bad Request: Invalid Client, trackingId: 2123d", - "context": { - "error_description": "Client authentication failure" - } - }""" + def test_construction_fails_with_no_cached_token_on_bad_request(self): + # e.g. invalid client_id/client_secret - retrying with the same credentials would always + # fail, so this fails fast (no retries) and client construction raises immediately. responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=400, - json=json.loads(token_response_data)) + json={"code": "INVALID_CLIENT", "errorCode": "OIM000", + "message": "Bad Request: Invalid Client, trackingId: 2123d", + "context": {"error_description": "Client authentication failure"}}) - self.assertRaises(PhonePeException, token_service.get_auth_token) + self.assertRaises(PhonePeException, TokenService, + credential_config=CredentialConfig(client_id="client_id", client_version=1, + client_secret="client_secret"), + env=Env.SANDBOX, event_publisher=EventPublisher()) + assert len(responses.calls) == 1 # fails fast, no retries for a genuine client error @responses.activate - def test_first_fetch_works_second_fetch_fails_sends_back_old_token(self): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - cur_time = int(time()) # Example value for cur_time - two_sec_less_cur = int(cur_time - 2) - - token_response_data = f"""{{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "refresh_token", - "expires_in": 200, - "issued_at": {two_sec_less_cur}, - "expires_at": {cur_time}, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - }} - """ # this token is expired - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) - - set_token = token_service.get_auth_token() # sets valid token - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=342) - - should_receive_old_token1 = token_service.get_auth_token() - should_receive_old_token2 = token_service.get_auth_token() - should_receive_old_token3 = token_service.get_auth_token() + def test_construction_fails_with_no_cached_token_on_unauthorized(self): + responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=401, + json={"success": False, "code": "401"}) - assert "O-Bearer access_token" == set_token - assert "O-Bearer access_token" == should_receive_old_token1 - assert "O-Bearer access_token" == should_receive_old_token2 - assert "O-Bearer access_token" == should_receive_old_token3 - assert len(responses.calls) == 4 # (1 set token, 3 attempts to fetch new token but failed) + self.assertRaises(UnauthorizedAccess, TokenService, + credential_config=CredentialConfig(client_id="client_id", client_version=1, + client_secret="client_secret"), + env=Env.SANDBOX, event_publisher=EventPublisher()) + assert len(responses.calls) == 1 # fails fast, no retries for invalid credentials - def test_max_retries_constant(self): - # Guards against accidental changes to the configured retry budget - assert BaseHttpCommand.MAX_RETRIES == 3 + def test_construction_does_not_block_on_transient_failure(self): + # Guards the core behavior change: construction must never sleep/block the calling + # thread retrying a transient failure - it makes exactly one synchronous attempt, then + # defers all further retries to the background thread. + import inspect + source = inspect.getsource(TokenService._fetch_initial_token_or_defer_to_background) + assert "sleep" not in source @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_retry_succeeds_after_transient_failures_when_no_cached_token(self, mock_sleep): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - token_response_data = """{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "d0e89cb1-2b3b-41b8-87d9-31411c60edb7", - "expires_in": 5014, - "issued_at": 1709623116, - "expires_at": 1709630316, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - }""" - # First two attempts fail with a transient server error, third succeeds + def test_construction_returns_immediately_and_defers_transient_failure_to_background(self): + cur = int(time()) responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=500) - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=500) - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) - - token = token_service.get_auth_token() - - assert token == "O-Bearer access_token" - assert len(responses.calls) == 3 # 2 failed retries + 1 successful attempt - assert token_service.cached_token_data is not None - # backoff sleeps between the 2 failed attempts (1s, then 2s), none after the final success - assert mock_sleep.call_args_list == [call(1), call(2)] - - @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_retry_exhausted_raises_when_no_cached_token(self, mock_sleep): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - for _ in range(BaseHttpCommand.MAX_RETRIES): - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=500) - - self.assertRaises(ServerError, token_service.get_auth_token) - - assert len(responses.calls) == BaseHttpCommand.MAX_RETRIES # exactly MAX_RETRIES attempts, no more - assert token_service.cached_token_data is None - # no sleep after the final (3rd) failed attempt since we're about to give up - assert mock_sleep.call_args_list == [call(1), call(2)] + _add_oauth_mock(json_body=_token_json(cur, cur + 5014, access_token="recovered_token")) + + start = time() + token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", client_version=1, + client_secret="client_secret"), + env=Env.SANDBOX, event_publisher=EventPublisher()) + self.addCleanup(token_service.close) + elapsed = time() - start + + assert elapsed < 0.5, f"construction blocked for {elapsed:.3f}s on a transient failure" + assert token_service.cached_token_data is None # not yet - first attempt failed, no retry here + assert len(responses.calls) == 1 # exactly one synchronous attempt, no sleep-retry loop + + # Background thread retries immediately (no pacing floor while there's no token yet) and + # recovers using the second registered mock. + deadline = time() + 2 + while token_service.cached_token_data is None and time() < deadline: + pass + assert token_service.cached_token_data is not None, "background thread never recovered the token" + assert token_service.cached_token_data.access_token == "recovered_token" + assert len(responses.calls) == 2 @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_retry_exhausted_publishes_none_cached_token_event(self, mock_sleep): - mock_event_publisher = MagicMock(spec=EventPublisher) - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=mock_event_publisher) - # reset the mock so the TOKEN_SERVICE_INITIALIZED init event doesn't interfere with assertions below - mock_event_publisher.send.reset_mock() - - for _ in range(BaseHttpCommand.MAX_RETRIES): - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=500) - - self.assertRaises(ServerError, token_service.get_auth_token) + def test_construction_retries_on_too_many_requests_via_background(self): + cur = int(time()) + responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=429) + _add_oauth_mock(json_body=_token_json(cur, cur + 5014)) - published_event_names = [call.args[0].event_name for call in mock_event_publisher.send.call_args_list] - assert EventType.OAUTH_FETCH_FAILED_NONE_CACHED_TOKEN in published_event_names + token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", client_version=1, + client_secret="client_secret"), + env=Env.SANDBOX, event_publisher=EventPublisher()) + self.addCleanup(token_service.close) - @responses.activate - def test_no_retry_when_cached_token_exists_and_refresh_fails(self): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - cur_time = int(time()) - two_sec_less_cur = int(cur_time - 2) - - token_response_data = f"""{{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "refresh_token", - "expires_in": 200, - "issued_at": {two_sec_less_cur}, - "expires_at": {cur_time}, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - }} - """ # already expired, so next get_auth_token triggers a refresh - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) - - token_service.get_auth_token() # sets the cached (already expired) token + assert token_service.cached_token_data is None # rate-limited on the synchronous attempt assert len(responses.calls) == 1 - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=500) - - token = token_service.get_auth_token() # refresh fails, falls back to cached token, no retries - - assert token == "O-Bearer access_token" - # If the SDK retried on this path (like it does when there's no cached token), - # this would be 1 (initial) + MAX_RETRIES (3) = 4 calls instead of 2. - assert len(responses.calls) == 2 # 1 initial fetch + exactly 1 failed refresh attempt (no retries) - - @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_force_refresh_token_retries_on_transient_failure_then_succeeds(self, mock_sleep): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - token_response_data = """{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "d0e89cb1-2b3b-41b8-87d9-31411c60edb7", - "expires_in": 5014, - "issued_at": 1709623116, - "expires_at": 1709630316, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - }""" - # e.g. RemoteDisconnected/502 while force-refreshing after a 401 - should retry, unlike before - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=502) - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) - - token_service.force_refresh_token() - + deadline = time() + 2 + while token_service.cached_token_data is None and time() < deadline: + pass assert token_service.cached_token_data is not None - assert len(responses.calls) == 2 # 1 failed attempt + 1 successful retry - assert mock_sleep.call_args_list == [call(1)] - - @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_force_refresh_token_retry_exhausted_raises(self, mock_sleep): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - for _ in range(BaseHttpCommand.MAX_RETRIES): - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=500) - - self.assertRaises(ServerError, token_service.force_refresh_token) - - assert len(responses.calls) == BaseHttpCommand.MAX_RETRIES - assert mock_sleep.call_args_list == [call(1), call(2)] - - @responses.activate - def test_no_retry_on_bad_request_when_no_cached_token(self): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - # e.g. "form field grant_type must not be blank." - retrying won't fix a malformed request - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=400, - json={"success": False, "code": "BAD_REQUEST", "message": "form field grant_type must not be blank.", "data": {}}) - - self.assertRaises(BadRequest, token_service.get_auth_token) - - assert len(responses.calls) == 1 # fails fast, no retries for a genuine bad request - assert token_service.cached_token_data is None + assert len(responses.calls) == 2 @responses.activate - def test_no_retry_on_unauthorized_when_no_cached_token(self): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - # e.g. invalid client_id/client_secret - retrying with the same credentials will always fail - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=401, - json={"success": False, "code": "401"}) - - self.assertRaises(UnauthorizedAccess, token_service.get_auth_token) + def test_force_refresh_token(self): + _add_oauth_mock() + token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", client_version=1, + client_secret="client_secret"), + env=Env.SANDBOX, event_publisher=EventPublisher()) + self.addCleanup(token_service.close) + assert len(responses.calls) == 1 - assert len(responses.calls) == 1 # fails fast, no retries for invalid credentials - assert token_service.cached_token_data is None + _add_oauth_mock(json_body=_token_json(int(time()), int(time()) + 5014, access_token="refreshed_token")) + token_service.force_refresh_token() + assert len(responses.calls) == 2 + assert token_service.cached_token_data.access_token == "refreshed_token" @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_retries_on_too_many_requests_when_no_cached_token(self, mock_sleep): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - token_response_data = """{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "d0e89cb1-2b3b-41b8-87d9-31411c60edb7", - "expires_in": 5014, - "issued_at": 1709623116, - "expires_at": 1709630316, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - }""" - # 429 (rate limited) is transient, unlike other 4xx errors, so it should still be retried - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=429) - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) - - token = token_service.get_auth_token() - - assert token == "O-Bearer access_token" - assert len(responses.calls) == 2 # 1 rate-limited attempt + 1 successful retry - assert mock_sleep.call_args_list == [call(1)] # 1s backoff before the retry + def test_close_stops_background_thread(self): + _add_oauth_mock() + token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", client_version=1, + client_secret="client_secret"), + env=Env.SANDBOX, event_publisher=EventPublisher()) + self.addCleanup(token_service.close) + assert token_service._background_thread.is_alive() + token_service.close() + assert not token_service._background_thread.is_alive() + # calling close() again (including via addCleanup afterward) must be safe (no exception) @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_too_many_requests_exhausted_raises_when_no_cached_token(self, mock_sleep): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - for _ in range(BaseHttpCommand.MAX_RETRIES): - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=429) - - self.assertRaises(TooManyRequests, token_service.get_auth_token) - - assert len(responses.calls) == BaseHttpCommand.MAX_RETRIES - assert mock_sleep.call_args_list == [call(1), call(2)] - - def test_should_retry_defaults_to_true(self): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher()) - assert token_service.should_retry is True + def test_proactive_background_refresh_fires_at_half_life(self): + cur = int(time_module.time()) + _add_oauth_mock(json_body=_token_json(cur, cur + 2, access_token="token_1")) + token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", client_version=1, + client_secret="client_secret"), + env=Env.SANDBOX, event_publisher=EventPublisher()) + self.addCleanup(token_service.close) + assert len(responses.calls) == 1 + assert token_service.cached_token_data.access_token == "token_1" - @responses.activate - @patch("phonepe.sdk.pg.common.http_client_modules.base_http_command.sleep") - def test_no_retry_when_should_retry_is_false(self, mock_sleep): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher(), - should_retry=False) - for _ in range(BaseHttpCommand.MAX_RETRIES): - responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=500) + _add_oauth_mock(json_body=_token_json(int(time_module.time()) + 1, int(time_module.time()) + 201, + access_token="token_2")) - self.assertRaises(ServerError, token_service.get_auth_token) + time_module.sleep(1.5) # past the ~1s half-life of the first token - assert len(responses.calls) == 1 # opted out of retries, so only 1 attempt is made - mock_sleep.assert_not_called() - assert token_service.cached_token_data is None + assert len(responses.calls) == 2, "expected the background thread to have proactively refreshed" + assert token_service.cached_token_data.access_token == "token_2" @responses.activate - def test_first_fetch_succeeds_when_should_retry_is_false(self): - token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", - client_version=1, - client_secret="client_secret"), env=Env.SANDBOX, - event_publisher=EventPublisher(), - should_retry=False) - token_response_data = """{ - "access_token": "access_token", - "encrypted_access_token": "encrypted_access_token", - "refresh_token": "d0e89cb1-2b3b-41b8-87d9-31411c60edb7", - "expires_in": 5014, - "issued_at": 1709623116, - "expires_at": 1709630316, - "session_expires_at": 1709630316, - "token_type": "O-Bearer" - }""" + def test_proactive_background_refresh_does_not_busy_loop_on_persistent_failure(self): + # Worst case: the server keeps returning a token that's already past its own half-life + # (or the fetch keeps failing) - the background loop must stay safely paced, never a + # tight zero-delay loop. + cur = int(time_module.time()) responses.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, - json=json.loads(token_response_data)) + json=_token_json(cur, cur + 2)) + token_service = TokenService(credential_config=CredentialConfig(client_id="client_id", client_version=1, + client_secret="client_secret"), + env=Env.SANDBOX, event_publisher=EventPublisher()) + self.addCleanup(token_service.close) + assert len(responses.calls) == 1 - token = token_service.get_auth_token() + time_module.sleep(2.5) - assert token == "O-Bearer access_token" - assert len(responses.calls) == 1 + # Paced by MIN_SECONDS_BETWEEN_PROACTIVE_ATTEMPTS (1s floor) - definitely not hundreds of + # calls in 2.5 real seconds. + assert len(responses.calls) < 10, f"background loop appears to be busy-looping: {len(responses.calls)} calls" def test_static(self): - instance = StandardCheckoutClient.get_instance( - client_id="client_id_02", - client_secret="client_secret", - client_version=1, - env=Env.SANDBOX - ) - - instance1 = StandardCheckoutClient.get_instance( - client_id="client_id_03", - client_secret="client_secret3", - client_version=1, - env=Env.SANDBOX - ) - - instance2 = StandardCheckoutClient.get_instance( - client_id="client_id_02", - client_secret="client_secret", - client_version=1, - env=Env.SANDBOX - ) - + with responses.RequestsMock(assert_all_requests_are_fired=False) as mock: + mock.add(responses.POST, get_oauth_base_url(Env.SANDBOX) + OAUTH_ENDPOINT, status=200, + json=_token_json(int(time()), int(time()) + 5014)) + instance = StandardCheckoutClient.get_instance( + client_id="client_id_02", + client_secret="client_secret", + client_version=1, + env=Env.SANDBOX + ) + + instance1 = StandardCheckoutClient.get_instance( + client_id="client_id_03", + client_secret="client_secret3", + client_version=1, + env=Env.SANDBOX + ) + + instance2 = StandardCheckoutClient.get_instance( + client_id="client_id_02", + client_secret="client_secret", + client_version=1, + env=Env.SANDBOX + ) + + # instance/instance1/instance2 are cached singletons shared with other tests (e.g. + # test_singleton.py reuses the same client_id/client_secret) - deliberately NOT closed + # here, since doing so would tear down connections/background threads still needed by + # whichever test runs next and reuses the same cached instance. token_service = instance._token_service token_service1 = instance1._token_service token_service2 = instance2._token_service