mirror of
https://github.com/fastapi-users/fastapi-users.git
synced 2025-11-01 10:25:45 +08:00
fix: associate callback test
This commit is contained in:
committed by
François Voron
parent
5863445774
commit
3d33e3da8a
@ -606,20 +606,26 @@ class TestAssociateCallback:
|
|||||||
async def test_callback_token_expired(
|
async def test_callback_token_expired(
|
||||||
self,
|
self,
|
||||||
async_method_mocker: AsyncMethodMocker,
|
async_method_mocker: AsyncMethodMocker,
|
||||||
test_app_client: httpx.AsyncClient,
|
test_app_client_redirect_url: httpx.AsyncClient,
|
||||||
oauth_client: BaseOAuth2,
|
oauth_client: BaseOAuth2,
|
||||||
user_oauth: UserOAuthModel,
|
user_oauth: UserOAuthModel,
|
||||||
user_manager_oauth: UserManagerMock,
|
user_manager_oauth: UserManagerMock,
|
||||||
access_token: str,
|
access_token: str,
|
||||||
):
|
):
|
||||||
state_jwt = generate_state_token({}, "SECRET", lifetime_seconds=-1)
|
state_jwt = generate_state_token({"sub": str(user_oauth.id)}, "SECRET", lifetime_seconds=-1)
|
||||||
async_method_mocker(oauth_client, "get_access_token", return_value=access_token)
|
async_method_mocker(
|
||||||
|
oauth_client, "get_access_token", return_value=access_token
|
||||||
|
)
|
||||||
async_method_mocker(
|
async_method_mocker(
|
||||||
oauth_client, "get_id_email", return_value=("user_oauth1", user_oauth.email)
|
oauth_client, "get_id_email", return_value=("user_oauth1", user_oauth.email)
|
||||||
)
|
)
|
||||||
response = await test_app_client.get(
|
async_method_mocker(
|
||||||
"/oauth/callback",
|
user_manager_oauth, "oauth_callback", return_value=user_oauth
|
||||||
|
)
|
||||||
|
response = await test_app_client_redirect_url.get(
|
||||||
|
"/oauth-associate/callback",
|
||||||
params={"code": "CODE", "state": state_jwt},
|
params={"code": "CODE", "state": state_jwt},
|
||||||
|
headers={"Authorization": f"Bearer {user_oauth.id}"},
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||||
@ -627,33 +633,35 @@ class TestAssociateCallback:
|
|||||||
data = cast(dict[str, Any], response.json())
|
data = cast(dict[str, Any], response.json())
|
||||||
assert data["detail"] == ErrorCode.ACCESS_TOKEN_ALREADY_EXPIRED
|
assert data["detail"] == ErrorCode.ACCESS_TOKEN_ALREADY_EXPIRED
|
||||||
|
|
||||||
assert user_manager_oauth.on_after_login.called is False
|
|
||||||
|
|
||||||
async def test_callback_decode_token_error(
|
async def test_callback_decode_token_error(
|
||||||
self,
|
self,
|
||||||
async_method_mocker: AsyncMethodMocker,
|
async_method_mocker: AsyncMethodMocker,
|
||||||
test_app_client: httpx.AsyncClient,
|
test_app_client_redirect_url: httpx.AsyncClient,
|
||||||
oauth_client: BaseOAuth2,
|
oauth_client: BaseOAuth2,
|
||||||
user_oauth: UserOAuthModel,
|
user_oauth: UserOAuthModel,
|
||||||
user_manager_oauth: UserManagerMock,
|
user_manager_oauth: UserManagerMock,
|
||||||
access_token: str,
|
access_token: str,
|
||||||
):
|
):
|
||||||
state_jwt = generate_state_token({}, "RANDOM")
|
state_jwt = generate_state_token({"sub": str(user_oauth.id)}, "RANDOM")
|
||||||
async_method_mocker(oauth_client, "get_access_token", return_value=access_token)
|
async_method_mocker(
|
||||||
|
oauth_client, "get_access_token", return_value=access_token
|
||||||
|
)
|
||||||
async_method_mocker(
|
async_method_mocker(
|
||||||
oauth_client, "get_id_email", return_value=("user_oauth1", user_oauth.email)
|
oauth_client, "get_id_email", return_value=("user_oauth1", user_oauth.email)
|
||||||
)
|
)
|
||||||
response = await test_app_client.get(
|
async_method_mocker(
|
||||||
"/oauth/callback",
|
user_manager_oauth, "oauth_callback", return_value=user_oauth
|
||||||
|
)
|
||||||
|
response = await test_app_client_redirect_url.get(
|
||||||
|
"/oauth-associate/callback",
|
||||||
params={"code": "CODE", "state": state_jwt},
|
params={"code": "CODE", "state": state_jwt},
|
||||||
|
headers={"Authorization": f"Bearer {user_oauth.id}"},
|
||||||
)
|
)
|
||||||
|
|
||||||
data = cast(dict[str, Any], response.json())
|
data = cast(dict[str, Any], response.json())
|
||||||
assert data["detail"] == ErrorCode.ACCESS_TOKEN_DECODE_ERROR
|
assert data["detail"] == ErrorCode.ACCESS_TOKEN_DECODE_ERROR
|
||||||
|
|
||||||
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
||||||
assert user_manager_oauth.on_after_login.called is False
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
@pytest.mark.oauth
|
@pytest.mark.oauth
|
||||||
|
|||||||
Reference in New Issue
Block a user