sync

Author: Lucas McDonald

TestVectors/runtimes/python/src/aws_dbesdk_dynamodb_test_vectors/internaldafny/extern/CreateInterceptedDDBResource.py

@@ -0,0 +1,146 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+import boto3
+import types
+import aws_dbesdk_dynamodb_test_vectors.internaldafny.generated.CreateInterceptedDDBClient
+import aws_cryptography_internal_dynamodb.internaldafny.extern
+from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb.dafny_to_smithy import aws_cryptography_dbencryptionsdk_dynamodb_DynamoDbTablesEncryptionConfig
+from aws_dbesdk_dynamodb.encrypted.resource import EncryptedResource
+from smithy_dafny_standard_library.internaldafny.generated import Wrappers
+from aws_dbesdk_dynamodb.smithygenerated.aws_cryptography_dbencryptionsdk_dynamodb.errors import _smithy_error_to_dafny_error
+from aws_dbesdk_dynamodb_test_vectors.waiting_boto3_ddb_client import WaitingLocalDynamoClient
+from aws_dbesdk_dynamodb.internal.resource_to_client import ResourceShapeToClientShapeConverter
+from aws_dbesdk_dynamodb.internal.client_to_resource import ClientShapeToResourceShapeConverter
+
+class DynamoDBClientWrapperForDynamoDBResource:
+    """
+    Internal-only wrapper class for DBESDK TestVectors.
+
+    TestVectors Dafny code only knows how to interact with DynamoDB clients.
+    However, Python DDBEC and DBESDK have an EncryptedResource class that wraps boto3 DynamoDB Resources.
+    These classes create EncryptedTables that wrap boto3 DynamoDB Table Resources.
+    This class interfaces between Dafny TestVectors' DynamoDB client-calling code
+      and Python DBESDK's EncryptedResource/EncryptedTable classes.
+
+    This class defers to a boto3 client for create_table and delete_table,
+      which are not supported on boto3 DynamoDB Table resources.
+
+    TODO: Transact not supported on table. What do?
+    """
+
+    def __init__(self, resource, client):
+        self._resource = resource
+        self._client = client
+        self._client_shape_to_resource_shape_converter = ClientShapeToResourceShapeConverter()
+        self._resource_shape_to_client_shape_converter = ResourceShapeToClientShapeConverter()
+
+    def batch_write_item(self, **kwargs):
+        # The input here is from the DBESDK TestVectors, which is in the shape of a client request.
+        # Convert the client request to a resource request to be passed to the table.
+        resource_input = self._client_shape_to_resource_shape_converter.batch_write_item_request(kwargs)
+        resource_output = self._resource.batch_write_item(**resource_input)
+        client_output = self._resource_shape_to_client_shape_converter.batch_write_item_response(resource_output)
+        return client_output
+    
+    def batch_get_item(self, **kwargs):
+        resource_input = self._client_shape_to_resource_shape_converter.batch_get_item_request(kwargs)
+        resource_output = self._resource.batch_get_item(**resource_input)
+        client_output = self._resource_shape_to_client_shape_converter.batch_get_item_response(resource_output)
+        return client_output
+    
+    def scan(self, **kwargs):
+        # Resources don't have scan, but EncryptedResources can provide EncryptedTables that do support scan.
+        # This path tests that the EncryptedTables provided by EncryptedResources can used for scan.
+        table_name = kwargs["TableName"]
+        # Note: Any ConditionExpression strings are not converted to boto3 Condition objects
+        # and are passed as-is to the resource.
+        # They absolutely could be converted, but that is tested in the boto3 Table tests.
+        # Not doing this conversion here expands test coverage to both cases.
+        table_input = self._client_shape_to_resource_shape_converter.scan_request(kwargs)
+        encrypted_table = self._resource.Table(table_name)
+        table_output = encrypted_table.scan(**table_input)
+        table_shape_converter = ResourceShapeToClientShapeConverter(table_name=table_name)
+        client_output = table_shape_converter.scan_response(table_output)
+        return client_output
+    
+    def put_item(self, **kwargs):
+        # Resources don't have put_item, but EncryptedResources can provide EncryptedTables that do support put_item.
+        # This path tests that the EncryptedTables provided by EncryptedResources can used for put_item.
+        table_name = kwargs["TableName"]
+        table_input = self._client_shape_to_resource_shape_converter.put_item_request(kwargs)
+        encrypted_table = self._resource.Table(table_name)
+        table_output = encrypted_table.put_item(**table_input)
+        table_shape_converter = ResourceShapeToClientShapeConverter(table_name=table_name)
+        client_output = table_shape_converter.put_item_response(table_output)
+        return client_output
+    
+    def get_item(self, **kwargs):
+        # Resources don't have get_item, but EncryptedResources can provide EncryptedTables that do support get_item.
+        # This path tests that the EncryptedTables provided by EncryptedResources can used for get_item.
+        table_name = kwargs["TableName"]
+        table_input = self._client_shape_to_resource_shape_converter.get_item_request(kwargs)
+        encrypted_table = self._resource.Table(table_name)
+        table_output = encrypted_table.get_item(**table_input)
+        table_shape_converter = ResourceShapeToClientShapeConverter(table_name=table_name)
+        client_output = table_shape_converter.get_item_response(table_output)
+        return client_output
+    
+    def query(self, **kwargs):
+        # Resources don't have query, but EncryptedResources can provide EncryptedTables that do support query.
+        # This path tests that the EncryptedTables provided by EncryptedResources can used for query.
+        table_name = kwargs["TableName"]
+        # Note: Any ConditionExpression strings are not converted to boto3 Condition objects
+        # and are passed as-is to the resource.
+        # They absolutely could be converted, but that is tested in the boto3 Table tests.
+        # Not doing this conversion here expands test coverage to both cases.
+        table_input = self._client_shape_to_resource_shape_converter.query_request(kwargs)
+        encrypted_table = self._resource.Table(table_name)
+        table_output = encrypted_table.query(**table_input)
+        table_shape_converter = ResourceShapeToClientShapeConverter(table_name=table_name)
+        client_output = table_shape_converter.query_response(table_output)
+        return client_output
+
+    def transact_get_items(self, **kwargs):
+        raise NotImplementedError("transact_get_items not supported on resources")
+
+    def transact_write_items(self, **kwargs):
+        raise NotImplementedError("transact_get_items not supported on resources")
+
+    def delete_table(self, **kwargs):
+        # Resources don't have delete_table. Plus, DBESDK doesn't intercept DeleteTable calls.
+        # TestVectors only use this to ensure a new, clean table is created for each test.
+        # Defer to the underlying boto3 client to delete the table.
+        return self._client.delete_table(**kwargs)
+
+    def create_table(self, **kwargs):
+        # Resources don't have create_table. Plus, DBESDK doesn't intercept CreateTable calls.
+        # TestVectors only use this to ensure a new, clean table is created for each test.
+        # Defer to the underlying boto3 client to create a table.
+        return self._client.create_table(**kwargs)
+
+
+class default__:
+    @staticmethod
+    def CreateVanillaDDBClient():
+        try:
+            return aws_cryptography_internal_dynamodb.internaldafny.extern.Com_Amazonaws_Dynamodb.default__.DynamoDBClient(WaitingLocalDynamoClient())
+        except Exception as e:
+            return Wrappers.Result_Failure(_smithy_error_to_dafny_error(e))
+
+    @staticmethod
+    def CreateInterceptedDDBClient(dafny_encryption_config):
+        try:
+            native_encryption_config = aws_cryptography_dbencryptionsdk_dynamodb_DynamoDbTablesEncryptionConfig(dafny_encryption_config)
+            boto3_client = WaitingLocalDynamoClient()
+            table_config_names = list(native_encryption_config.table_encryption_configs.keys())
+            if len(table_config_names) > 1:
+                raise ValueError("TODO more than 1 table; need EncryptedTablesManager")
+            # For TestVectors, use local DynamoDB endpoint
+            resource = boto3.resource('dynamodb', endpoint_url="http://localhost:8000")
+            encrypted_resource = EncryptedResource(resource = resource, encryption_config = native_encryption_config)
+            wrapped_encrypted_resource = DynamoDBClientWrapperForDynamoDBResource(resource = encrypted_resource, client = boto3_client)
+            return aws_cryptography_internal_dynamodb.internaldafny.extern.Com_Amazonaws_Dynamodb.default__.DynamoDBClient(wrapped_encrypted_resource)
+        except Exception as e:
+            return Wrappers.Result_Failure(_smithy_error_to_dafny_error(e))
+
+aws_dbesdk_dynamodb_test_vectors.internaldafny.generated.CreateInterceptedDDBClient.default__ = default__

