m

Author: Lucas McDonald

.github/workflows/ci_test_python.yml

@@ -25,9 +25,9 @@ jobs:
       fail-fast: false
       matrix:
         library: [DynamoDbEncryption]
-        python-version: ["3.11"]
+        python-version: ["3.11", "3.12", "3.13"]
         os: [
-            # macos-13,
+            macos-13,
             ubuntu-22.04,
             # Dafny-transpiled Python tests use a PYTHONPATH hack that doesn't work on Windows.
             # Windows is tested with non-Dafny-transpiled Python tests.

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

@@ -8,6 +8,8 @@
 from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_transforms.errors import (
     DynamoDbEncryptionTransformsException,
 )
+import uuid
+from copy import deepcopy
 
 from ...constants import (
     INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME,
@@ -110,45 +112,80 @@ def client(encrypted, expect_standard_dictionaries):
 def use_complex_item(request):
     return request.param
 
+# Append a suffix to the partition key to avoid collisions between test runs.
+@pytest.fixture(scope="module")
+def test_run_suffix():
+    return str(uuid.uuid4())
 
 @pytest.fixture
-def test_item(expect_standard_dictionaries, use_complex_item):
+def test_item(expect_standard_dictionaries, use_complex_item, test_run_suffix):
     """Get a single test item in the appropriate format for the client."""
     if expect_standard_dictionaries:
         if use_complex_item:
-            return complex_item_dict
-        return simple_item_dict
-    if use_complex_item:
-        return complex_item_ddb
-    return simple_item_ddb
-
+            item = deepcopy(complex_item_dict)
+        else:
+            item = deepcopy(simple_item_dict)
+    else:
+        if use_complex_item:
+            item = deepcopy(complex_item_ddb)
+        else:
+            item = deepcopy(simple_item_ddb)
+    # Add a suffix to the partition key to avoid collisions between test runs.
+    if isinstance(item["partition_key"], dict):
+        item["partition_key"]["S"] += test_run_suffix
+    else:
+        item["partition_key"] += test_run_suffix
+    return item
 
 @pytest.fixture
-def test_key(expect_standard_dictionaries, use_complex_item):
+def test_key(expect_standard_dictionaries, use_complex_item, test_run_suffix):
     """Get a single test item in the appropriate format for the client."""
     if expect_standard_dictionaries:
         if use_complex_item:
-            return complex_key_dict
-        return simple_key_dict
-    if use_complex_item:
-        return complex_key_ddb
-    return simple_key_ddb
+            key = deepcopy(complex_key_dict)
+        else:
+            key = deepcopy(simple_key_dict)
+    else:
+        if use_complex_item:
+            key = deepcopy(complex_key_ddb)
+        else:
+            key = deepcopy(simple_key_ddb)
+    # Add a suffix to the partition key to avoid collisions between test runs.
+    if isinstance(key["partition_key"], dict):
+        key["partition_key"]["S"] += test_run_suffix
+    else:
+        key["partition_key"] += test_run_suffix
+    return key
 
 
 @pytest.fixture
-def multiple_test_items(expect_standard_dictionaries):
+def multiple_test_items(expect_standard_dictionaries, test_run_suffix):
     """Get two test items in the appropriate format for the client."""
     if expect_standard_dictionaries:
-        return [simple_item_dict, complex_item_dict]
-    return [simple_item_ddb, complex_item_ddb]
+        items = [deepcopy(simple_item_dict), deepcopy(complex_item_dict)]
+    else:
+        items = [deepcopy(simple_item_ddb), deepcopy(complex_item_ddb)]
+    for item in items:
+        if isinstance(item["partition_key"], dict):
+            item["partition_key"]["S"] += test_run_suffix
+        else:
+            item["partition_key"] += test_run_suffix
+    return items
 
 
 @pytest.fixture
-def multiple_test_keys(expect_standard_dictionaries):
+def multiple_test_keys(expect_standard_dictionaries, test_run_suffix):
     """Get two test keys in the appropriate format for the client."""
     if expect_standard_dictionaries:
-        return [simple_key_dict, complex_key_dict]
-    return [simple_key_ddb, complex_key_ddb]
+        keys = [deepcopy(simple_key_dict), deepcopy(complex_key_dict)]
+    else:
+        keys = [deepcopy(simple_key_ddb), deepcopy(complex_key_ddb)]
+    for key in keys:
+        if isinstance(key["partition_key"], dict):
+            key["partition_key"]["S"] += test_run_suffix
+        else:
+            key["partition_key"] += test_run_suffix
+    return keys
 
 
 @pytest.fixture
@@ -303,10 +340,18 @@ def test_GIVEN_valid_put_and_query_requests_WHEN_put_and_query_THEN_round_trip_p
 
 
 @pytest.fixture
-def scan_request(expect_standard_dictionaries, test_item):
+def scan_request(expect_standard_dictionaries, test_item, encrypted):
     if expect_standard_dictionaries:
-        return {**basic_scan_request_dict(test_item), "TableName": INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME}
-    return basic_scan_request_ddb(test_item)
+        request = {**basic_scan_request_dict(test_item), "TableName": INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME}
+    else:
+        request = basic_scan_request_ddb(test_item)
+    if encrypted:
+        # If the encrypted scan encounters a plaintext item, the scan will fail.
+        # To avoid this, encrypted scans add a filter expression that matches only encrypted items.
+        request["FilterExpression"] = request["FilterExpression"] + " AND attribute_exists (#sig)"
+        request["ExpressionAttributeNames"] = {}
+        request["ExpressionAttributeNames"]["#sig"] = "amzn-ddb-map-sig"
+    return request
 
 
 def test_GIVEN_valid_put_and_scan_requests_WHEN_put_and_scan_THEN_round_trip_passes(
@@ -323,7 +368,6 @@ def test_GIVEN_valid_put_and_scan_requests_WHEN_put_and_scan_THEN_round_trip_pas
     scan_response = client.scan(**scan_request)
     # Then: scan succeeds
     assert scan_response["ResponseMetadata"]["HTTPStatusCode"] == 200
-    assert len(scan_response["Items"]) >= 1
     # Can't assert anything about the scan;
     # there are too many items.
     # The critical assertion is that the scan succeeds.
@@ -590,3 +634,13 @@ def test_WHEN_call_passthrough_method_THEN_correct_response_is_returned():
     response = encrypted_client(expect_standard_dictionaries=False).list_backups()
     # Then: Correct response is returned, i.e. EncryptedClient forwards the call to the underlying boto3 client
     assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
+
+# Delete the items in the table after the module runs
+@pytest.fixture(scope="module", autouse=True)
+def cleanup_after_module(test_run_suffix):
+    yield
+    table = boto3.client("dynamodb")
+    items = [deepcopy(simple_item_ddb), deepcopy(complex_item_ddb)]
+    for item in items:
+        item["partition_key"]["S"] += test_run_suffix
+        table.delete_item(**basic_delete_item_request_ddb(item))

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

@@ -1,5 +1,7 @@
 import boto3
 import pytest
+import uuid
+from copy import deepcopy
 
 from aws_dbesdk_dynamodb.encrypted.client import EncryptedClient
 
@@ -86,37 +88,65 @@ def scan_paginator(client):
 def use_complex_item(request):
     return request.param
 
+# Append a suffix to the partition key to avoid collisions between test runs.
+@pytest.fixture(scope="module")
+def test_run_suffix():
+    return str(uuid.uuid4())
 
 @pytest.fixture
-def test_key(expect_standard_dictionaries, use_complex_item):
+def test_key(expect_standard_dictionaries, use_complex_item, test_run_suffix):
     """Get a single test item in the appropriate format for the client."""
     if expect_standard_dictionaries:
         if use_complex_item:
-            return complex_key_dict
-        return simple_key_dict
-    if use_complex_item:
-        return complex_key_ddb
-    return simple_key_ddb
-
+            key = deepcopy(complex_key_dict)
+        else:
+            key = deepcopy(simple_key_dict)
+    else:
+        if use_complex_item:
+            key = deepcopy(complex_key_ddb)
+        else:
+            key = deepcopy(simple_key_ddb)
+    # Add a suffix to the partition key to avoid collisions between test runs.
+    if isinstance(key["partition_key"], dict):
+        key["partition_key"]["S"] += test_run_suffix
+    else:
+        key["partition_key"] += test_run_suffix
+    return key
 
 @pytest.fixture
-def multiple_test_keys(expect_standard_dictionaries):
+def multiple_test_keys(expect_standard_dictionaries, test_run_suffix):
     """Get two test keys in the appropriate format for the client."""
     if expect_standard_dictionaries:
-        return [simple_key_dict, complex_key_dict]
-    return [simple_key_ddb, complex_key_ddb]
-
+        keys = [deepcopy(simple_key_dict), deepcopy(complex_key_dict)]
+    else:
+        keys = [deepcopy(simple_key_ddb), deepcopy(complex_key_ddb)]
+    # Add a suffix to the partition key to avoid collisions between test runs.
+    for key in keys:
+        if isinstance(key["partition_key"], dict):
+            key["partition_key"]["S"] += test_run_suffix
+        else:
+            key["partition_key"] += test_run_suffix
+    return keys
 
 @pytest.fixture
-def test_item(expect_standard_dictionaries, use_complex_item):
+def test_item(expect_standard_dictionaries, use_complex_item, test_run_suffix):
     """Get a single test item in the appropriate format for the client."""
     if expect_standard_dictionaries:
         if use_complex_item:
-            return complex_item_dict
-        return simple_item_dict
-    if use_complex_item:
-        return complex_item_ddb
-    return simple_item_ddb
+            item = deepcopy(complex_item_dict)
+        else:
+            item = deepcopy(simple_item_dict)
+    else:
+        if use_complex_item:
+            item = deepcopy(complex_item_ddb)
+        else:
+            item = deepcopy(simple_item_ddb)
+    # Add a suffix to the partition key to avoid collisions between test runs.
+    if isinstance(item["partition_key"], dict):
+        item["partition_key"]["S"] += test_run_suffix
+    else:
+        item["partition_key"] += test_run_suffix
+    return item
 
 
 @pytest.fixture

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

@@ -1,5 +1,7 @@
 import boto3
 import pytest
+import uuid
+from copy import deepcopy
 
 from aws_dbesdk_dynamodb.encrypted.resource import EncryptedResource, EncryptedTablesCollectionManager
 from aws_dbesdk_dynamodb.encrypted.table import EncryptedTable
@@ -41,23 +43,43 @@ def resource(encrypted):
 def tables(resource):
     return resource.tables
 
+@pytest.fixture(scope="module")
+def test_run_suffix():
+    return str(uuid.uuid4())
+
+@pytest.fixture
+def test_items(test_run_suffix):
+    items = [deepcopy(complex_item_dict), deepcopy(simple_item_dict)]
+    for item in items:
+        item["partition_key"] += test_run_suffix
+    return items
+
+@pytest.fixture
+def test_keys(test_run_suffix):
+    keys = [deepcopy(complex_key_dict), deepcopy(simple_key_dict)]
+    for key in keys:
+        key["partition_key"] += test_run_suffix
+    return keys
+
 
 def test_GIVEN_items_WHEN_batch_write_and_get_THEN_round_trip_passes(
     resource,
+    test_items,
+    test_keys,
 ):
-    batch_write_item_put_request = basic_batch_write_item_put_request_dict([simple_item_dict, complex_item_dict])
+    batch_write_item_put_request = basic_batch_write_item_put_request_dict(test_items)
     batch_write_response = resource.batch_write_item(**batch_write_item_put_request)
     assert batch_write_response["ResponseMetadata"]["HTTPStatusCode"] == 200
 
-    batch_get_item_request = basic_batch_get_item_request_dict([simple_key_dict, complex_key_dict])
+    batch_get_item_request = basic_batch_get_item_request_dict(test_keys)
     batch_get_response = resource.batch_get_item(**batch_get_item_request)
     assert batch_get_response["ResponseMetadata"]["HTTPStatusCode"] == 200
     responses = batch_get_response["Responses"][INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME]
     assert len(responses) == 2
     for response in responses:
-        assert response in [simple_item_dict, complex_item_dict]
+        assert response in test_items
 
-    batch_write_item_delete_request = basic_batch_write_item_delete_request_dict([simple_key_dict, complex_key_dict])
+    batch_write_item_delete_request = basic_batch_write_item_delete_request_dict(test_keys)
     batch_write_response = resource.batch_write_item(**batch_write_item_delete_request)
     assert batch_write_response["ResponseMetadata"]["HTTPStatusCode"] == 200
 

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

@@ -1,5 +1,7 @@
 import boto3
 import pytest
+import uuid
+from copy import deepcopy
 
 from aws_dbesdk_dynamodb.encrypted.table import EncryptedTable
 from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb_transforms.errors import (
@@ -57,14 +59,19 @@ def table(encrypted):
     else:
         return plaintext_table()
 
+@pytest.fixture(scope="module")
+def test_run_suffix():
+    return str(uuid.uuid4())
 
 # 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=[simple_item_dict, complex_item_dict], ids=["simple_item", "complex_item"])
-def test_item(request):
-    return request.param
+def test_item(request, test_run_suffix):
+    item = deepcopy(request.param)
+    item["partition_key"] += test_run_suffix
+    return item
 
 
 def test_GIVEN_item_WHEN_basic_put_AND_basic_get_AND_basic_delete_THEN_round_trip_passes(table, test_item):
@@ -183,7 +190,10 @@ def test_GIVEN_valid_put_and_scan_requests_WHEN_put_and_scan_THEN_round_trip_pas
     # When: Scanning items
     scan_request_dict = scan_request
     scan_response = table.scan(**scan_request_dict)
-    # Then: Scan returns both test items
+    # Then: Scan succeeds
+    # Can't assert anything about the scan;
+    # there are too many items.
+    # The critical assertion is that the scan succeeds.
     assert scan_response["ResponseMetadata"]["HTTPStatusCode"] == 200
 
 
@@ -228,3 +238,13 @@ def test_WHEN_call_passthrough_method_THEN_correct_response_is_returned():
     response = encrypted_table().table_name
     # Then: Correct response is returned, i.e. EncryptedTable forwards the call to the underlying boto3 table
     assert response == INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME
+
+# Delete the items in the table after the module runs
+@pytest.fixture(scope="module", autouse=True)
+def cleanup_after_module(test_run_suffix):
+    yield
+    table = boto3.resource("dynamodb").Table(INTEG_TEST_DEFAULT_DYNAMODB_TABLE_NAME)
+    items = [deepcopy(simple_item_dict), deepcopy(complex_item_dict)]
+    for item in items:
+        item["partition_key"] = item["partition_key"] + test_run_suffix
+        table.delete_item(**basic_delete_item_request_dict(item))