Repository: hikari-py/hikari
Pull request #627 (2021-05-30 17:24:16)

Add a token type property to the rest interface to allow for token introspection
### Summary
Add a token type property to the rest interface to allow for token introspection

### Checklist
<!-- Make sure to tick all the following boxes by putting an `x` in between (like this `[x]`) -->
- [x] I have run `nox` and all the pipelines have passed.
- [x] I have made unittests according to the code I have added/modified/deleted.

### Related issues
<!--
To mention an issue use `#issue-id` and to mention a merge request use `!merge-request-id`
To close/fix an issue use `Close #issue-id` or `Fix #issue-id` (depending on the merge request)
-->


davfsa: I'm just going to leave this here as we were discussing it in the server.

Would exposing the token type only be good enough? You can act on that information, no need to expose the whole token
FasterSpeeding: I'll refactor this to only return the "token type" at one point, will probably have it return Bearer if the token type is a strategy or otherwise will have to add a token type property to token strategy which this relays but will have to spend a bit more time to consider this

Patch:
diff --git a/hikari/api/rest.py b/hikari/api/rest.py
index 27131be8..34f53d53 100644
--- a/hikari/api/rest.py
+++ b/hikari/api/rest.py
@@ -61,9 +61,20 @@ class TokenStrategy(abc.ABC):
 
     __slots__: typing.Sequence[str] = ()
 
+    @property
+    @abc.abstractmethod
+    def token_type(self) -> typing.Union[applications.TokenType, str]:
+        """Type of token this strategy returns.
+
+        Returns
+        -------
+        typing.Union[hikari.applications.TokenType, builtins.str]
+            The type of token this strategy returns.
+        """
+
     @abc.abstractmethod
     async def acquire(self, client: RESTClient) -> str:
-        """Acquire an authorization token (with the prefix).
+        """Acquire an authorization token (including the prefix).
 
         Returns
         -------
@@ -98,6 +109,20 @@ class RESTClient(traits.NetworkSettingsAware, abc.ABC):
 
     __slots__: typing.Sequence[str] = ()
 
+    @property
+    @abc.abstractmethod
+    def token_type(self) -> typing.Union[str, applications.TokenType, None]:
+        """Type of token this client is using for most requests.
+
+        Returns
+        -------
+        typing.Union[builtins.str, hikari.applications.TokenType, builtins.None]
+            The type of token this client is using for most requests.
+
+            If this is `builtins.None` then this client will likely only work
+            for some endpoints such as public and webhook ones.
+        """
+
     @abc.abstractmethod
     async def close(self) -> None:
         """Close the client session."""
