Authentication updates

.flake8

@@ -0,0 +1,3 @@
+[flake8]
+per-file-ignores =
+    tests/*:E501

bookops_worldcat/authorize.py

@@ -6,13 +6,11 @@
 
 import datetime
 import sys
-from typing import Dict, List, Optional, Tuple, Union
+from typing import Dict, Optional, Tuple, Union
 
 import requests
-from requests import Response
 
-
-from . import __title__, __version__  # type: ignore
+from . import __title__, __version__
 from .errors import WorldcatAuthorizationError
 
 
@@ -23,13 +21,14 @@ class WorldcatAccessToken:
     Explicit Authorization Code and Refresh Token flows. Token with correctly
     bonded scopes can then be passed into a session of particular web service
     to authorize requests for resources.
-    More on OCLC's web services authorization:
-    https://www.oclc.org/developer/develop/authentication/oauth/client-credentials-grant.en.html
+    More on OCLC's client credentials grant:
+    https://www.oclc.org/developer/api/keys/oauth/client-credentials-grant.en.html
 
     Args:
         key:                    your WSKey public client_id
         secret:                 your WSKey secret
-        scopes:                 request scopes for the access token
+        scopes:                 request scopes for the access token as a string,
+                                separate different scopes with space
         principal_id:           principalID (required for read/write endpoints)
         principal_idns:         principalIDNS (required for read/write endpoints)
         agent:                  "User-agent" parameter to be passed in the request
@@ -75,10 +74,10 @@ def __init__(
         self,
         key: str,
         secret: str,
-        scopes: Union[str, List[str]],
+        scopes: str,
         principal_id: str,
         principal_idns: str,
-        agent: Optional[str] = None,
+        agent: str = "",
         timeout: Optional[
             Union[int, float, Tuple[int, int], Tuple[float, float]]
         ] = None,
@@ -93,51 +92,49 @@ def __init__(
         self.principal_idns = principal_idns
         self.scopes = scopes
         self.secret = secret
-        self.server_response = None
+        self.server_response: Optional[requests.Response] = None
         self.timeout = timeout
-        self.token_expires_at = None
-        self.token_str = None
-        self.token_type = None
+        self.token_expires_at: Optional[datetime.datetime] = None
+        self.token_str = ""
+        self.token_type = ""
 
         # default bookops-worldcat request header
-        if self.agent is None:
+        if not self.agent:
             self.agent = f"{__title__}/{__version__}"
         else:
-            if type(self.agent) is not str:
+            if not isinstance(self.agent, str):
                 raise WorldcatAuthorizationError("Argument 'agent' must be a string.")
 
         # asure passed arguments are valid
         if not self.key:
             raise WorldcatAuthorizationError("Argument 'key' is required.")
         else:
-            if type(self.key) is not str:
+            if not isinstance(self.key, str):
                 raise WorldcatAuthorizationError("Argument 'key' must be a string.")
 
         if not self.secret:
             raise WorldcatAuthorizationError("Argument 'secret' is required.")
         else:
-            if type(self.secret) is not str:
+            if not isinstance(self.secret, str):
                 raise WorldcatAuthorizationError("Argument 'secret' must be a string.")
 
         if not self.principal_id:
             raise WorldcatAuthorizationError(
-                "Argument 'principal_id' is required for read/write endpoint of Metadata API."
+                "Argument 'principal_id' is required for read/write endpoint of "
+                "Metadata API."
             )
         if not self.principal_idns:
             raise WorldcatAuthorizationError(
-                "Argument 'principal_idns' is required for read/write endpoint of Metadata API."
+                "Argument 'principal_idns' is required for read/write endpoint of "
+                "Metadata API."
             )
 
         # validate passed scopes
-        if type(self.scopes) is list:
-            self.scopes = " ".join(self.scopes)
-        elif type(self.scopes) is not str:
-            raise WorldcatAuthorizationError(
-                "Argument 'scopes' must a string or a list."
-            )
-        self.scopes = self.scopes.strip()  # type: ignore
-        if self.scopes == "":
-            raise WorldcatAuthorizationError("Argument 'scope' is missing.")
+        if not self.scopes:
+            raise WorldcatAuthorizationError("Argument 'scopes' is required.")
+        elif not isinstance(self.scopes, str):
+            raise WorldcatAuthorizationError("Argument 'scopes' must a string.")
+        self.scopes = self.scopes.strip()
 
         # assign default value for timout
         if not self.timeout:
@@ -149,7 +146,7 @@ def __init__(
     def _auth(self) -> Tuple[str, str]:
         return (self.key, self.secret)
 
-    def _hasten_expiration_time(self, utc_stamp_str: str) -> str:
+    def _hasten_expiration_time(self, utc_stamp_str: str) -> datetime.datetime:
         """
         Resets expiration time one second earlier to account
         for any delays between expiration check and request for
@@ -164,14 +161,14 @@ def _hasten_expiration_time(self, utc_stamp_str: str) -> str:
         utcstamp = datetime.datetime.strptime(
             utc_stamp_str, "%Y-%m-%d %H:%M:%SZ"
         ) - datetime.timedelta(seconds=1)
-        return datetime.datetime.strftime(utcstamp, "%Y-%m-%d %H:%M:%SZ")
+        return utcstamp
 
-    def _parse_server_response(self, response: Response) -> None:
+    def _parse_server_response(self, response: requests.Response) -> None:
         """Parses authorization server response"""
-        self.server_response = response  # type: ignore
+        self.server_response = response
         if response.status_code == requests.codes.ok:
             self.token_str = response.json()["access_token"]
-            self.token_expires_at = self._hasten_expiration_time(  # type: ignore
+            self.token_expires_at = self._hasten_expiration_time(
                 response.json()["expires_at"]
             )
             self.token_type = response.json()["token_type"]
@@ -182,12 +179,12 @@ def _payload(self) -> Dict[str, str]:
         """Preps requests params"""
         return {
             "grant_type": self.grant_type,
-            "scope": self.scopes,  # type: ignore
+            "scope": self.scopes,
             "principalID": self.principal_id,
             "principalIDNS": self.principal_idns,
         }
 
-    def _post_token_request(self) -> Response:
+    def _post_token_request(self) -> requests.Response:
         """
         Fetches Worldcat access token for specified scope (web service)
 
@@ -222,7 +219,7 @@ def _request_token(self):
         self._parse_server_response(response)
 
     def _token_headers(self) -> Dict[str, str]:
-        return {"User-Agent": self.agent, "Accept": "application/json"}  # type: ignore
+        return {"User-Agent": self.agent, "Accept": "application/json"}
 
     def _token_url(self) -> str:
         return f"{self.oauth_server}/token"
@@ -238,20 +235,16 @@ def is_expired(self) -> bool:
         >>> token.is_expired()
         False
         """
-        try:
-            if (
-                datetime.datetime.strptime(self.token_expires_at, "%Y-%m-%d %H:%M:%SZ")  # type: ignore
-                < datetime.datetime.utcnow()
-            ):
+        if isinstance(self.token_expires_at, datetime.datetime):
+            if self.token_expires_at < datetime.datetime.utcnow():
                 return True
             else:
                 return False
-        except TypeError:
-            raise
-        except ValueError:
-            raise
+        else:
+            raise TypeError
 
     def __repr__(self):
         return (
-            f"access_token: '{self.token_str}', expires_at: '{self.token_expires_at}'"
+            f"access_token: '{self.token_str}', "
+            f"expires_at: '{self.token_expires_at:%Y-%m-%d %H:%M:%SZ}'"
         )

pyproject.toml

@@ -32,7 +32,10 @@ classifiers = [
 python = "^3.8"
 requests = "^2.31"
 
-[tool.poetry.dev-dependencies]
+[tool.poetry.urls]
+"Bug Tracker" = "https://github.com/BookOps-CAT/bookops-worldcat/issues"
+
+[tool.poetry.group.dev.dependencies]
 pytest = "^7.0"
 pytest-cov = "^3.0"
 pytest-mock = "^3.7"
@@ -41,9 +44,7 @@ black = "^23.3.0"
 mike = "^2.0.0"
 mkapi = "^1.0.14"
 mypy = "^1.8"
-
-[tool.poetry.urls]
-"Bug Tracker" = "https://github.com/BookOps-CAT/bookops-worldcat/issues"
+types-requests = "^2.31.0.20240125"
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

tests/conftest.py

@@ -134,7 +134,7 @@ def mock_credentials():
     return {
         "key": "my_WSkey",
         "secret": "my_WSsecret",
-        "scopes": ["scope1", "scope2"],
+        "scopes": "scope1 scope2",
         "principal_id": "my_principalID",
         "principal_idns": "my_principalIDNS",
     }

tests/test_authorize.py

@@ -156,22 +156,22 @@ def test_principal_idns_exception(self, arg, expectation, msg):
             (
                 None,
                 pytest.raises(WorldcatAuthorizationError),
-                "Argument 'scope' must a string or a list.",
+                "Argument 'scopes' must a string.",
             ),
             (
                 123,
                 pytest.raises(WorldcatAuthorizationError),
-                "Argument 'scope' must a string or a list.",
+                "Argument 'scopes' must a string.",
             ),
             (
                 " ",
                 pytest.raises(WorldcatAuthorizationError),
-                "Argument 'scope' is missing.",
+                "Argument 'scopes' is required.",
             ),
             (
                 ["", ""],
                 pytest.raises(WorldcatAuthorizationError),
-                "Argument 'scope' is missing.",
+                "Argument 'scopes' is required.",
             ),
         ],
     )
@@ -211,7 +211,7 @@ def test_timeout_argument(
 
     @pytest.mark.parametrize(
         "argm,expectation",
-        [("scope1", "scope1"), (["scope1", "scope2"], "scope1 scope2")],
+        [("scope1 ", "scope1"), (" scope1 scope2 ", "scope1 scope2")],
     )
     def test_scope_manipulation(
         self, argm, expectation, mock_successful_post_token_response
@@ -263,7 +263,9 @@ def test_auth(self, mock_successful_post_token_response):
     def test_hasten_expiration_time(self, mock_token):
         utc_stamp = "2020-01-01 17:19:59Z"
         token = mock_token
-        assert token._hasten_expiration_time(utc_stamp) == "2020-01-01 17:19:58Z"
+        timestamp = token._hasten_expiration_time(utc_stamp)
+        assert isinstance(timestamp, datetime.datetime)
+        assert timestamp == datetime.datetime(2020, 1, 1, 17, 19, 58, 0)
 
     def test_payload(self, mock_successful_post_token_response):
         token = WorldcatAccessToken(
@@ -346,16 +348,15 @@ def test_is_expired_false(
 
     def test_is_expired_true(self, mock_utcnow, mock_token):
         mock_token.is_expired() is False
-        mock_token.token_expires_at = datetime.datetime.strftime(
-            datetime.datetime.utcnow() - datetime.timedelta(0, 1),
-            "%Y-%m-%d %H:%M:%SZ",
+        mock_token.token_expires_at = datetime.datetime.utcnow() - datetime.timedelta(
+            0, 1
         )
 
         assert mock_token.is_expired() is True
 
     @pytest.mark.parametrize(
         "arg,expectation",
-        [(None, pytest.raises(TypeError)), ("20-01-01", pytest.raises(ValueError))],
+        [(None, pytest.raises(TypeError))],
     )
     def test_is_expired_exception(self, arg, expectation, mock_token):
         mock_token.token_expires_at = arg
@@ -436,5 +437,6 @@ def test_post_token_request_with_live_service(self, live_keys):
         assert sorted(params) == sorted(response.keys())
 
         # test if token looks right
-        assert token.token_str is not None
+        assert token.token_str.startswith("tk_")
         assert token.is_expired() is False
+        assert isinstance(token.token_expires_at, datetime.datetime)