| from moto import mock_aws
|
| import boto3
|
| import pytest
|
| from helpers.dynamodb_helper import create_table_if_not_exists, DDB_TABLE
|
| from botocore.exceptions import ClientError
|
| import os
|
|
|
|
|
| class TestDynamoDBHelper:
|
| dynamodb = boto3.resource("dynamodb", region_name="ca-central-1")
|
|
|
| @mock_aws
|
| def test_create_table_if_not_exists(self):
|
|
|
| client = self.dynamodb.meta.client
|
| existing_tables = client.list_tables()["TableNames"]
|
| assert DDB_TABLE not in existing_tables
|
|
|
|
|
| table = create_table_if_not_exists(self.dynamodb)
|
| assert table is not None
|
|
|
|
|
| existing_tables = client.list_tables()["TableNames"]
|
| assert DDB_TABLE in existing_tables
|
|
|
|
|
| table = create_table_if_not_exists(self.dynamodb)
|
| assert table is not None
|
|
|
| @mock_aws
|
| def test_log_event(self):
|
| table = create_table_if_not_exists(self.dynamodb)
|
|
|
| table_resource = self.dynamodb.Table(DDB_TABLE)
|
| user_id = "user123"
|
| session_id = "test-session-456"
|
| data = {"event": "test_event", "value": 26, "float_value": 3.14}
|
| from helpers.dynamodb_helper import log_event
|
|
|
| log_event(user_id, session_id, data)
|
| response = table_resource.scan()
|
| assert response["Count"] == 1
|
| item = response["Items"][0]
|
| assert item["PK"] == f"SESSION#{session_id}"
|
| assert item["data"]["event"] == "test_event"
|
| from decimal import Decimal
|
|
|
| assert item["data"]["value"] == Decimal(26)
|
| assert item["data"]["float_value"] == Decimal("3.14")
|
|
|