160 lines
4.2 KiB
Python
160 lines
4.2 KiB
Python
import json
|
|
from base64 import b64encode
|
|
from http import HTTPMethod, HTTPStatus
|
|
from urllib.parse import urlencode
|
|
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer
|
|
|
|
from ..conftest import HttpApiProxy, LambdaContext
|
|
|
|
CLIENT_ID = 'd72d4005-1fa7-4430-9754-80d5e2487bb6'
|
|
CLIENT_SECRET = '1nFD8alDbGHgc3g1RLY960xyRJVee0SlMoIB0MUlSuiJy28W'
|
|
AUTH = b64encode(f'{CLIENT_ID}:{CLIENT_SECRET}'.encode()).decode()
|
|
|
|
|
|
def test_token(
|
|
app,
|
|
seeds,
|
|
dynamodb_persistence_layer: DynamoDBPersistenceLayer,
|
|
http_api_proxy: HttpApiProxy,
|
|
lambda_context: LambdaContext,
|
|
):
|
|
client_id = '8c5e92b0-9ed4-451e-8935-66084d8544b1'
|
|
|
|
r = app.lambda_handler(
|
|
http_api_proxy(
|
|
raw_path='/token',
|
|
method=HTTPMethod.POST,
|
|
headers={
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
},
|
|
body=urlencode(
|
|
{
|
|
'grant_type': 'authorization_code',
|
|
'redirect_uri': 'https://localhost/callback',
|
|
'code': 'kyqp3oSuRFTfuBaCmq3XOgGWg67l42Kt3D6xPEj7Yd3MLdi9',
|
|
'client_id': client_id,
|
|
'code_verifier': (
|
|
'9072df2d3709425993e733f38fb27a825b8860e699364ce9'
|
|
'abafdf51077c0bdb4e456ddb741147a4bec4eeda782d92cc'
|
|
),
|
|
}
|
|
),
|
|
),
|
|
lambda_context,
|
|
)
|
|
|
|
assert r['statusCode'] == HTTPStatus.OK
|
|
|
|
r = json.loads(r['body'])
|
|
assert r['expires_in'] == 3600
|
|
|
|
tokens = dynamodb_persistence_layer.query(
|
|
key_cond_expr='#pk = :pk',
|
|
expr_attr_name={
|
|
'#pk': 'id',
|
|
},
|
|
expr_attr_values={
|
|
':pk': 'OAUTH2#TOKEN',
|
|
},
|
|
)
|
|
assert len(tokens['items']) == 3
|
|
|
|
r = app.lambda_handler(
|
|
http_api_proxy(
|
|
raw_path='/token',
|
|
method=HTTPMethod.POST,
|
|
headers={
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
},
|
|
body=urlencode(
|
|
{
|
|
'grant_type': 'refresh_token',
|
|
'refresh_token': r['refresh_token'],
|
|
'client_id': client_id,
|
|
}
|
|
),
|
|
),
|
|
lambda_context,
|
|
)
|
|
|
|
r = dynamodb_persistence_layer.query(
|
|
key_cond_expr='#pk = :pk',
|
|
expr_attr_name={
|
|
'#pk': 'id',
|
|
},
|
|
expr_attr_values={
|
|
':pk': 'OAUTH2#TOKEN',
|
|
},
|
|
)
|
|
|
|
assert len(r['items']) == 4
|
|
|
|
|
|
def test_refresh_token(
|
|
app,
|
|
seeds,
|
|
dynamodb_persistence_layer: DynamoDBPersistenceLayer,
|
|
http_api_proxy: HttpApiProxy,
|
|
lambda_context: LambdaContext,
|
|
):
|
|
r = app.lambda_handler(
|
|
http_api_proxy(
|
|
raw_path='/token',
|
|
method=HTTPMethod.POST,
|
|
headers={
|
|
'Authorization': f'Basic {AUTH}',
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
},
|
|
body=urlencode(
|
|
{
|
|
'grant_type': 'refresh_token',
|
|
'refresh_token': 'CyF3Ik3b9hMIo3REVv27gZAHd7dvwZq6QrkhWr7qHEen4UVy',
|
|
'client_id': CLIENT_ID,
|
|
}
|
|
),
|
|
),
|
|
lambda_context,
|
|
)
|
|
|
|
r = dynamodb_persistence_layer.query(
|
|
key_cond_expr='#pk = :pk',
|
|
expr_attr_name={
|
|
'#pk': 'id',
|
|
},
|
|
expr_attr_values={
|
|
':pk': 'OAUTH2#TOKEN',
|
|
},
|
|
)
|
|
|
|
assert len(r['items']) == 2
|
|
|
|
|
|
def test_refresh_token_invalid_grant(
|
|
app,
|
|
seeds,
|
|
dynamodb_persistence_layer: DynamoDBPersistenceLayer,
|
|
http_api_proxy: HttpApiProxy,
|
|
lambda_context: LambdaContext,
|
|
):
|
|
r = app.lambda_handler(
|
|
http_api_proxy(
|
|
raw_path='/token',
|
|
method=HTTPMethod.POST,
|
|
headers={
|
|
'Authorization': f'Basic {AUTH}',
|
|
'Content-Type': 'application/x-www-form-urlencoded',
|
|
},
|
|
body=urlencode(
|
|
{
|
|
'grant_type': 'refresh_token',
|
|
'refresh_token': '1234',
|
|
'client_id': CLIENT_ID,
|
|
}
|
|
),
|
|
),
|
|
lambda_context,
|
|
)
|
|
|
|
assert r['statusCode'] == 400
|