173 lines
5.1 KiB
Python
173 lines
5.1 KiB
Python
import base64
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal
|
|
from http import HTTPMethod
|
|
from urllib.parse import urlencode
|
|
|
|
import jsonlines
|
|
import pytest
|
|
|
|
PYTEST_TABLE_NAME = 'pytest'
|
|
PK = 'id'
|
|
SK = 'sk'
|
|
|
|
|
|
# https://docs.pytest.org/en/7.1.x/reference/reference.html#pytest.hookspec.pytest_configure
|
|
def pytest_configure():
|
|
os.environ['TZ'] = 'America/Sao_Paulo'
|
|
os.environ['DYNAMODB_PARTITION_KEY'] = PK
|
|
os.environ['DYNAMODB_SORT_KEY'] = SK
|
|
os.environ['USER_TABLE'] = PYTEST_TABLE_NAME
|
|
os.environ['COURSE_TABLE'] = PYTEST_TABLE_NAME
|
|
os.environ['ENROLLMENT_TABLE'] = PYTEST_TABLE_NAME
|
|
os.environ['ORDER_TABLE'] = PYTEST_TABLE_NAME
|
|
os.environ['LOG_LEVEL'] = 'DEBUG'
|
|
os.environ['BUCKET_NAME'] = 'saladeaula.digital'
|
|
|
|
|
|
@dataclass
|
|
class LambdaContext:
|
|
function_name: str = 'test'
|
|
memory_limit_in_mb: int = 128
|
|
invoked_function_arn: str = 'arn:aws:lambda:eu-west-1:809313241:function:test'
|
|
aws_request_id: str = '52fdfc07-2182-154f-163f-5f0f9a621d72'
|
|
|
|
|
|
@pytest.fixture
|
|
def lambda_context() -> LambdaContext:
|
|
return LambdaContext()
|
|
|
|
|
|
class HttpApiProxy:
|
|
def __call__(
|
|
self,
|
|
raw_path: str,
|
|
method: str = HTTPMethod.GET,
|
|
body: dict | str | None = None,
|
|
*,
|
|
headers: dict = {},
|
|
cookies: list[str] = [],
|
|
query_string_parameters: dict = {},
|
|
is_base64_encoded: bool = True,
|
|
**kwargs,
|
|
) -> dict:
|
|
if isinstance(body, dict):
|
|
body = json.dumps(body)
|
|
|
|
if is_base64_encoded and body:
|
|
body = _base64_encode(body)
|
|
|
|
return {
|
|
'version': '2.0',
|
|
'routeKey': '$default',
|
|
'rawPath': raw_path,
|
|
'rawQueryString': urlencode(query_string_parameters),
|
|
'cookies': cookies,
|
|
'headers': headers,
|
|
'queryStringParameters': query_string_parameters,
|
|
'requestContext': {
|
|
'accountId': '123456789012',
|
|
'apiId': 'api-id',
|
|
'authorizer': {
|
|
'jwt': {
|
|
'claims': {
|
|
'aud': '1db63660-063d-4280-b2ea-388aca4a9459',
|
|
'client_id': '1db63660-063d-4280-b2ea-388aca4a9459',
|
|
'email': 'sergio@somosbeta.com.br',
|
|
'email_verified': 'true',
|
|
'exp': '1765205975',
|
|
'iat': '1765202375',
|
|
'iss': 'https://id.saladeaula.digital',
|
|
'jti': 'Fbbyvwwze3npdEgs',
|
|
'name': 'Sérgio R Siqueira',
|
|
'scope': 'openid profile email offline_access apps:admin',
|
|
'sub': '5OxmMjL-ujoR5IMGegQz',
|
|
},
|
|
'scopes': None,
|
|
}
|
|
},
|
|
'domainName': 'id.execute-api.us-east-1.amazonaws.com',
|
|
'domainPrefix': 'id',
|
|
'http': {
|
|
'method': str(method),
|
|
'path': raw_path,
|
|
'protocol': 'HTTP/1.1',
|
|
'sourceIp': '192.168.0.1/32',
|
|
'userAgent': 'agent',
|
|
},
|
|
'requestId': 'id',
|
|
'routeKey': '$default',
|
|
'stage': '$default',
|
|
'time': '12/Mar/2020:19:03:58 +0000',
|
|
'timeEpoch': 1583348638390,
|
|
},
|
|
'body': body,
|
|
'pathParameters': {'parameter1': 'value1'},
|
|
'isBase64Encoded': is_base64_encoded,
|
|
'stageVariables': {'stageVariable1': 'value1', 'stageVariable2': 'value2'},
|
|
}
|
|
|
|
|
|
def _base64_encode(s: str) -> str | None:
|
|
if not s:
|
|
return None
|
|
return base64.b64encode(s.encode()).decode()
|
|
|
|
|
|
@pytest.fixture
|
|
def http_api_proxy():
|
|
return HttpApiProxy()
|
|
|
|
|
|
@pytest.fixture
|
|
def dynamodb_client():
|
|
from boto3clients import dynamodb_client as client
|
|
|
|
client.create_table(
|
|
AttributeDefinitions=[
|
|
{'AttributeName': PK, 'AttributeType': 'S'},
|
|
{'AttributeName': SK, 'AttributeType': 'S'},
|
|
],
|
|
TableName=PYTEST_TABLE_NAME,
|
|
KeySchema=[
|
|
{'AttributeName': PK, 'KeyType': 'HASH'},
|
|
{'AttributeName': SK, 'KeyType': 'RANGE'},
|
|
],
|
|
ProvisionedThroughput={
|
|
'ReadCapacityUnits': 123,
|
|
'WriteCapacityUnits': 123,
|
|
},
|
|
)
|
|
|
|
yield client
|
|
|
|
client.delete_table(TableName=PYTEST_TABLE_NAME)
|
|
|
|
|
|
@pytest.fixture()
|
|
def dynamodb_persistence_layer(dynamodb_client):
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer
|
|
|
|
return DynamoDBPersistenceLayer(PYTEST_TABLE_NAME, dynamodb_client)
|
|
|
|
|
|
@pytest.fixture()
|
|
def dynamodb_seeds(dynamodb_persistence_layer):
|
|
with open('tests/seeds.jsonl', 'rb') as fp:
|
|
reader = jsonlines.Reader(
|
|
fp,
|
|
loads=lambda s: json.loads(s, parse_float=Decimal), # type: ignore
|
|
)
|
|
|
|
for line in reader.iter(type=dict, skip_invalid=True):
|
|
dynamodb_persistence_layer.put_item(item=line)
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
import app
|
|
|
|
return app
|