1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
|
from unittest import mock
import pytest
import globus_sdk
REFRESH_TOKEN = "refresh_token_1"
ACCESS_TOKEN = "access_token_1"
EXPIRES_AT = -1
@pytest.fixture(params=["simple", "with_new_refresh_token"])
def response(request):
r = mock.Mock()
r.by_resource_server = {
"simple": {"rs1": {"expires_at_seconds": -1, "access_token": "access_token_2"}},
"with_new_refresh_token": {
"rs1": {
"expires_at_seconds": -1,
"access_token": "access_token_2",
"refresh_token": "refresh_token_2",
}
},
}[request.param]
return r
@pytest.fixture
def client(response):
c = mock.Mock()
c.oauth2_refresh_token = mock.Mock(return_value=response)
return c
@pytest.fixture
def authorizer(client):
return globus_sdk.RefreshTokenAuthorizer(
REFRESH_TOKEN, client, access_token=ACCESS_TOKEN, expires_at=EXPIRES_AT
)
def test_get_token_response(authorizer, client, response):
"""
Calls _get_token_response, confirms that the mock
AuthClient is used and the known data was returned.
"""
# get new_access_token
res = authorizer._get_token_response()
assert res == response
# confirm mock ConfidentialAppAuthClient was used as expected
client.oauth2_refresh_token.assert_called_once_with(REFRESH_TOKEN)
def test_multiple_resource_servers(authorizer, response):
"""
Sets the mock client to return multiple resource servers.
Confirms GlobusError is raised when _extract_token_data is called.
"""
response.by_resource_server["rs2"] = {
"expires_at_seconds": -1,
"access_token": "access_token_3",
}
with pytest.raises(ValueError) as excinfo:
authorizer._extract_token_data(response)
assert "didn't return exactly one token" in str(excinfo.value)
def test_conditional_refresh_token_update(authorizer, response):
"""
Call ensure_valid_token (triggering a refresh)
Confirm that the authorizer always updates its access token and only updates
refresh_token if one was present in the response
"""
authorizer.ensure_valid_token() # trigger refresh
token_data = response.by_resource_server["rs1"]
if "refresh_token" in token_data: # if present, confirm refresh token was updated
assert authorizer.access_token == "access_token_2"
assert authorizer.refresh_token == "refresh_token_2"
else: # otherwise, confirm no change
assert authorizer.access_token == "access_token_2"
assert authorizer.refresh_token == "refresh_token_1"
def test_refresh_token_authorizer_rejects_auth_service_client():
with pytest.raises(
globus_sdk.GlobusSDKUsageError,
match="RefreshTokenAuthorizer requires an AuthLoginClient",
):
globus_sdk.RefreshTokenAuthorizer(REFRESH_TOKEN, globus_sdk.AuthClient())
|
