sync

Author: Lucas McDonald

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

@@ -21,16 +21,15 @@
     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")
+from . import sort_dynamodb_json_lists
 
 
+# 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 +51,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
@@ -75,6 +78,10 @@ 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
@@ -128,22 +135,6 @@ def put_item_request(expect_standard_dictionaries, test_item):
     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 +159,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 +181,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/unit/encrypted/test_paginator.py

@@ -61,3 +61,18 @@ def test_GIVEN_kwargs_has_PaginationConfig_WHEN_paginate_THEN_PaginationConfig_i
     mock_input_transform_method.assert_called_once_with(
         QueryInputTransformInput(sdk_input=kwargs_without_pagination_config)
     )
+
+
+def test_GIVEN_invalid_class_attribute_WHEN_getattr_THEN_raise_error():
+    # Create a mock with a specific spec that excludes our unknown attribute
+    mock_boto3_dynamodb_client = MagicMock(spec=["put_item", "get_item", "query", "scan"])
+    encrypted_paginator = EncryptedPaginator(
+        paginator=mock_boto3_dynamodb_client,
+        encryption_config=mock_tables_encryption_config,
+    )
+
+    # Then: AttributeError is raised
+    with pytest.raises(AttributeError):
+        # Given: Invalid class attribute: not_a_valid_attribute_on_EncryptedClient_nor_boto3_client
+        # When: getattr is called
+        encrypted_paginator.not_a_valid_attribute_on_EncryptedPaginator_nor_boto3_paginator()