wip

api.saladeaula.digital/tests/conftest.py

@@ -0,0 +1,154 @@
+import base64
+import json
+import os
+from dataclasses import dataclass
+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['COURSE_TABLE'] = PYTEST_TABLE_NAME
+    os.environ['DYNAMODB_PARTITION_KEY'] = PK
+    os.environ['DYNAMODB_SORT_KEY'] = SK
+
+
+@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'
+
+
+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': {},
+                '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 lambda_context() -> LambdaContext:
+    return LambdaContext()
+
+
+@pytest.fixture
+def http_api_proxy():
+    return HttpApiProxy()
+
+
+@pytest.fixture
+def dynamodb_client():
+    from boto3clients import dynamodb_client as client
+
+    try:
+        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,
+            },
+        )
+    except Exception:
+        pass
+
+    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 seeds(dynamodb_client):
+    from layercake.dynamodb import serialize
+
+    with open('tests/seeds.jsonl', 'rb') as fp:
+        reader = jsonlines.Reader(fp)
+
+        for line in reader.iter(type=dict, skip_invalid=True):
+            dynamodb_client.put_item(
+                TableName=PYTEST_TABLE_NAME,
+                Item=serialize(line),
+            )
+
+
+@pytest.fixture
+def app():
+    import app
+
+    return app