test: add tests for EventsManager

box · Feb 16, 2024 · 3efc750 · 3efc750
1 parent 4a42b1a
commit 3efc750
Show file tree

Hide file tree

Showing 9 changed files with 181 additions and 53 deletions.
diff --git a/box_sdk_gen/base_object.py b/box_sdk_gen/base_object.py
@@ -83,7 +83,8 @@ def _deserialize_union(cls, key, value, annotation):
 
         for possible_type in possible_types:
             if (
-                value.get(possible_type._discriminator[0], None)
+                issubclass(possible_type, BaseObject)
+                and value.get(possible_type._discriminator[0], None)
                 in possible_type._discriminator[1]
             ):
                 return cls._deserialize(key, value, possible_type)

diff --git a/box_sdk_gen/ccg_auth.py b/box_sdk_gen/ccg_auth.py
@@ -2,9 +2,9 @@
 
 from typing import List
 
-from box_sdk_gen.managers.authorization import RequestAccessTokenGrantType
+from box_sdk_gen.schemas import PostOAuth2TokenGrantTypeField
 
-from box_sdk_gen.managers.authorization import RequestAccessTokenSubjectTokenType
+from box_sdk_gen.schemas import PostOAuth2TokenSubjectTokenTypeField
 
 from box_sdk_gen.schemas import AccessToken
 
@@ -61,13 +61,21 @@ def __init__(self, config: CCGConfig, **kwargs):
         """
         super().__init__(**kwargs)
         self.config = config
-        self.token_storage = self.config.token_storage
+        self.token_storage = (
+            InMemoryTokenStorage()
+            if self.config.token_storage == None
+            else self.config.token_storage
+        )
         self.subject_id = (
             self.config.user_id
             if not self.config.user_id == None
             else self.config.enterprise_id
         )
-        self.subject_type = 'user' if not self.config.user_id == None else 'enterprise'
+        self.subject_type = (
+            PostOAuth2TokenBoxSubjectTypeField.USER.value
+            if not self.config.user_id == None
+            else PostOAuth2TokenBoxSubjectTypeField.ENTERPRISE.value
+        )
 
     def refresh_token(
         self, network_session: Optional[NetworkSession] = None
@@ -82,8 +90,8 @@ def refresh_token(
             if not network_session == None
             else AuthorizationManager()
         )
-        token: AccessToken = auth_manager.request_access_token(
-            grant_type=RequestAccessTokenGrantType.CLIENT_CREDENTIALS.value,
+        token: Optional[AccessToken] = auth_manager.request_access_token(
+            grant_type=PostOAuth2TokenGrantTypeField.CLIENT_CREDENTIALS.value,
             client_id=self.config.client_id,
             client_secret=self.config.client_secret,
             box_subject_type=self.subject_type,
@@ -100,7 +108,7 @@ def retrieve_token(
         :param network_session: An object to keep network session state
         :type network_session: Optional[NetworkSession], optional
         """
-        old_token = self.token_storage.get()
+        old_token: Optional[AccessToken] = self.token_storage.get()
         if old_token == None:
             new_token: AccessToken = self.refresh_token(network_session)
             return new_token
@@ -184,9 +192,9 @@ def downscope_token(
             else AuthorizationManager()
         )
         downscoped_token: AccessToken = auth_manager.request_access_token(
-            grant_type=RequestAccessTokenGrantType.URN_IETF_PARAMS_OAUTH_GRANT_TYPE_TOKEN_EXCHANGE.value,
+            grant_type=PostOAuth2TokenGrantTypeField.URN_IETF_PARAMS_OAUTH_GRANT_TYPE_TOKEN_EXCHANGE.value,
             subject_token=token.access_token,
-            subject_token_type=RequestAccessTokenSubjectTokenType.URN_IETF_PARAMS_OAUTH_TOKEN_TYPE_ACCESS_TOKEN.value,
+            subject_token_type=PostOAuth2TokenSubjectTokenTypeField.URN_IETF_PARAMS_OAUTH_TOKEN_TYPE_ACCESS_TOKEN.value,
             resource=resource,
             scope=' '.join(scopes),
             box_shared_link=shared_link,

diff --git a/box_sdk_gen/fetch.py b/box_sdk_gen/fetch.py
@@ -113,7 +113,7 @@ def fetch(url: str, options: FetchOptions) -> FetchResponse:
         if response.raised_exception and attempt_nr > 1:
             break
 
-        if response.network_response != None:
+        if response.network_response is not None:
             network_response = response.network_response
             if network_response.ok:
                 if options.response_format == 'binary':
@@ -137,7 +137,8 @@ def fetch(url: str, options: FetchOptions) -> FetchResponse:
                     )
 
             if (
-                network_response.status_code != 429
+                not response.reauthentication_needed
+                and network_response.status_code != 429
                 and network_response.status_code < 500
             ):
                 __raise_on_unsuccessful_request(request=request, response=response)

