From 19d6bd02276e9fd628f026b8aaaaf7cb636815a4 Mon Sep 17 00:00:00 2001 From: Colin-b Date: Sun, 18 Feb 2024 19:04:31 +0100 Subject: [PATCH] Ensure client can be reused in token refresh step as well --- .../test_oauth2_authorization_code_async.py | 67 +++++++++++++++++- .../test_oauth2_authorization_code_sync.py | 66 ++++++++++++++++- ...st_oauth2_authorization_code_pkce_async.py | 65 ++++++++++++++++- ...est_oauth2_authorization_code_pkce_sync.py | 64 ++++++++++++++++- ...uth2_resource_owner_password_okta_async.py | 70 +++++++++++++++++++ ...auth2_resource_owner_password_okta_sync.py | 68 ++++++++++++++++++ ...st_oauth2_resource_owner_password_async.py | 62 ++++++++++++++++ ...est_oauth2_resource_owner_password_sync.py | 60 ++++++++++++++++ 8 files changed, 518 insertions(+), 4 deletions(-) diff --git a/tests/oauth2/authorization_code/test_oauth2_authorization_code_async.py b/tests/oauth2/authorization_code/test_oauth2_authorization_code_async.py index 3d63b29..c70a4d4 100644 --- a/tests/oauth2/authorization_code/test_oauth2_authorization_code_async.py +++ b/tests/oauth2/authorization_code/test_oauth2_authorization_code_async.py @@ -133,7 +133,6 @@ async def test_oauth2_authorization_code_flow_is_able_to_reuse_client( "access_token": "2YotnFZFEjr1zCsicMWpAA", "token_type": "example", "expires_in": 10, - "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", "example_parameter": "example_value", }, match_content=b"grant_type=authorization_code&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F&response_type=code&code=SplxlOBeZQQYbYS6WxSbIA", @@ -164,6 +163,72 @@ async def test_oauth2_authorization_code_flow_is_able_to_reuse_client( tab.assert_success() +@pytest.mark.asyncio +async def test_oauth2_authorization_code_flow_is_able_to_reuse_client_with_token_refresh( + token_cache, httpx_mock: HTTPXMock, browser_mock: BrowserMock +): + client = httpx.Client(headers={"x-test": "Test value"}) + auth = httpx_auth.OAuth2AuthorizationCode( + "https://provide_code", "https://provide_access_token", client=client + ) + tab = browser_mock.add_response( + opened_url="https://provide_code?response_type=code&state=ce9c755b41b5e3c5b64c70598715d5de271023a53f39a67a70215d265d11d2bfb6ef6e9c701701e998e69cbdbf2cee29fd51d2a950aa05f59a20cf4a646099d5&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", + reply_url="http://localhost:5000#code=SplxlOBeZQQYbYS6WxSbIA&state=ce9c755b41b5e3c5b64c70598715d5de271023a53f39a67a70215d265d11d2bfb6ef6e9c701701e998e69cbdbf2cee29fd51d2a950aa05f59a20cf4a646099d5", + ) + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "2YotnFZFEjr1zCsicMWpAA", + "token_type": "example", + "expires_in": 10, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"grant_type=authorization_code&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F&response_type=code&code=SplxlOBeZQQYbYS6WxSbIA", + match_headers={"x-test": "Test value"}, + ) + + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer 2YotnFZFEjr1zCsicMWpAA", + }, + ) + async with httpx.AsyncClient() as client: + await client.get("https://authorized_only", auth=auth) + + tab.assert_success() + + time.sleep(10) + + # response for refresh token grant + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "rVR7Syg5bjZtZYjbZIW", + "token_type": "example", + "expires_in": 3600, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"grant_type=refresh_token&response_type=code&refresh_token=tGzv3JOkF0XG5Qx2TlKWIA", + match_headers={"x-test": "Test value"}, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer rVR7Syg5bjZtZYjbZIW", + }, + ) + + async with httpx.AsyncClient() as client: + await client.get("https://authorized_only", auth=auth) + + @pytest.mark.asyncio async def test_oauth2_authorization_code_flow_get_code_is_sent_in_authorization_header_by_default( token_cache, httpx_mock: HTTPXMock, browser_mock: BrowserMock diff --git a/tests/oauth2/authorization_code/test_oauth2_authorization_code_sync.py b/tests/oauth2/authorization_code/test_oauth2_authorization_code_sync.py index 392c764..f3571b0 100644 --- a/tests/oauth2/authorization_code/test_oauth2_authorization_code_sync.py +++ b/tests/oauth2/authorization_code/test_oauth2_authorization_code_sync.py @@ -129,7 +129,6 @@ def test_oauth2_authorization_code_flow_is_able_to_reuse_client( "access_token": "2YotnFZFEjr1zCsicMWpAA", "token_type": "example", "expires_in": 10, - "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", "example_parameter": "example_value", }, match_content=b"grant_type=authorization_code&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F&response_type=code&code=SplxlOBeZQQYbYS6WxSbIA", @@ -160,6 +159,71 @@ def test_oauth2_authorization_code_flow_is_able_to_reuse_client( tab.assert_success() +def test_oauth2_authorization_code_flow_is_able_to_reuse_client_with_token_refresh( + token_cache, httpx_mock: HTTPXMock, browser_mock: BrowserMock +): + client = httpx.Client(headers={"x-test": "Test value"}) + auth = httpx_auth.OAuth2AuthorizationCode( + "https://provide_code", "https://provide_access_token", client=client + ) + tab = browser_mock.add_response( + opened_url="https://provide_code?response_type=code&state=ce9c755b41b5e3c5b64c70598715d5de271023a53f39a67a70215d265d11d2bfb6ef6e9c701701e998e69cbdbf2cee29fd51d2a950aa05f59a20cf4a646099d5&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F", + reply_url="http://localhost:5000#code=SplxlOBeZQQYbYS6WxSbIA&state=ce9c755b41b5e3c5b64c70598715d5de271023a53f39a67a70215d265d11d2bfb6ef6e9c701701e998e69cbdbf2cee29fd51d2a950aa05f59a20cf4a646099d5", + ) + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "2YotnFZFEjr1zCsicMWpAA", + "token_type": "example", + "expires_in": 10, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"grant_type=authorization_code&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F&response_type=code&code=SplxlOBeZQQYbYS6WxSbIA", + match_headers={"x-test": "Test value"}, + ) + + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer 2YotnFZFEjr1zCsicMWpAA", + }, + ) + with httpx.Client() as client: + client.get("https://authorized_only", auth=auth) + + tab.assert_success() + + time.sleep(10) + + # response for refresh token grant + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "rVR7Syg5bjZtZYjbZIW", + "token_type": "example", + "expires_in": 3600, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"grant_type=refresh_token&response_type=code&refresh_token=tGzv3JOkF0XG5Qx2TlKWIA", + match_headers={"x-test": "Test value"}, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer rVR7Syg5bjZtZYjbZIW", + }, + ) + + with httpx.Client() as client: + client.get("https://authorized_only", auth=auth) + + def test_oauth2_authorization_code_flow_get_code_is_sent_in_authorization_header_by_default( token_cache, httpx_mock: HTTPXMock, browser_mock: BrowserMock ): diff --git a/tests/oauth2/authorization_code_pkce/test_oauth2_authorization_code_pkce_async.py b/tests/oauth2/authorization_code_pkce/test_oauth2_authorization_code_pkce_async.py index 00ea406..28fd7e6 100644 --- a/tests/oauth2/authorization_code_pkce/test_oauth2_authorization_code_pkce_async.py +++ b/tests/oauth2/authorization_code_pkce/test_oauth2_authorization_code_pkce_async.py @@ -131,7 +131,6 @@ async def test_oauth2_pkce_flow_is_able_to_reuse_client( "access_token": "2YotnFZFEjr1zCsicMWpAA", "token_type": "example", "expires_in": 10, - "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", "example_parameter": "example_value", }, match_content=b"code_verifier=MTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTEx&grant_type=authorization_code&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F&response_type=code&code=SplxlOBeZQQYbYS6WxSbIA", @@ -161,6 +160,70 @@ async def test_oauth2_pkce_flow_is_able_to_reuse_client( tab.assert_success() +@pytest.mark.asyncio +async def test_oauth2_pkce_flow_is_able_to_reuse_client_with_token_refresh( + token_cache, httpx_mock: HTTPXMock, browser_mock: BrowserMock +): + client = httpx.Client(headers={"x-test": "Test value"}) + auth = httpx_auth.OAuth2AuthorizationCodePKCE( + "https://provide_code", "https://provide_access_token", client=client + ) + tab = browser_mock.add_response( + opened_url="https://provide_code?response_type=code&state=ce9c755b41b5e3c5b64c70598715d5de271023a53f39a67a70215d265d11d2bfb6ef6e9c701701e998e69cbdbf2cee29fd51d2a950aa05f59a20cf4a646099d5&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F&code_challenge=5C_ph_KZ3DstYUc965SiqmKAA-ShvKF4Ut7daKd3fjc&code_challenge_method=S256", + reply_url="http://localhost:5000#code=SplxlOBeZQQYbYS6WxSbIA&state=ce9c755b41b5e3c5b64c70598715d5de271023a53f39a67a70215d265d11d2bfb6ef6e9c701701e998e69cbdbf2cee29fd51d2a950aa05f59a20cf4a646099d5", + ) + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "2YotnFZFEjr1zCsicMWpAA", + "token_type": "example", + "expires_in": 10, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"code_verifier=MTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTEx&grant_type=authorization_code&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F&response_type=code&code=SplxlOBeZQQYbYS6WxSbIA", + match_headers={"x-test": "Test value"}, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer 2YotnFZFEjr1zCsicMWpAA", + }, + ) + + async with httpx.AsyncClient() as client: + await client.get("https://authorized_only", auth=auth) + + tab.assert_success() + time.sleep(10) + + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "rVR7Syg5bjZtZYjbZIW", + "token_type": "example", + "expires_in": 3600, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"grant_type=refresh_token&response_type=code&refresh_token=tGzv3JOkF0XG5Qx2TlKWIA", + match_headers={"x-test": "Test value"}, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer rVR7Syg5bjZtZYjbZIW", + }, + ) + + async with httpx.AsyncClient() as client: + await client.get("https://authorized_only", auth=auth) + + @pytest.mark.asyncio async def test_oauth2_pkce_flow_get_code_is_sent_in_authorization_header_by_default( token_cache, httpx_mock: HTTPXMock, browser_mock: BrowserMock diff --git a/tests/oauth2/authorization_code_pkce/test_oauth2_authorization_code_pkce_sync.py b/tests/oauth2/authorization_code_pkce/test_oauth2_authorization_code_pkce_sync.py index ea1f827..ceda680 100644 --- a/tests/oauth2/authorization_code_pkce/test_oauth2_authorization_code_pkce_sync.py +++ b/tests/oauth2/authorization_code_pkce/test_oauth2_authorization_code_pkce_sync.py @@ -127,7 +127,6 @@ def test_oauth2_pkce_flow_is_able_to_reuse_client( "access_token": "2YotnFZFEjr1zCsicMWpAA", "token_type": "example", "expires_in": 10, - "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", "example_parameter": "example_value", }, match_content=b"code_verifier=MTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTEx&grant_type=authorization_code&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F&response_type=code&code=SplxlOBeZQQYbYS6WxSbIA", @@ -157,6 +156,69 @@ def test_oauth2_pkce_flow_is_able_to_reuse_client( tab.assert_success() +def test_oauth2_pkce_flow_is_able_to_reuse_client_with_token_refresh( + token_cache, httpx_mock: HTTPXMock, browser_mock: BrowserMock +): + client = httpx.Client(headers={"x-test": "Test value"}) + auth = httpx_auth.OAuth2AuthorizationCodePKCE( + "https://provide_code", "https://provide_access_token", client=client + ) + tab = browser_mock.add_response( + opened_url="https://provide_code?response_type=code&state=ce9c755b41b5e3c5b64c70598715d5de271023a53f39a67a70215d265d11d2bfb6ef6e9c701701e998e69cbdbf2cee29fd51d2a950aa05f59a20cf4a646099d5&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F&code_challenge=5C_ph_KZ3DstYUc965SiqmKAA-ShvKF4Ut7daKd3fjc&code_challenge_method=S256", + reply_url="http://localhost:5000#code=SplxlOBeZQQYbYS6WxSbIA&state=ce9c755b41b5e3c5b64c70598715d5de271023a53f39a67a70215d265d11d2bfb6ef6e9c701701e998e69cbdbf2cee29fd51d2a950aa05f59a20cf4a646099d5", + ) + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "2YotnFZFEjr1zCsicMWpAA", + "token_type": "example", + "expires_in": 10, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"code_verifier=MTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTExMTEx&grant_type=authorization_code&redirect_uri=http%3A%2F%2Flocalhost%3A5000%2F&response_type=code&code=SplxlOBeZQQYbYS6WxSbIA", + match_headers={"x-test": "Test value"}, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer 2YotnFZFEjr1zCsicMWpAA", + }, + ) + + with httpx.Client() as client: + client.get("https://authorized_only", auth=auth) + + tab.assert_success() + time.sleep(10) + + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "rVR7Syg5bjZtZYjbZIW", + "token_type": "example", + "expires_in": 3600, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"grant_type=refresh_token&response_type=code&refresh_token=tGzv3JOkF0XG5Qx2TlKWIA", + match_headers={"x-test": "Test value"}, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer rVR7Syg5bjZtZYjbZIW", + }, + ) + + with httpx.Client() as client: + client.get("https://authorized_only", auth=auth) + + def test_oauth2_pkce_flow_get_code_is_sent_in_authorization_header_by_default( token_cache, httpx_mock: HTTPXMock, browser_mock: BrowserMock ): diff --git a/tests/oauth2/resource_owner_password/okta/test_oauth2_resource_owner_password_okta_async.py b/tests/oauth2/resource_owner_password/okta/test_oauth2_resource_owner_password_okta_async.py index 3355f3f..1351fd6 100644 --- a/tests/oauth2/resource_owner_password/okta/test_oauth2_resource_owner_password_okta_async.py +++ b/tests/oauth2/resource_owner_password/okta/test_oauth2_resource_owner_password_okta_async.py @@ -54,6 +54,52 @@ async def test_oauth2_password_credentials_flow_uses_provided_client( @pytest.mark.asyncio async def test_oauth2_password_credentials_flow_is_able_to_reuse_client( token_cache, httpx_mock: HTTPXMock +): + # TODO Add support for AsyncClient + client = httpx.Client(headers={"x-test": "Test value"}) + auth = httpx_auth.OktaResourceOwnerPasswordCredentials( + "testserver.okta-emea.com", + username="test_user", + password="test_pwd", + client_id="test_user2", + client_secret="test_pwd2", + client=client, + ) + httpx_mock.add_response( + method="POST", + url="https://testserver.okta-emea.com/oauth2/default/v1/token", + json={ + "access_token": "2YotnFZFEjr1zCsicMWpAA", + "token_type": "example", + "expires_in": 10, + "example_parameter": "example_value", + }, + match_content=b"grant_type=password&username=test_user&password=test_pwd&scope=openid", + match_headers={ + "x-test": "Test value", + "Authorization": "Basic dGVzdF91c2VyMjp0ZXN0X3B3ZDI=", + }, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer 2YotnFZFEjr1zCsicMWpAA", + }, + ) + + async with httpx.AsyncClient() as client: + await client.get("https://authorized_only", auth=auth) + + time.sleep(10) + + async with httpx.AsyncClient() as client: + await client.get("https://authorized_only", auth=auth) + + +@pytest.mark.asyncio +async def test_oauth2_password_credentials_flow_is_able_to_reuse_client_with_token_refresh( + token_cache, httpx_mock: HTTPXMock ): # TODO Add support for AsyncClient client = httpx.Client(headers={"x-test": "Test value"}) @@ -94,6 +140,30 @@ async def test_oauth2_password_credentials_flow_is_able_to_reuse_client( time.sleep(10) + httpx_mock.add_response( + method="POST", + url="https://testserver.okta-emea.com/oauth2/default/v1/token", + json={ + "access_token": "rVR7Syg5bjZtZYjbZIW", + "token_type": "example", + "expires_in": 3600, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"grant_type=refresh_token&scope=openid&refresh_token=tGzv3JOkF0XG5Qx2TlKWIA", + match_headers={ + "x-test": "Test value", + "Authorization": "Basic dGVzdF91c2VyMjp0ZXN0X3B3ZDI=", + }, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer rVR7Syg5bjZtZYjbZIW", + }, + ) + async with httpx.AsyncClient() as client: await client.get("https://authorized_only", auth=auth) diff --git a/tests/oauth2/resource_owner_password/okta/test_oauth2_resource_owner_password_okta_sync.py b/tests/oauth2/resource_owner_password/okta/test_oauth2_resource_owner_password_okta_sync.py index 8a55c30..dc5ddf4 100644 --- a/tests/oauth2/resource_owner_password/okta/test_oauth2_resource_owner_password_okta_sync.py +++ b/tests/oauth2/resource_owner_password/okta/test_oauth2_resource_owner_password_okta_sync.py @@ -51,6 +51,50 @@ def test_oauth2_password_credentials_flow_uses_provided_client( def test_oauth2_password_credentials_flow_is_able_to_reuse_client( token_cache, httpx_mock: HTTPXMock +): + client = httpx.Client(headers={"x-test": "Test value"}) + auth = httpx_auth.OktaResourceOwnerPasswordCredentials( + "testserver.okta-emea.com", + username="test_user", + password="test_pwd", + client_id="test_user2", + client_secret="test_pwd2", + client=client, + ) + httpx_mock.add_response( + method="POST", + url="https://testserver.okta-emea.com/oauth2/default/v1/token", + json={ + "access_token": "2YotnFZFEjr1zCsicMWpAA", + "token_type": "example", + "expires_in": 10, + "example_parameter": "example_value", + }, + match_content=b"grant_type=password&username=test_user&password=test_pwd&scope=openid", + match_headers={ + "x-test": "Test value", + "Authorization": "Basic dGVzdF91c2VyMjp0ZXN0X3B3ZDI=", + }, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer 2YotnFZFEjr1zCsicMWpAA", + }, + ) + + with httpx.Client() as client: + client.get("https://authorized_only", auth=auth) + + time.sleep(10) + + with httpx.Client() as client: + client.get("https://authorized_only", auth=auth) + + +def test_oauth2_password_credentials_flow_is_able_to_reuse_client_with_token_refresh( + token_cache, httpx_mock: HTTPXMock ): client = httpx.Client(headers={"x-test": "Test value"}) auth = httpx_auth.OktaResourceOwnerPasswordCredentials( @@ -90,6 +134,30 @@ def test_oauth2_password_credentials_flow_is_able_to_reuse_client( time.sleep(10) + httpx_mock.add_response( + method="POST", + url="https://testserver.okta-emea.com/oauth2/default/v1/token", + json={ + "access_token": "rVR7Syg5bjZtZYjbZIW", + "token_type": "example", + "expires_in": 3600, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"grant_type=refresh_token&scope=openid&refresh_token=tGzv3JOkF0XG5Qx2TlKWIA", + match_headers={ + "x-test": "Test value", + "Authorization": "Basic dGVzdF91c2VyMjp0ZXN0X3B3ZDI=", + }, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer rVR7Syg5bjZtZYjbZIW", + }, + ) + with httpx.Client() as client: client.get("https://authorized_only", auth=auth) diff --git a/tests/oauth2/resource_owner_password/test_oauth2_resource_owner_password_async.py b/tests/oauth2/resource_owner_password/test_oauth2_resource_owner_password_async.py index 7fb53e0..6c1fedc 100644 --- a/tests/oauth2/resource_owner_password/test_oauth2_resource_owner_password_async.py +++ b/tests/oauth2/resource_owner_password/test_oauth2_resource_owner_password_async.py @@ -49,6 +49,47 @@ async def test_oauth2_password_credentials_flow_uses_provided_client( @pytest.mark.asyncio async def test_oauth2_password_credentials_flow_is_able_to_reuse_client( token_cache, httpx_mock: HTTPXMock +): + # TODO Add support for AsyncClient + client = httpx.Client(headers={"x-test": "Test value"}) + auth = httpx_auth.OAuth2ResourceOwnerPasswordCredentials( + "https://provide_access_token", + username="test_user", + password="test_pwd", + client=client, + ) + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "2YotnFZFEjr1zCsicMWpAA", + "token_type": "example", + "expires_in": 10, + "example_parameter": "example_value", + }, + match_content=b"grant_type=password&username=test_user&password=test_pwd", + match_headers={"x-test": "Test value"}, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer 2YotnFZFEjr1zCsicMWpAA", + }, + ) + + async with httpx.AsyncClient() as client: + await client.get("https://authorized_only", auth=auth) + + time.sleep(10) + + async with httpx.AsyncClient() as client: + await client.get("https://authorized_only", auth=auth) + + +@pytest.mark.asyncio +async def test_oauth2_password_credentials_flow_is_able_to_reuse_client_with_token_refresh( + token_cache, httpx_mock: HTTPXMock ): # TODO Add support for AsyncClient client = httpx.Client(headers={"x-test": "Test value"}) @@ -84,6 +125,27 @@ async def test_oauth2_password_credentials_flow_is_able_to_reuse_client( time.sleep(10) + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "rVR7Syg5bjZtZYjbZIW", + "token_type": "example", + "expires_in": 3600, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"grant_type=refresh_token&refresh_token=tGzv3JOkF0XG5Qx2TlKWIA", + match_headers={"x-test": "Test value"}, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer rVR7Syg5bjZtZYjbZIW", + }, + ) + async with httpx.AsyncClient() as client: await client.get("https://authorized_only", auth=auth) diff --git a/tests/oauth2/resource_owner_password/test_oauth2_resource_owner_password_sync.py b/tests/oauth2/resource_owner_password/test_oauth2_resource_owner_password_sync.py index cf41815..aafc22e 100644 --- a/tests/oauth2/resource_owner_password/test_oauth2_resource_owner_password_sync.py +++ b/tests/oauth2/resource_owner_password/test_oauth2_resource_owner_password_sync.py @@ -46,6 +46,45 @@ def test_oauth2_password_credentials_flow_uses_provided_client( def test_oauth2_password_credentials_flow_is_able_to_reuse_client( token_cache, httpx_mock: HTTPXMock +): + client = httpx.Client(headers={"x-test": "Test value"}) + auth = httpx_auth.OAuth2ResourceOwnerPasswordCredentials( + "https://provide_access_token", + username="test_user", + password="test_pwd", + client=client, + ) + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "2YotnFZFEjr1zCsicMWpAA", + "token_type": "example", + "expires_in": 10, + "example_parameter": "example_value", + }, + match_content=b"grant_type=password&username=test_user&password=test_pwd", + match_headers={"x-test": "Test value"}, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer 2YotnFZFEjr1zCsicMWpAA", + }, + ) + + with httpx.Client() as client: + client.get("https://authorized_only", auth=auth) + + time.sleep(10) + + with httpx.Client() as client: + client.get("https://authorized_only", auth=auth) + + +def test_oauth2_password_credentials_flow_is_able_to_reuse_client_with_token_refresh( + token_cache, httpx_mock: HTTPXMock ): client = httpx.Client(headers={"x-test": "Test value"}) auth = httpx_auth.OAuth2ResourceOwnerPasswordCredentials( @@ -80,6 +119,27 @@ def test_oauth2_password_credentials_flow_is_able_to_reuse_client( time.sleep(10) + httpx_mock.add_response( + method="POST", + url="https://provide_access_token", + json={ + "access_token": "rVR7Syg5bjZtZYjbZIW", + "token_type": "example", + "expires_in": 3600, + "refresh_token": "tGzv3JOkF0XG5Qx2TlKWIA", + "example_parameter": "example_value", + }, + match_content=b"grant_type=refresh_token&refresh_token=tGzv3JOkF0XG5Qx2TlKWIA", + match_headers={"x-test": "Test value"}, + ) + httpx_mock.add_response( + url="https://authorized_only", + method="GET", + match_headers={ + "Authorization": "Bearer rVR7Syg5bjZtZYjbZIW", + }, + ) + with httpx.Client() as client: client.get("https://authorized_only", auth=auth)