m

Author: Lucas McDonald

DynamoDbEncryption/runtimes/python/test/integ/encrypted/test_client.py

@@ -2,7 +2,6 @@
 # SPDX-License-Identifier: Apache-2.0
 import boto3
 import pytest
-from boto3.dynamodb.types import TypeDeserializer, TypeSerializer
 
 from aws_dbesdk_dynamodb.encrypted.client import EncryptedClient
 from aws_dbesdk_dynamodb.encrypted.paginator import EncryptedPaginator
@@ -45,9 +44,6 @@
 )
 from . import sort_dynamodb_json_lists
 
-serializer = TypeSerializer()
-deserializer = TypeDeserializer()
-
 
 # Creates a matrix of tests for each value in param,
 # with a user-friendly string for test output:

DynamoDbEncryption/runtimes/python/test/integ/encrypted/test_paginator.py

@@ -3,6 +3,8 @@
 
 from aws_dbesdk_dynamodb.encrypted.client import EncryptedClient
 
+from . import sort_dynamodb_json_lists
+
 from ...constants import (
     INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME,
     INTEG_TEST_DEFAULT_TABLE_CONFIGS,
@@ -21,16 +23,14 @@
     basic_put_item_request_ddb,
     basic_put_item_request_dict,
     basic_query_paginator_request,
-    basic_scan_request_ddb,
-    basic_scan_request_dict,
+    basic_scan_paginator_request,
 )
 
-BOTO3_CLIENT = boto3.client("dynamodb")
-ENCRYPTED_CLIENT = EncryptedClient(client=BOTO3_CLIENT, encryption_config=INTEG_TEST_DEFAULT_TABLE_CONFIGS)
-SCAN_PAGINATOR = ENCRYPTED_CLIENT.get_paginator("scan")
-QUERY_PAGINATOR = ENCRYPTED_CLIENT.get_paginator("query")
-
 
+# Creates a matrix of tests for each value in param,
+# with a user-friendly string for test output:
+# expect_standard_dictionaries = True -> "standard_dicts"
+# expect_standard_dictionaries = False -> "ddb_json"
 @pytest.fixture(params=[True, False], ids=["standard_dicts", "ddb_json"])
 def expect_standard_dictionaries(request):
     return request.param
@@ -52,6 +52,10 @@ def plaintext_client(expect_standard_dictionaries):
     return client
 
 
+# Creates a matrix of tests for each value in param,
+# with a user-friendly string for test output:
+# encrypted = True -> "encrypted"
+# encrypted = False -> "plaintext"
 @pytest.fixture(params=[True, False], ids=["encrypted", "plaintext"])
 def encrypted(request):
     return request.param
@@ -74,7 +78,10 @@ def query_paginator(client):
 def scan_paginator(client):
     return client.get_paginator("scan")
 
-
+# Creates a matrix of tests for each value in param,
+# with a user-friendly string for test output:
+# use_complex_item = True -> "complex_item"
+# use_complex_item = False -> "simple_item"
 @pytest.fixture(params=[True, False], ids=["complex_item", "simple_item"])
 def use_complex_item(request):
     return request.param
@@ -127,23 +134,6 @@ def put_item_request(expect_standard_dictionaries, test_item):
         return {**basic_put_item_request_dict(test_item), "TableName": INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME}
     return basic_put_item_request_ddb(test_item)
 
-
-def sort_dynamodb_json_lists(obj):
-    """
-    Utility that recursively sorts all lists in a DynamoDB JSON-like structure.
-    DynamoDB JSON uses lists to represent sets, so strict equality can fail.
-    Sort lists to ensure consistent ordering when comparing expected and actual items.
-    """
-    if isinstance(obj, dict):
-        return {k: sort_dynamodb_json_lists(v) for k, v in obj.items()}
-    elif isinstance(obj, list):
-        try:
-            return sorted(obj)  # Sort lists for consistent comparison
-        except TypeError:
-            return obj  # Not all lists are sortable; ex. complex_item_ddb's "list" attribute
-    return obj
-
-
 def test_GIVEN_query_paginator_WHEN_paginate_THEN_returns_expected_items(
     client, query_paginator, paginate_query_request, put_item_request, test_item
 ):