diff --git a/box_sdk_gen/jwt_auth.py b/box_sdk_gen/jwt_auth.py
@@ -8,9 +8,9 @@
 
 from typing import List
 
-from box_sdk_gen.managers.authorization import RequestAccessTokenGrantType
+from box_sdk_gen.schemas import PostOAuth2TokenGrantTypeField
 
-from box_sdk_gen.managers.authorization import RequestAccessTokenSubjectTokenType
+from box_sdk_gen.schemas import PostOAuth2TokenSubjectTokenTypeField
 
 from box_sdk_gen.auth import Authentication
 
@@ -50,8 +50,6 @@
 
 from box_sdk_gen.errors import BoxSDKError
 
-box_jwt_audience: str = 'https://api.box.com/oauth2/token'
-
 
 class JwtConfigAppSettingsAppAuth(BaseObject):
     _fields_to_json_mapping: Dict[str, str] = {
@@ -160,7 +158,7 @@ def __init__(
         private_key_passphrase: str,
         enterprise_id: Optional[str] = None,
         user_id: Optional[str] = None,
-        jwt_algorithm: Optional[JwtAlgorithm] = 'RS256',
+        algorithm: Optional[JwtAlgorithm] = JwtAlgorithm.RS256.value,
         token_storage: TokenStorage = None,
     ):
         """
@@ -188,7 +186,7 @@ def __init__(
         self.private_key_passphrase = private_key_passphrase
         self.enterprise_id = enterprise_id
         self.user_id = user_id
-        self.jwt_algorithm = jwt_algorithm
+        self.algorithm = algorithm
         self.token_storage = token_storage
 
     @staticmethod
@@ -208,7 +206,7 @@ def from_config_json_string(
         config_json: JwtConfigFile = deserialize(
             json_to_serialized_data(config_json_string), JwtConfigFile
         )
-        new_config = (
+        new_config: 'JWTConfig' = (
             JWTConfig(
                 client_id=config_json.box_app_settings.client_id,
                 client_secret=config_json.box_app_settings.client_secret,
@@ -258,7 +256,11 @@ def __init__(self, config: JWTConfig, **kwargs):
         """
         super().__init__(**kwargs)
         self.config = config
-        self.token_storage = self.config.token_storage
+        self.token_storage = (
+            InMemoryTokenStorage()
+            if self.config.token_storage == None
+            else self.config.token_storage
+        )
         self.subject_id = (
             self.config.enterprise_id
             if not self.config.enterprise_id == None
@@ -281,17 +283,17 @@ def refresh_token(
                 message='JWT auth is not supported in browser environment.'
             )
         alg: JwtAlgorithm = (
-            self.config.jwt_algorithm
-            if not self.config.jwt_algorithm == None
-            else 'RS256'
+            self.config.algorithm
+            if not self.config.algorithm == None
+            else JwtAlgorithm.RS256.value
         )
         claims: Dict = {
             'exp': get_epoch_time_in_seconds() + 30,
             'box_sub_type': self.subject_type,
         }
         jwt_options: JwtSignOptions = JwtSignOptions(
             algorithm=alg,
-            audience=box_jwt_audience,
+            audience='https://api.box.com/oauth2/token',
             subject=self.subject_id,
             issuer=self.config.client_id,
             jwtid=get_uuid(),
@@ -307,7 +309,7 @@ def refresh_token(
             else AuthorizationManager()
         )
         token: AccessToken = auth_manager.request_access_token(
-            grant_type=RequestAccessTokenGrantType.URN_IETF_PARAMS_OAUTH_GRANT_TYPE_JWT_BEARER.value,
+            grant_type=PostOAuth2TokenGrantTypeField.URN_IETF_PARAMS_OAUTH_GRANT_TYPE_JWT_BEARER.value,
             assertion=assertion,
             client_id=self.config.client_id,
             client_secret=self.config.client_secret,
@@ -323,7 +325,7 @@ def retrieve_token(
         :param network_session: An object to keep network session state
         :type network_session: Optional[NetworkSession], optional
         """
-        old_token = self.token_storage.get()
+        old_token: Optional[AccessToken] = self.token_storage.get()
         if old_token == None:
             new_token: AccessToken = self.refresh_token(network_session)
             return new_token
@@ -415,18 +417,15 @@ def downscope_token(
             else AuthorizationManager()
         )
         downscoped_token: AccessToken = auth_manager.request_access_token(
-            grant_type=RequestAccessTokenGrantType.URN_IETF_PARAMS_OAUTH_GRANT_TYPE_TOKEN_EXCHANGE.value,
+            grant_type=PostOAuth2TokenGrantTypeField.URN_IETF_PARAMS_OAUTH_GRANT_TYPE_TOKEN_EXCHANGE.value,
             subject_token=token.access_token,
-            subject_token_type=RequestAccessTokenSubjectTokenType.URN_IETF_PARAMS_OAUTH_TOKEN_TYPE_ACCESS_TOKEN.value,
+            subject_token_type=PostOAuth2TokenSubjectTokenTypeField.URN_IETF_PARAMS_OAUTH_TOKEN_TYPE_ACCESS_TOKEN.value,
             resource=resource,
             scope=' '.join(scopes),
             box_shared_link=shared_link,
         )
         return downscoped_token
 
-    def add_space(self, text: str) -> str:
-        return ''.join([])
-
     def revoke_token(self, network_session: Optional[NetworkSession] = None) -> None:
         """
         Revoke the current access token and remove it from token storage.

diff --git a/box_sdk_gen/managers/authorization.py b/box_sdk_gen/managers/authorization.py
@@ -10,6 +10,12 @@
 
 from box_sdk_gen.serialization import deserialize
 
+from box_sdk_gen.schemas import PostOAuth2TokenGrantTypeField
+
+from box_sdk_gen.schemas import PostOAuth2TokenSubjectTokenTypeField
+
+from box_sdk_gen.schemas import PostOAuth2TokenBoxSubjectTypeField
+
 from box_sdk_gen.schemas import AccessToken
 
 from box_sdk_gen.schemas import OAuth2Error

diff --git a/box_sdk_gen/oauth.py b/box_sdk_gen/oauth.py
@@ -2,11 +2,13 @@
 
 from typing import Dict
 
+from box_sdk_gen.serialization import serialize
+
 from typing import List
 
-from box_sdk_gen.managers.authorization import RequestAccessTokenGrantType
+from box_sdk_gen.schemas import PostOAuth2TokenGrantTypeField
 
-from box_sdk_gen.managers.authorization import RequestAccessTokenSubjectTokenType
+from box_sdk_gen.schemas import PostOAuth2TokenSubjectTokenTypeField
 
 from box_sdk_gen.auth import Authentication
 
@@ -30,7 +32,7 @@
 
 from box_sdk_gen.errors import BoxSDKError
 
-box_oauth_2_auth_url: str = 'https://account.box.com/api/oauth2/authorize'
+from box_sdk_gen.json_data import SerializedData
 
 
 class OAuthConfig:
@@ -80,15 +82,19 @@ def __init__(self, config: OAuthConfig, **kwargs):
         """
         super().__init__(**kwargs)
         self.config = config
-        self.token_storage = self.config.token_storage
+        self.token_storage = (
+            InMemoryTokenStorage()
+            if self.config.token_storage == None
+            else self.config.token_storage
+        )
 
     def get_authorize_url(self, options: GetAuthorizeUrlOptions = None) -> str:
         """
         Get the authorization URL for the app user.
         """
         if options is None:
             options = GetAuthorizeUrlOptions()
-        params: Dict[str, str] = prepare_params(
+        params_map: Dict[str, str] = prepare_params(
             {
                 'client_id': (
                     options.client_id
@@ -105,7 +111,12 @@ def get_authorize_url(self, options: GetAuthorizeUrlOptions = None) -> str:
                 'scope': options.scope,
             }
         )
-        return ''.join([box_oauth_2_auth_url, '?', sd_to_url_params(params)])
+        return ''.join(
+            [
+                'https://account.box.com/api/oauth2/authorize?',
+                sd_to_url_params(serialize(params_map)),
+            ]
+        )
 
     def get_tokens_authorization_code_grant(
         self, authorization_code: str, network_session: Optional[NetworkSession] = None
@@ -123,7 +134,7 @@ def get_tokens_authorization_code_grant(
             else AuthorizationManager()
         )
         token: AccessToken = auth_manager.request_access_token(
-            grant_type=RequestAccessTokenGrantType.AUTHORIZATION_CODE.value,
+            grant_type=PostOAuth2TokenGrantTypeField.AUTHORIZATION_CODE.value,
             code=authorization_code,
             client_id=self.config.client_id,
             client_secret=self.config.client_secret,
@@ -139,38 +150,32 @@ def retrieve_token(
         :param network_session: An object to keep network session state
         :type network_session: Optional[NetworkSession], optional
         """
-        token = self.token_storage.get()
+        token: Optional[AccessToken] = self.token_storage.get()
         if token == None:
             raise BoxSDKError(
                 message='Access and refresh tokens not available. Authenticate before making any API call first.'
             )
         return token
 
     def refresh_token(
-        self,
-        network_session: Optional[NetworkSession] = None,
-        refresh_token: Optional[str] = None,
+        self, network_session: Optional[NetworkSession] = None
     ) -> AccessToken:
         """
         Get a new access token for the app user.
         :param network_session: An object to keep network session state
         :type network_session: Optional[NetworkSession], optional
-        :param refresh_token: A refresh token to use
-        :type refresh_token: Optional[str], optional
         """
         old_token: Optional[AccessToken] = self.token_storage.get()
-        token_used_for_refresh = (
-            refresh_token
-            if not refresh_token == None
-            else old_token.refresh_token if not old_token == None else None
+        token_used_for_refresh: Optional[str] = (
+            old_token.refresh_token if not old_token == None else None
         )
         auth_manager: AuthorizationManager = (
             AuthorizationManager(network_session=network_session)
             if not network_session == None
             else AuthorizationManager()
         )
         token: AccessToken = auth_manager.request_access_token(
-            grant_type=RequestAccessTokenGrantType.REFRESH_TOKEN.value,
+            grant_type=PostOAuth2TokenGrantTypeField.REFRESH_TOKEN.value,
             client_id=self.config.client_id,
             client_secret=self.config.client_secret,
             refresh_token=token_used_for_refresh,
@@ -225,9 +230,9 @@ def downscope_token(
             else AuthorizationManager()
         )
         downscoped_token: AccessToken = auth_manager.request_access_token(
-            grant_type=RequestAccessTokenGrantType.URN_IETF_PARAMS_OAUTH_GRANT_TYPE_TOKEN_EXCHANGE.value,
+            grant_type=PostOAuth2TokenGrantTypeField.URN_IETF_PARAMS_OAUTH_GRANT_TYPE_TOKEN_EXCHANGE.value,
             subject_token=token.access_token,
-            subject_token_type=RequestAccessTokenSubjectTokenType.URN_IETF_PARAMS_OAUTH_TOKEN_TYPE_ACCESS_TOKEN.value,
+            subject_token_type=PostOAuth2TokenSubjectTokenTypeField.URN_IETF_PARAMS_OAUTH_TOKEN_TYPE_ACCESS_TOKEN.value,
             resource=resource,
             scope=' '.join(scopes),
             box_shared_link=shared_link,

diff --git a/docs/events.md b/docs/events.md
@@ -20,7 +20,11 @@ This operation is performed by calling function `get_events`.
 See the endpoint docs at
 [API Reference](https://developer.box.com/reference/get-events/).
 
-_Currently we don't have an example for calling `get_events` in integration tests_
+<!-- sample get_events -->
+
+```python
+client.events.get_events(stream_type=GetEventsStreamType.CHANGES.value)
+```
 
 ### Arguments
 
@@ -92,7 +96,11 @@ This operation is performed by calling function `get_events_with_long_polling`.
 See the endpoint docs at
 [API Reference](https://developer.box.com/reference/options-events/).
 
-_Currently we don't have an example for calling `get_events_with_long_polling` in integration tests_
+<!-- sample options_events -->
+
+```python
+client.events.get_events_with_long_polling()
+```
 
 ### Arguments