refactor: make session auth to use base service's http client

Author: eiri

ibmcloudant/cloudant_base_service.py

@@ -27,7 +27,6 @@
 from ibm_cloud_sdk_core import BaseService
 from ibm_cloud_sdk_core.authenticators import Authenticator
 from requests import Response, Session
-from requests.cookies import RequestsCookieJar
 
 from .common import get_sdk_headers
 from .couchdb_session_authenticator import CouchDbSessionAuthenticator
@@ -96,9 +95,8 @@ def __init__(
             self.set_http_config(new_http_config)
         # Custom actions for CouchDbSessionAuthenticator
         if isinstance(authenticator, CouchDbSessionAuthenticator):
-            # Replacing BaseService's http.cookiejar.CookieJar as RequestsCookieJar supports update(CookieJar)
-            self.jar = RequestsCookieJar(self.jar)
-            self.authenticator.set_jar(self.jar)  # Authenticators don't have access to cookie jars by default
+            # Make token manager of CouchDbSessionAuthenticator to use the same http client as main service
+            self.authenticator.set_http_client(self.get_http_client(), self.jar)
         add_hooks(self)
 
     def set_service_url(self, service_url: str):

ibmcloudant/couchdb_session_authenticator.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# © Copyright IBM Corporation 2020, 2022.
+# © Copyright IBM Corporation 2020, 2025.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,7 +16,9 @@
 """
 Module for handling session authentication
 """
-from requests import Request
+from requests import Request, Session
+from requests.cookies import RequestsCookieJar
+
 
 from ibm_cloud_sdk_core.authenticators import Authenticator
 from .couchdb_session_token_manager import CouchDbSessionTokenManager
@@ -46,20 +48,22 @@ def __init__(self,
         if not isinstance(disable_ssl_verification, bool):
             raise TypeError('disable_ssl_verification must be a bool')
 
-        self.jar = None
-
         self.token_manager = CouchDbSessionTokenManager(
             username,
             password,
             disable_ssl_verification=disable_ssl_verification
         )
         self.validate()
 
-    def set_jar(self, jar):
-        """Sets the cookie jar for the authenticator.
+    def set_http_client(self, http_client: Session, jar: RequestsCookieJar) -> None:
+        """Sets base serivice's http client for the authenticator.
         This is an internal method called by BaseService. Not to be called directly.
         """
-        self.jar = jar
+        if isinstance(http_client, Session):
+            self.token_manager.http_client = http_client
+            self.token_manager.jar = jar
+        else:
+            raise TypeError("http_client parameter must be a requests.sessions.Session")
 
     def validate(self):
         """Validates the username, and password for session token requests.
@@ -82,11 +86,7 @@ def authenticate(self, req: Request):
         Args:
             req: Ignored. BaseService uses the cookie jar for every request
         """
-        jar = self.token_manager.get_token()
-        # Requests seem to save cookies only for Sessions. BaseService is
-        # hard-coded to work with "regular" requests requests so updating
-        # the jar manually is necessary
-        self.jar.update(jar)
+        self.token_manager.get_token()
 
     def authentication_type(self) -> str:
         """Returns this authenticator's type ('COUCHDB_SESSION')."""

ibmcloudant/couchdb_session_token_manager.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-# © Copyright IBM Corporation 2020, 2022.
+# © Copyright IBM Corporation 2020, 2025.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,6 +16,8 @@
 """
 Module for managing session authentication token
 """
+from requests import Session
+
 from ibm_cloud_sdk_core.token_managers.token_manager import TokenManager
 
 
@@ -52,7 +54,9 @@ def __init__(self, username: str, password: str,
 
         self.token = None
 
+        self.http_client = Session()
         self.http_config = {}
+        self.jar = None
         self.headers = None
 
     def request_token(self):
@@ -63,14 +67,16 @@ def request_token(self):
              A CookieJar of Cookies the server sent back.
         """
 
-        response = self._request(
+        response = self.http_client.request(
             method='POST',
             url=self.url + "/_session",
             headers=self.headers,
             json={
                 'username': self.username,
                 'password': self.password,
-            }
+            },
+            cookies=self.jar,
+            **self.http_config
         )
 
         return response

test/integration/test_timeout.py

@@ -1,6 +1,6 @@
 # coding: utf-8
 
-#  © Copyright IBM Corporation 2021.
+#  © Copyright IBM Corporation 2021, 2025.
 #  #
 #  Licensed under the Apache License, Version 2.0 (the "License");
 #  you may not use this file except in compliance with the License.
@@ -86,8 +86,9 @@ def get_authenticate_arguments(req):
         return req.mock_calls[0]
 
     @staticmethod
-    def get_request_arguments(tci, srv, idx):
-        tci.assertEqual(srv.http_client.request.call_count, idx + 1)
+    def get_request_arguments(tci, srv, idx, call_count=None):
+        call_count = idx + 1 if call_count is None else call_count
+        tci.assertEqual(srv.http_client.request.call_count, call_count)
         return srv.http_client.request.mock_calls[idx]
 
 # Every method tests an authenticator.
@@ -142,10 +143,6 @@ def test_timeout_cloudantv1_sessionauth(self):
         )
         my_service.set_service_url("http://cloudant.example")
 
-        # Mock out authentication
-        orig_request = requests.request
-        requests.request = Mock(return_value=Helpers.get_mocked_response())
-
         # Mock out request response
         Helpers.mock_out_cloudant_request(my_service)
 
@@ -157,17 +154,16 @@ def test_timeout_cloudantv1_sessionauth(self):
             # Call the server
             my_service.get_server_information()
 
-            # Assert timeout is set to the authenticator
-            auth_args = Helpers.get_authenticate_arguments(requests.request)
-            Helpers.assert_default_timeout_setting(self, auth_args)
+            # first call is to a _session end-point, so we have an additional call on our mock request
+            if tc_num == 0:
+                # Assert timeout is set to the authenticator
+                auth_args = Helpers.get_request_arguments(self, my_service.authenticator.token_manager, tc_num, tc_num+2)
+                Helpers.assert_default_timeout_setting(self, auth_args)
 
             # Assert timeout is set in the server request
-            req_args = Helpers.get_request_arguments(self, my_service, tc_num)
+            req_args = Helpers.get_request_arguments(self, my_service, tc_num+1, tc_num+2)
             tc['assert_func'](self, req_args)
 
-        # Set back requests.request
-        requests.request = orig_request
-
     def test_timeout_cloudantv1_iamauth(self):
         authenticator = IAMAuthenticator('apikey')
         my_service = CloudantV1(