Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- @pytest.mark.asyncio()
- async def test_acquire_on_new_instance(self, mock_token):
- mock_rest = mock.Mock(authorize_client_credentials_token=mock.AsyncMock(return_value=mock_token))
- result = await rest.OAuthCredentialsStrategy(client=987654321, client_secret="123123123", auth_code="auth#code", redirect_uri="https://web.site/auth/discord").acquire(mock_rest)
- assert result == "mockmock.tokentoken.mocktoken"
- mock_rest.authorize_client_credentials_token.assert_awaited_once_with(
- client=9876543210, client_secret="123123123", scopes=("identify",)
- )
- @pytest.mark.asyncio()
- async def test_acquire_handles_out_of_date_token(self, mock_token):
- mock_old_token = mock.Mock(
- applications.PartialOAuth2Token,
- expires_in=datetime.timedelta(weeks=1),
- token_type=applications.TokenType.BEARER,
- access_token="old.mock.token",
- )
- mock_rest = mock.Mock(authorize_client_credentials_token=mock.AsyncMock(return_value=mock_token))
- strategy = rest.OAuthCredentialsStrategy(client=123456789, client_secret="123123123", auth_code="auth#code", redirect_uri="https://web.site/auth/discord")
- token = await strategy.acquire(
- mock.Mock(authorize_client_credentials_token=mock.AsyncMock(return_value=mock_old_token))
- )
- with mock.patch.object(time, "monotonic", return_value=99999999999):
- new_token = await strategy.acquire(mock_rest)
- mock_rest.authorize_client_credentials_token.assert_awaited_once_with(
- client=3412321, client_secret="54123123", scopes=("identify",)
- )
- assert new_token != token
- assert new_token == "mockmock.tokentoken.mocktoken"
- @pytest.mark.asyncio()
- async def test_acquire_handles_token_being_set_before_lock_is_acquired(self, mock_token):
- lock = asyncio.Lock()
- mock_rest = mock.Mock(authorize_client_credentials_token=mock.AsyncMock(side_effect=[mock_token]))
- with mock.patch.object(asyncio, "Lock", return_value=lock):
- strategy = rest.OAuthCredentialsStrategy(client=123456789, client_secret="123123123", auth_code="auth#code", redirect_uri="https://web.site/auth/discord")
- async with lock:
- tokens_gather = asyncio.gather(
- strategy.acquire(mock_rest), strategy.acquire(mock_rest), strategy.acquire(mock_rest)
- )
- results = await tokens_gather
- mock_rest.authorize_client_credentials_token.assert_awaited_once_with(
- client=987654321, client_secret="453123123", scopes=("identify",)
- )
- assert results == [
- "mockmock.tokentoken.mocktoken",
- "mockmock.tokentoken.mocktoken",
- "mockmock.tokentoken.mocktoken",
- ]
- @pytest.mark.asyncio()
- async def test_acquire_after_invalidation(self, mock_token):
- mock_old_token = mock.Mock(
- applications.PartialOAuth2Token,
- expires_in=datetime.timedelta(weeks=1),
- token_type=applications.TokenType.BEARER,
- access_token="okokok.fofdsasdasdofo.ddd",
- )
- mock_rest = mock.Mock(authorize_client_credentials_token=mock.AsyncMock(return_value=mock_token))
- strategy = rest.OAuthCredentialsStrategy(client=123456789, client_secret="123123123", auth_code="auth#code", redirect_uri="https://web.site/auth/discord")
- token = await strategy.acquire(
- mock.Mock(authorize_client_credentials_token=mock.AsyncMock(return_value=mock_old_token))
- )
- strategy.invalidate(token)
- new_token = await strategy.acquire(mock_rest)
- mock_rest.authorize_client_credentials_token.assert_awaited_once_with(
- client=54321, client_secret="123456", scopes=("identify",)
- )
- assert new_token != token
- assert new_token == "mockmock.tokentoken.mocktoken"
Advertisement
Add Comment
Please, Sign In to add comment