From 01c10444b862ae348376e596cdefa7d52adc8ae1 Mon Sep 17 00:00:00 2001 From: Arthur Borem Date: Thu, 24 Jul 2025 20:57:47 -0500 Subject: [PATCH 1/2] Creates stateless OAuth2 Tumblr functions --- src/pardner/stateless/tumblr.py | 72 +++++++++++++++++++++++++++++ tests/test_stateless/test_tumblr.py | 44 ++++++++++++++++++ 2 files changed, 116 insertions(+) create mode 100644 src/pardner/stateless/tumblr.py create mode 100644 tests/test_stateless/test_tumblr.py diff --git a/src/pardner/stateless/tumblr.py b/src/pardner/stateless/tumblr.py new file mode 100644 index 0000000..0f4fc3a --- /dev/null +++ b/src/pardner/stateless/tumblr.py @@ -0,0 +1,72 @@ +from enum import StrEnum +from typing import Any, Iterable, Optional + +from pardner.stateless import Scope +from pardner.stateless.base import ( + generic_construct_authorization_url, + generic_fetch_token, +) +from pardner.verticals import Vertical + + +class URLs(StrEnum): + AuthorizationURL = 'https://www.tumblr.com/oauth2/authorize' + TokenURL = 'https://api.tumblr.com/v2/oauth2/token' + + +def scope_for_verticals(verticals: Iterable[Vertical]) -> set[str]: + # Tumblr only needs 'base' for read access requests + return {'base'} + + +def construct_authorization_url( + client_id: str, redirect_uri: str, scope: Scope = {'base'} +) -> tuple[str, str]: + """ + Builds the authorization URL and state for Tumblr. + + :param client_id: Client identifier given by the OAuth provider upon registration. + :param redirect_uri: The registered callback URI. + :param scope: The scope of the access request. These may be any string but are + commonly URIs or various categories such as ``videos`` or ``documents``. + + :returns: the authorization URL and state, respectively. + """ + return generic_construct_authorization_url( + URLs.AuthorizationURL, client_id, redirect_uri, scope + ) + + +def fetch_token( + client_id: str, + redirect_uri: str, + authorization_response: Optional[str] = None, + client_secret: Optional[str] = None, + code: Optional[str] = None, +) -> dict[str, Any]: + """ + Makes a request to Tumblr's resource server to obtain the access token. + + One of either `code` or `authorization_response` must not be None. + + :param client_id: Client identifier given by the OAuth provider upon registration. + :param redirect_uri: The registered callback URI. + :param scope: The scope of the access request. These may be any string but are + commonly URIs or various categories such as ``videos`` or ``documents``. + :param authorization_response: the URL (with parameters) the end-user's browser + redirected to after authorization. + :param client_secret: The `client_secret` paired to the `client_id`. + :param code: Authorization code (used by WebApplicationClients). + + :returns: the authorization URL and state, respectively. + """ + return generic_fetch_token( + client_id, + redirect_uri, + {'base'}, + URLs.TokenURL, + authorization_response, + client_secret, + code, + include_client_id=True, + ) diff --git a/tests/test_stateless/test_tumblr.py b/tests/test_stateless/test_tumblr.py new file mode 100644 index 0000000..c734183 --- /dev/null +++ b/tests/test_stateless/test_tumblr.py @@ -0,0 +1,44 @@ +from pardner.stateless.tumblr import ( + URLs, + construct_authorization_url, + fetch_token, + scope_for_verticals, +) +from pardner.verticals import Vertical +from tests.conftest import get_url_params + + +def test_scope_for_verticals(): + assert scope_for_verticals({Vertical.FeedPost}) == {'base'} + + +def test_construct_authorization_url(): + auth_url, state = construct_authorization_url( + 'fake_client_id', 'https://redirect_uri' + ) + assert auth_url.startswith(URLs.AuthorizationURL) + + auth_url_params = get_url_params(auth_url) + + assert 'client_id' in auth_url_params + assert auth_url_params['client_id'] == 'fake_client_id' + assert 'redirect_uri' in auth_url_params + assert auth_url_params['redirect_uri'] == 'https://redirect_uri' + assert 'state' in auth_url_params + assert auth_url_params['state'] == state + assert 'scope' in auth_url_params + assert 'base' in auth_url_params['scope'] + + +def test_fetch_token_with_code(mock_outbound_requests): + mock_oauth2session_request, mock_client_parse_request_body_response = ( + mock_outbound_requests + ) + fetch_token( + 'fake_client_id', + 'https://redirect_uri', + client_secret='fake client secret', + code='the_best_code', + ) + mock_oauth2session_request.assert_called_once() + mock_client_parse_request_body_response.assert_called_once() From db0935635f2b91ac4ea2b85f4070751b85c5b509 Mon Sep 17 00:00:00 2001 From: Arthur Borem Date: Thu, 24 Jul 2025 21:19:10 -0500 Subject: [PATCH 2/2] Uses URLs defined in stateless.tumblr in Tumblr class --- src/pardner/services/tumblr.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/pardner/services/tumblr.py b/src/pardner/services/tumblr.py index 8cb48bb..6e4ebb4 100644 --- a/src/pardner/services/tumblr.py +++ b/src/pardner/services/tumblr.py @@ -1,6 +1,7 @@ from typing import Any, Iterable, Optional from pardner.services import BaseTransferService +from pardner.stateless.tumblr import URLs from pardner.verticals import Vertical @@ -11,8 +12,8 @@ class TumblrTransferService(BaseTransferService): See API documentation: https://www.tumblr.com/docs/en/api/v2 """ - _authorization_url = 'https://www.tumblr.com/oauth2/authorize' - _token_url = 'https://api.tumblr.com/v2/oauth2/token' + _authorization_url = URLs.AuthorizationURL + _token_url = URLs.TokenURL def __init__( self,