Skip to content

Commit

Permalink
Ensure client can be reused in token refresh step as well
Browse files Browse the repository at this point in the history
  • Loading branch information
Colin-b committed Feb 18, 2024
1 parent 4ef0854 commit 19d6bd0
Show file tree
Hide file tree
Showing 8 changed files with 518 additions and 4 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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
):
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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
):
Expand Down
Loading

0 comments on commit 19d6bd0

Please sign in to comment.