diff --git a/hikari/impl/rest.py b/hikari/impl/rest.py
index a41e4ddf..77a01f4a 100644
--- a/hikari/impl/rest.py
+++ b/hikari/impl/rest.py
@@ -187,6 +187,10 @@ def scopes(self) -> typing.Sequence[typing.Union[applications.OAuth2Scope, str]]
         """
         return self._scopes
 
+    @property
+    def token_type(self) -> applications.TokenType:
+        return applications.TokenType.BEARER
+
     async def acquire(self, client: rest_api.RESTClient) -> str:
         if self._token and not self._is_expired:
             return self._token
@@ -331,17 +335,74 @@ def http_settings(self) -> config.HTTPSettings:
     def proxy_settings(self) -> config.ProxySettings:
         return self._proxy_settings
 
+    @typing.overload
+    def acquire(self, token: typing.Optional[rest_api.TokenStrategy] = None) -> RESTClientImpl:
+        ...
+
+    @typing.overload
     def acquire(
         self,
-        token: typing.Union[str, rest_api.TokenStrategy, None] = None,
+        token: str,
         token_type: typing.Union[str, applications.TokenType] = applications.TokenType.BEARER,
     ) -> RESTClientImpl:
+        ...
+
+    def acquire(
+        self,
+        token: typing.Union[str, rest_api.TokenStrategy, None] = None,
+        token_type: typing.Union[str, applications.TokenType, None] = None,
+    ) -> RESTClientImpl:
+        """Acquire an instance of this REST client.
+
+        !!! note
+            The returned REST client should be started before it can be used,
+            either by calling `RESTClientImpl.start` or by using it as an
+            asynchronous context manager.
+
+        Examples
+        --------
+        ```py
+        rest_app = RESTApp()
+
+        # Using the returned client as a context manager to implicitly start
+        # and stop it.
+        async with rest_app.acquire("A token", "Bot") as client:
+            user = await client.fetch_my_user()
+        ```
+
+        Parameters
+        ----------
+        token : typing.Union[builtins.str, builtins.None, hikari.api.rest.TokenStrategy]
+            The bot or bearer token. If no token is to be used,
+            this can be undefined.
+        token_type : typing.Union[builtins.str, hikari.applications.TokenType, builtins.None]
+            The type of token in use. This should only be passed when `builtins.str`
+            is passed for `token`, can be `"Bot"` or `"Bearer"` and will be
+            defaulted to `"Bearer"` in this situation.
+
+            This should be left as `builtins.None` when either
+            `hikari.api.rest.TokenStrategy` or `builtins.None` is passed for
+            `token`.
+
+        Returns
+        -------
+        RESTClientImpl
+            An instance of the REST client.
+
+        Raises
+        ------
+        builtins.ValueError
+            If `token_type` is provided when a token strategy is passed for `token`.
+        """
         # Since we essentially mimic a fake App instance, we need to make a circular provider.
         # We can achieve this using a lambda. This allows the entity factory to build models that
         # are also REST-aware
         provider = _RESTProvider(lambda: entity_factory, self._executor, lambda: rest_client)
         entity_factory = entity_factory_impl.EntityFactoryImpl(provider)
 
+        if token_type is None and isinstance(token, str):
+            token_type = applications.TokenType.BEARER
+
         rest_client = RESTClientImpl(
             cache=None,
             entity_factory=entity_factory,
@@ -382,11 +443,21 @@ class RESTClientImpl(rest_api.RESTClient):
         The bot or bearer token. If no token is to be used,
         this can be undefined.
     token_type : typing.Union[builtins.str, hikari.applications.TokenType, builtins.None]
-        The type of token in use. If no token is used, this can be ignored and
-        left to the default value. This can be `"Bot"` or `"Bearer"`.
+        The type of token in use. This must be passed when a `builtins.str` is
+        passed for `token` but and can be `"Bot"` or `"Bearer"`.
+
+        This should be left as `builtins.None` when either
+        `hikari.api.rest.TokenStrategy` or `builtins.None` is passed for
+        `token`.
     rest_url : builtins.str
         The HTTP API base URL. This can contain format-string specifiers to
         interpolate information such as API version in use.
+
+    Raises
+    ------
+    builtins.ValueError
+        * If `token_type` is provided when a token strategy is passed for `token`.
+        * if `token_type` is left as `builtins.None` when a string is passed for `token`.
     """
 
     __slots__: typing.Sequence[str] = (
@@ -402,6 +473,7 @@ class RESTClientImpl(rest_api.RESTClient):
         "_proxy_settings",
         "_rest_url",
         "_token",
+        "_token_type",
     )
 
     buckets: buckets_.RESTBucketManager
@@ -440,15 +512,21 @@ def __init__(
         self._http_settings = http_settings
         self._proxy_settings = proxy_settings
 
-        self._token: typing.Union[str, rest_api.TokenStrategy, None]
+        self._token: typing.Union[str, rest_api.TokenStrategy, None] = None
+        self._token_type: typing.Optional[str] = None
         if isinstance(token, str):
-            self._token = f"{token_type.title()} {token}" if token_type else token
+            if token_type is None:
+                raise ValueError("Token type required when a str is passed for `token`")
 
-        elif isinstance(token, rest_api.TokenStrategy) and token_type:
-            raise ValueError("Token type should be handled by the token strategy")
+            self._token = f"{token_type.title()} {token}"
+            self._token_type = applications.TokenType(token_type.title())
+
+        elif isinstance(token, rest_api.TokenStrategy):
+            if token_type is not None:
+                raise ValueError("Token type should be handled by the token strategy")
 
-        else:
             self._token = token
+            self._token_type = token.token_type
 
         self._rest_url = rest_url if rest_url is not None else urls.REST_API_URL
 
@@ -460,6 +538,10 @@ def http_settings(self) -> config.HTTPSettings:
     def proxy_settings(self) -> config.ProxySettings:
         return self._proxy_settings
 
+    @property
+    def token_type(self) -> typing.Union[str, applications.TokenType, None]:
+        return self._token_type
+
     @typing.final
     async def close(self) -> None:
         """Close the HTTP client and any open HTTP connections."""


Test patch:
diff --git a/tests/hikari/impl/test_rest.py b/tests/hikari/impl/test_rest.py
index 1f50c1b3..9e23d8c5 100644
--- a/tests/hikari/impl/test_rest.py
+++ b/tests/hikari/impl/test_rest.py
@@ -123,6 +123,11 @@ def test_scopes_property(self):
 
         assert token.scopes == (123, 5643)
 
+    def test_token_type_property(self):
+        token = rest.ClientCredentialsStrategy(client=123, client_secret="123123123", scopes=[])
+
+        assert token.token_type is applications.TokenType.BEARER
+
     async def test_acquire_on_new_instance(self, mock_token):
         mock_rest = mock.Mock(authorize_client_credentials_token=mock.AsyncMock(return_value=mock_token))
 
@@ -298,6 +303,30 @@ def test_acquire(self, rest_app):
             rest_url=rest_app._url,
         )
 
