Merge pull request #3 from nassauwinter/session_attrs

Modify signature of the ZephyrSession constructor to be able to pass any session attributes

Author: nassauwinter

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = zephyr-python-api
-version = 0.0.1
+version = 0.0.2
 author = Petr Sharapenko
 author_email = nassauwinter@gmail.com
 description = Zephyr (TM4J) Python REST API wrapper

tests/unit/test_zephyr_session.py

@@ -0,0 +1,95 @@
+import pytest
+from requests import Session
+
+from zephyr.scale.scale import DEFAULT_BASE_URL, ZephyrSession
+from zephyr.scale.zephyr_session import INIT_SESSION_MSG, InvalidAuthData
+
+REQUESTS_SESSION_PATH = "requests.sessions.Session"
+GETLOGGER_PATH = "logging.getLogger"
+LOGGER_DEBUG_PATH = "logging.Logger.debug"
+
+
+@pytest.mark.unit
+class TestZephyrSession:
+    def test_creation(self, mocker):
+        """Tests basic creation logic"""
+        logger_mock = mocker.patch(GETLOGGER_PATH)
+
+        zsession = ZephyrSession(DEFAULT_BASE_URL, token="token_test")
+
+        assert zsession.base_url == DEFAULT_BASE_URL, (f"Attribute base_url expected to be {DEFAULT_BASE_URL}, "
+                                                       f"not {zsession.base_url}")
+        assert isinstance(zsession._session, Session)
+        logger_mock.assert_called_with("zephyr.scale.zephyr_session")
+
+    def test_token_auth(self, mocker):
+        """Test token auth"""
+        token = "test_token"
+        logger_mock = mocker.patch(LOGGER_DEBUG_PATH)
+
+        zsession = ZephyrSession(DEFAULT_BASE_URL, token=token)
+
+        logger_mock.assert_called_with(INIT_SESSION_MSG.format("token"))
+        assert f"Bearer {token}" == zsession._session.headers.get("Authorization")
+
+    def test_credentials_auth(self, mocker):
+        """Test auth with username and password"""
+        username = "usertest"
+        password = "pwdtest"
+        logger_mock = mocker.patch(LOGGER_DEBUG_PATH)
+
+        zsession = ZephyrSession(DEFAULT_BASE_URL, username=username, password=password)
+
+        logger_mock.assert_called_with(INIT_SESSION_MSG.format("username and password"))
+        assert (username, password) == zsession._session.auth
+
+    def test_cookie_auth(self, mocker):
+        """Test auth with cookie dict"""
+        test_cookie = {"cookies": {"cookie.token": "cookie_test"}}
+        logger_mock = mocker.patch(LOGGER_DEBUG_PATH)
+
+        zsession = ZephyrSession(DEFAULT_BASE_URL, cookies=test_cookie)
+
+        logger_mock.assert_called_with(INIT_SESSION_MSG.format("cookies"))
+        assert test_cookie['cookies'] in zsession._session.cookies.values()
+
+    @pytest.mark.parametrize("auth_data, exception", [(dict(), InvalidAuthData),
+                                                      ({"username": "user"}, InvalidAuthData),
+                                                      ({"password": "pwd"}, InvalidAuthData)])
+    def test_auth_exception(self, auth_data, exception):
+        """Test exceptions on auth"""
+        with pytest.raises(exception):
+            ZephyrSession(DEFAULT_BASE_URL, **auth_data)
+
+    @pytest.mark.parametrize("creation_kwargs",
+                             [{"token": "token_test",
+                               "session_attrs": {'verify': False, "max_redirects": 333}}])
+    def test_requests_session_attrs(self, creation_kwargs, mocker):
+        """The test checks ZephyrScale (not) provided with "session_attrs"."""
+        logger_mock = mocker.patch(LOGGER_DEBUG_PATH)
+        session_attrs = creation_kwargs.get('session_attrs')
+
+        zsession = ZephyrSession(DEFAULT_BASE_URL, **creation_kwargs)
+
+        logger_mock.assert_called_with(
+            f"Modify requests session object with {session_attrs}")
+
+        for attrib, value in session_attrs.items():
+            actual = getattr(zsession._session, attrib)
+            assert actual == session_attrs[attrib], (f"Request session attr {attrib} is {actual}, "
+                                                     f"but expected{session_attrs[attrib]}")
+
+    @pytest.mark.skip
+    def test_crud_request(self):
+        """Test GET, POST, PUT, DELETE requests"""
+        pass
+
+    @pytest.mark.skip
+    def test_get_paginated(self):
+        """Test paginated request"""
+        pass
+
+    @pytest.mark.skip
+    def test_post_file(self):
+        """Test Post file wrapper"""
+        pass

zephyr/scale/zephyr_session.py

@@ -1,12 +1,16 @@
 import logging
 from urllib.parse import urlparse, parse_qs
 
-from requests import Session
+from requests import HTTPError, Session
 
 
 INIT_SESSION_MSG = "Initialize session by {}"
 
 
+class InvalidAuthData(Exception):
+    """Invalid authentication data provided"""
+
+
 class ZephyrSession:
     """
     Zephyr Scale basic session object.
@@ -16,11 +20,15 @@ class ZephyrSession:
     :param username: username
     :param password: password
     :param cookies: cookie dict
+
+    :keyword session_attrs: a dict with session attrs to be set as keys and their values
     """
-    def __init__(self, base_url, token=None, username=None, password=None, cookies=None):
+    def __init__(self, base_url, token=None, username=None, password=None, cookies=None, **kwargs):
         self.base_url = base_url
         self._session = Session()
+
         self.logger = logging.getLogger(__name__)
+
         if token:
             self.logger.debug(INIT_SESSION_MSG.format("token"))
             self._session.headers.update({"Authorization": f"Bearer {token}"})
@@ -31,12 +39,21 @@ def __init__(self, base_url, token=None, username=None, password=None, cookies=N
             self.logger.debug(INIT_SESSION_MSG.format("cookies"))
             self._session.cookies.update(cookies)
         else:
-            raise Exception("Insufficient auth data")
+            raise InvalidAuthData("Insufficient auth data")
+
+        if kwargs.get("session_attrs"):
+            self._modify_session(**kwargs.get("session_attrs"))
 
     def _create_url(self, *args):
         """Helper for URL creation"""
         return self.base_url + "/".join(args)
 
+    def _modify_session(self, **kwargs):
+        """Modify requests session with extra arguments"""
+        self.logger.debug(f"Modify requests session object with {kwargs}")
+        for session_attr, value in kwargs.items():
+            setattr(self._session, session_attr, value)
+
     def _request(self, method: str, endpoint: str, return_raw: bool = False, **kwargs):
         """General request wrapper with logging and handling response"""
         self.logger.debug(f"{method.capitalize()} data: endpoint={endpoint} and {kwargs}")
@@ -48,7 +65,7 @@ def _request(self, method: str, endpoint: str, return_raw: bool = False, **kwarg
             if response.text:
                 return response.json()
             return ""
-        raise Exception(f"Error {response.status_code}. Response: {response.content}")
+        raise HTTPError(f"Error {response.status_code}. Response: {response.content}")
 
     def get(self, endpoint: str, params: dict = None, **kwargs):
         """Get request wrapper"""