TestVectors/runtimes/python/test/__init__.py

@@ -0,0 +1,2 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0

TestVectors/runtimes/python/test/resource/__init__.py

@@ -0,0 +1,64 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""
+Wrapper file for executing Dafny tests from pytest.
+This allows us to import modules required by Dafny-generated tests
+before executing Dafny-generated tests.
+pytest will find and execute the `test_dafny` method below,
+which will execute the `internaldafny_test_executor.py` file in the `dafny` directory.
+"""
+
+import sys
+from functools import partial
+
+# Different from standard test_dafny_wrapper due to weird test structure.
+test_dir = '/'.join(__file__.split("/")[:-2])
+
+sys.path.append(test_dir + "/internaldafny/extern")
+sys.path.append(test_dir + "/internaldafny/generated")
+
+# Import extern to use an EncryptedResource as the wrapped DBESDK client.
+import aws_dbesdk_dynamodb_test_vectors.internaldafny.extern.CreateInterceptedDDBResource
+# Import extern to use the ItemEncryptor with Python dictionary-formatted items.
+# (EncryptedResources use Python dictionary-formatted items.)
+import aws_dbesdk_dynamodb_test_vectors.internaldafny.extern.CreateWrappedDictItemEncryptor
+
+# Remove invalid tests.
+# Supported operations on Resources that are also supported by DBESDK are:
+# - batch_get_item
+# - batch_write_item
+# https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb/service-resource/index.html
+# 
+# However, Resources can provide Tables. 
+# Unsupported operations on Resources are that are supported by provided Tables are:
+# - put_item
+# - get_item
+# - query
+# - scan
+# https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb/table/index.html#DynamoDB.Table
+# These operations will be tested on EncryptedResources via provided EncryptedTables.
+# 
+# Unsupported operations on both Resources and Tables are that are supported by DBESDK are:
+# - transact_get_items
+# - transact_write_items
+# Remove any tests that call unsupported operations by overriding the test method to do nothing..
+# If more tests that call these operations are added, remove them below.
+# If the list below becomes unmaintainable, or if other languages add clients with unsupported operations,
+# refactor the Dafny code to conditionally call tests based on whether the client supports the operation under test.
+
+def EmptyTest(*args, **kwargs):
+  print(f"Skipping test {kwargs['test_name']} because {kwargs['reason']}")
+
+aws_dbesdk_dynamodb_test_vectors.internaldafny.generated.DdbEncryptionTestVectors.TestVectorConfig.BasicIoTestTransactWriteItems = partial(
+  EmptyTest,
+  test_name="BasicIoTestTransactWriteItems",
+  reason="neither DDB resources nor DDB tables support transact_write_items"
+)
+aws_dbesdk_dynamodb_test_vectors.internaldafny.generated.DdbEncryptionTestVectors.TestVectorConfig.BasicIoTestTransactGetItems = partial(
+  EmptyTest,
+  test_name="BasicIoTestTransactGetItems",
+  reason="neither DDB resources nor DDB tables support transact_get_items"
+)
+
+def test_dafny():
+  from ..internaldafny.generated import __main__