+    def test_acquire_defaults_to_bearer_for_a_string_token(self, rest_app):
+        mock_event_loop = object()
+        rest_app._event_loop = mock_event_loop
+
+        stack = contextlib.ExitStack()
+        _entity_factory = stack.enter_context(mock.patch.object(entity_factory, "EntityFactoryImpl"))
+        mock_client = stack.enter_context(mock.patch.object(rest, rest.RESTClientImpl.__qualname__))
+        stack.enter_context(mock.patch.object(asyncio, "get_running_loop", return_value=mock_event_loop))
+
+        with stack:
+            rest_app.acquire(token="token")
+
+        mock_client.assert_called_once_with(
+            cache=None,
+            entity_factory=_entity_factory(),
+            executor=rest_app._executor,
+            http_settings=rest_app._http_settings,
+            max_rate_limit=float("inf"),
+            proxy_settings=rest_app._proxy_settings,
+            token="token",
+            token_type=applications.TokenType.BEARER,
+            rest_url=rest_app._url,
+        )
+
 
 ##################
 # RESTClientImpl #
@@ -410,6 +439,23 @@ def test__init__when_token_strategy_passed_with_token_type(self):
                 entity_factory=None,
             )
 
+    def test__init__when_token_strategy_passed(self):
+        mock_strategy = mock.Mock(rest_api.TokenStrategy)
+        obj = rest.RESTClientImpl(
+            cache=None,
+            http_settings=mock.Mock(),
+            max_rate_limit=float("inf"),
+            proxy_settings=mock.Mock(),
+            token=mock_strategy,
+            token_type=None,
+            rest_url=None,
+            executor=None,
+            entity_factory=None,
+        )
+
+        assert obj._token is mock_strategy
+        assert obj._token_type is mock_strategy.token_type
+
     def test__init__when_token_is_None_sets_token_to_None(self):
         obj = rest.RESTClientImpl(
             cache=None,
@@ -423,6 +469,7 @@ def test__init__when_token_is_None_sets_token_to_None(self):
             entity_factory=None,
         )
         assert obj._token is None
+        assert obj._token_type is None
 
     def test__init__when_token_and_token_type_is_not_None_generates_token_with_type(self):
         obj = rest.RESTClientImpl(
@@ -437,6 +484,21 @@ def test__init__when_token_and_token_type_is_not_None_generates_token_with_type(
             entity_factory=None,
         )
         assert obj._token == "Type some_token"
+        assert obj._token_type == "Type"
+
+    def test__init__when_token_provided_as_string_without_type(self):
+        with pytest.raises(ValueError, match="Token type required when a str is passed for `token`"):
+            rest.RESTClientImpl(
+                cache=None,
+                http_settings=mock.Mock(),
+                max_rate_limit=float("inf"),
+                proxy_settings=mock.Mock(),
+                token="some_token",
+                token_type=None,
+                rest_url=None,
+                executor=None,
+                entity_factory=None,
+            )
 
     def test__init__when_rest_url_is_None_generates_url_using_default_url(self):
         obj = rest.RESTClientImpl(
@@ -482,11 +544,16 @@ def test_http_settings_property(self, rest_client):
         rest_client._http_settings = mock_http_settings
         assert rest_client.http_settings is mock_http_settings
 
-    def test_proxy_settings(self, rest_client):
+    def test_proxy_settings_property(self, rest_client):
         mock_proxy_settings = object()
         rest_client._proxy_settings = mock_proxy_settings
         assert rest_client.proxy_settings is mock_proxy_settings
 
+    def test_token_type_property(self, rest_client):
+        mock_type = object()
+        rest_client._token_type = mock_type
+        assert rest_client.token_type is mock_type
+
     def test__acquire_client_session_when_None(self, rest_client):
         client_session_mock = object()
         rest_client._acquire_tcp_connector = mock.Mock()