Spaces:
Paused
Paused
| from moto import mock_aws | |
| import boto3 | |
| import pytest | |
| from helpers.dynamodb_helper import create_table_if_not_exists, DDB_TABLE | |
| class TestDynamoDBHelper: | |
| def dynamodb(self): | |
| with mock_aws(): | |
| yield boto3.resource("dynamodb", region_name="ca-central-1") | |
| def test_create_table_if_not_exists(self, dynamodb, aws_credentials): | |
| # Ensure the table does not exist initially | |
| client = dynamodb.meta.client | |
| existing_tables = client.list_tables()["TableNames"] | |
| assert DDB_TABLE not in existing_tables | |
| # Create the table | |
| table = create_table_if_not_exists(dynamodb) | |
| assert table is not None | |
| # Verify the table now exists | |
| existing_tables = client.list_tables()["TableNames"] | |
| assert DDB_TABLE in existing_tables | |
| # Attempt to create the table again, should not raise an error | |
| table = create_table_if_not_exists(dynamodb) | |
| assert table is not None | |
| def test_log_event(self, dynamodb, aws_credentials): | |
| table = create_table_if_not_exists(dynamodb) | |
| table_resource = 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") | |