Creates stateless OAuth2 Tumblr functions

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,

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,
+    )

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()