@@ -168,15 +158,11 @@ def test_GIVEN_query_paginator_WHEN_paginate_THEN_returns_expected_items(
 
 
 @pytest.fixture
-def paginate_scan_request(expect_standard_dictionaries, encrypted, test_item):
+def paginate_scan_request(expect_standard_dictionaries, test_item):
     if expect_standard_dictionaries:
-        request = {**basic_scan_request_dict(test_item), "TableName": INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME}
+        request = {**basic_scan_paginator_request(test_item), "TableName": INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME}
     else:
-        request = basic_scan_request_ddb(test_item)
-    if encrypted:
-        request["FilterExpression"] = request["FilterExpression"] + " AND attribute_exists (#sig)"
-        request["ExpressionAttributeNames"] = {}
-        request["ExpressionAttributeNames"]["#sig"] = "amzn-ddb-map-sig"
+        request = basic_scan_paginator_request(test_item)
     return request
 
 
@@ -194,70 +180,10 @@ def test_GIVEN_scan_paginator_WHEN_paginate_THEN_returns_expected_items(
         if "Items" in page:
             for item in page["Items"]:
                 items.append(item)
-
-
-# TODO: set up scan table and tests
-
-# @pytest.fixture
-# def default_requests(simple_item_ddb):
-#     """Fixture to provide default scan and query requests."""
-#     return {
-#         "scan": get_scan_request(simple_item_ddb["partition_key"]),
-#         "query": get_query_request(simple_item_ddb["partition_key"]),
-#     }
-
-# @pytest.fixture(scope="module", autouse=True)
-# def setup_module_fixture(simple_item_ddb):
-#     # Runs before all tests in this module
-#     for i in range(10):
-#         ENCRYPTED_CLIENT.put_item(
-#             TableName=INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME,
-#             Item={
-#                 "partition_key": simple_item_ddb["partition_key"],
-#                 "sort_key": {"N": str(i)},
-#                 "attribute1": {"S": "encrypt and sign me!"},
-#                 "attribute2": {"S": "sign me!"},
-#                 ":attribute3": {"S": "ignore me!"},
-#             },
-#         )
-#     # Yield to all tests in this module
-#     yield
-#     # TODO: Delete items? not needed but nice
-
-# @pytest.mark.parametrize("paginator_name, request_key", [
-#     ("scan", "scan"),
-#     ("query", "query")
-# ])
-# def test_GIVEN_incomplete_search_WHEN_build_full_result_THEN_NextToken_completes_search(
-#     paginator_name,
-#     request_key,
-#     default_requests,
-# ):
-#     # TODO: test
-#     return
-#     # Given: PaginationConfig that will result in an incomplete search.
-#     # There are 10 items, and returning 2 per operation results in multiple uses of NextToken.
-#     pagination_config = {
-#         "MaxItems": 2,
-#     }
-
-#     paginator = getattr(ENCRYPTED_CLIENT, f"get_paginator")(paginator_name)
-#     incomplete_search_request = default_requests[request_key].copy()
-#     incomplete_search_request["PaginationConfig"] = pagination_config
-
-#     # When: Build full result from paginator
-#     full_result = paginator.paginate(**incomplete_search_request)
-
-#     print(f"{full_result=}")
-
-#     collected_items = []
-
-#     print(f'{full_result=}')
-
-#     for item in full_result:
-#         # assert item["attribute1"]["S"] == "encrypt and sign me!"
-#         collected_items.append(item)
-
-#     # Then: NextToken should be present
-#     # assert "NextToken" in full_result
-#     # Then: NextToken is able to complete the search
+    assert len(items) == 1
+    # DynamoDB JSON uses lists to represent sets, so strict equality can fail.
+    # Sort lists to ensure consistent ordering when comparing expected and actual items.
+    expected_item = sort_dynamodb_json_lists(test_item)
+    actual_item = sort_dynamodb_json_lists(items[0])
+    # Then: Items are equal
+    assert expected_item == actual_item

DynamoDbEncryption/runtimes/python/test/integ/encrypted/test_table.py

@@ -1,6 +1,5 @@
 import boto3
 import pytest
-from boto3.dynamodb.types import TypeDeserializer, TypeSerializer
 
 from aws_dbesdk_dynamodb.encrypted.table import EncryptedTable
 
@@ -16,10 +15,6 @@
     basic_scan_request_dict,
 )
 
-serializer = TypeSerializer()
-deserializer = TypeDeserializer()
-
-
 def encrypted_table():
     """Create an encrypted table."""
     table = boto3.resource("dynamodb").Table(INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME)

DynamoDbEncryption/runtimes/python/test/requests.py

@@ -281,6 +281,14 @@ def basic_query_paginator_request(key):
         "ExpressionAttributeValues": {":pk": key["partition_key"], ":sk": key["sort_key"]},
     }
 
+def basic_scan_paginator_request(item):
+    """Get a scan paginator request in DDB format for any item."""
+    return {
+        "TableName": INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME,
+        "FilterExpression": "partition_key = :pk AND sort_key = :sk",
+        "ExpressionAttributeValues": {":pk": item["partition_key"], ":sk": item["sort_key"]},
+    }
+
 
 # Dict format request functions
 

Examples/runtimes/python/DynamoDBEncryption/src/basic_put_get_example/with_encrypted_table.py

@@ -20,11 +20,11 @@
 )
 from aws_cryptographic_material_providers.mpl.references import IKeyring
 from aws_dbesdk_dynamodb.encrypted.table import EncryptedTable
-from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb.models import (
+from aws_dbesdk_dynamodb.structures.dynamodb import (
     DynamoDbTableEncryptionConfig,
     DynamoDbTablesEncryptionConfig,
 )
-from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_structuredencryption.models import (
+from aws_dbesdk_dynamodb.structures.structured_encryption import (
     CryptoAction,
 )