autoformat

Author: gary-xero

test/functional/encrypted/test_resource.py

@@ -13,7 +13,7 @@
 """Functional tests for ``dynamodb_encryption_sdk.encrypted.resource``."""
 import pytest
 
-from ..functional_test_utils import example_table # noqa pylint: disable=unused-import
+from ..functional_test_utils import example_table  # noqa pylint: disable=unused-import
 from ..functional_test_utils import (
     TEST_TABLE_NAME,
     build_static_jce_cmp,

test/functional/encrypted/test_table.py

@@ -14,7 +14,7 @@
 import hypothesis
 import pytest
 
-from ..functional_test_utils import example_table # noqa pylint: disable=unused-import
+from ..functional_test_utils import example_table  # noqa pylint: disable=unused-import
 from ..functional_test_utils import (
     TEST_TABLE_NAME,
     build_static_jce_cmp,

test/functional/functional_test_utils.py

@@ -338,9 +338,7 @@ def diverse_item():
 
 
 def return_requestitems_as_unprocessed(*args, **kwargs):
-    return {
-        "UnprocessedItems": kwargs['RequestItems']
-    }
+    return {"UnprocessedItems": kwargs["RequestItems"]}
 
 
 def check_encrypted_item(plaintext_item, ciphertext_item, attribute_actions):
@@ -495,10 +493,7 @@ def cycle_batch_writer_check(raw_table, encrypted_table, initial_actions, initia
 
 
 def batch_write_item_unprocessed_check(
-    encrypted,
-    initial_item,
-    write_transformer=_nop_transformer,
-    table_name=TEST_TABLE_NAME,
+    encrypted, initial_item, write_transformer=_nop_transformer, table_name=TEST_TABLE_NAME
 ):
     """Check that unprocessed items in a batch result are unencrypted."""
     items = _generate_items(initial_item, write_transformer)
@@ -565,11 +560,7 @@ def table_cycle_batch_writer_check(materials_provider, initial_actions, initial_
 
 
 def table_batch_writer_unprocessed_items_check(
-    materials_provider,
-    initial_actions,
-    initial_item,
-    table_name,
-    region_name=None
+    materials_provider, initial_actions, initial_item, table_name, region_name=None
 ):
     kwargs = {}
     if region_name is not None:
@@ -582,7 +573,7 @@ def table_batch_writer_unprocessed_items_check(
 
     with patch.object(table.meta.client, "batch_write_item") as batch_write_mock:
         # Check that unprocessed items returned to a BatchWriter are successfully retried
-        batch_write_mock.side_effect = [{"UnprocessedItems": request_items}, {'UnprocessedItems': {}}]
+        batch_write_mock.side_effect = [{"UnprocessedItems": request_items}, {"UnprocessedItems": {}}]
         e_table = EncryptedTable(table=table, materials_provider=materials_provider, attribute_actions=initial_actions)
 
         with e_table.batch_writer() as writer:
@@ -616,27 +607,20 @@ def resource_cycle_batch_items_check(materials_provider, initial_actions, initia
 
 
 def resource_batch_items_unprocessed_check(
-    materials_provider,
-    initial_actions,
-    initial_item,
-    table_name,
-    region_name=None
+    materials_provider, initial_actions, initial_item, table_name, region_name=None
 ):
     kwargs = {}
     if region_name is not None:
         kwargs["region_name"] = region_name
-    resource = boto3.resource('dynamodb', **kwargs)
+    resource = boto3.resource("dynamodb", **kwargs)
 
     with patch.object(resource, "batch_write_item", return_requestitems_as_unprocessed):
         e_resource = EncryptedResource(
             resource=resource, materials_provider=materials_provider, attribute_actions=initial_actions
         )
 
         batch_write_item_unprocessed_check(
-            encrypted=e_resource,
-            initial_item=initial_item,
-            write_transformer=dict_to_ddb,
-            table_name=table_name,
+            encrypted=e_resource, initial_item=initial_item, write_transformer=dict_to_ddb, table_name=table_name
         )
 
 
@@ -691,27 +675,20 @@ def client_cycle_batch_items_check(materials_provider, initial_actions, initial_
 
 
 def client_batch_items_unprocessed_check(
-    materials_provider,
-    initial_actions,
-    initial_item,
-    table_name,
-    region_name=None
+    materials_provider, initial_actions, initial_item, table_name, region_name=None
 ):
     kwargs = {}
     if region_name is not None:
         kwargs["region_name"] = region_name
-    client = boto3.client('dynamodb', **kwargs)
+    client = boto3.client("dynamodb", **kwargs)
 
     with patch.object(client, "batch_write_item", return_requestitems_as_unprocessed):
         e_client = EncryptedClient(
             client=client, materials_provider=materials_provider, attribute_actions=initial_actions
         )
 
         batch_write_item_unprocessed_check(
-            encrypted=e_client,
-            initial_item=initial_item,
-            write_transformer=dict_to_ddb,
-            table_name=table_name,
+            encrypted=e_client, initial_item=initial_item, write_transformer=dict_to_ddb, table_name=table_name
         )
 
 

test/functional/internal/test_utils.py

@@ -62,14 +62,11 @@ def get_test_items(standard_dict_format, table_name="table", with_sort_keys=Fals
 def get_dummy_crypto_config(partition_key_name=None, sort_key_name=None, sign_keys=False):
     context = EncryptionContext(partition_key_name=partition_key_name, sort_key_name=sort_key_name)
     actions = AttributeActions(
-        default_action=CryptoAction.DO_NOTHING,
-        attribute_actions={
-            "encrypt-me": CryptoAction.ENCRYPT_AND_SIGN
-        },
+        default_action=CryptoAction.DO_NOTHING, attribute_actions={"encrypt-me": CryptoAction.ENCRYPT_AND_SIGN}
     )
     if sign_keys:
-        actions.attribute_actions['partition-key'] = CryptoAction.SIGN_ONLY
-        actions.attribute_actions['sort-key'] = CryptoAction.SIGN_ONLY
+        actions.attribute_actions["partition-key"] = CryptoAction.SIGN_ONLY
+        actions.attribute_actions["sort-key"] = CryptoAction.SIGN_ONLY
 
     materials = Mock(spec=CryptographicMaterialsProvider)  # type: CryptographicMaterialsProvider
     return CryptoConfig(materials_provider=materials, encryption_context=context, attribute_actions=actions)
@@ -78,7 +75,7 @@ def get_dummy_crypto_config(partition_key_name=None, sort_key_name=None, sign_ke
 def check_encrypt_batch_write_item_call(request_items, crypto_config):
     def dummy_encrypt(item, **kwargs):
         result = item.copy()
-        result['encrypt-me'] = "pretend Im encrypted"
+        result["encrypt-me"] = "pretend Im encrypted"
         return result
 
     # execute a batch write, but make the write method return ALL the provided items as unprocessed
@@ -96,11 +93,7 @@ def dummy_encrypt(item, **kwargs):
 
 
 @pytest.mark.parametrize(
-    "items",
-    (
-        get_test_items(standard_dict_format=True),
-        get_test_items(standard_dict_format=False),
-    )
+    "items", (get_test_items(standard_dict_format=True), get_test_items(standard_dict_format=False))
 )
 def test_encrypt_batch_write_returns_plaintext_unprocessed_items_with_known_partition_key(items):
     crypto_config = get_dummy_crypto_config("partition-key")
@@ -112,7 +105,7 @@ def test_encrypt_batch_write_returns_plaintext_unprocessed_items_with_known_part
     (
         get_test_items(standard_dict_format=True, with_sort_keys=True),
         get_test_items(standard_dict_format=False, with_sort_keys=True),
-    )
+    ),
 )
 def test_encrypt_batch_write_returns_plaintext_unprocessed_items_with_known_partition_and_sort_keys(items):
     crypto_config = get_dummy_crypto_config("partition-key", "sort-key")
@@ -126,7 +119,7 @@ def test_encrypt_batch_write_returns_plaintext_unprocessed_items_with_known_part
         get_test_items(standard_dict_format=False),
         get_test_items(standard_dict_format=True, with_sort_keys=True),
         get_test_items(standard_dict_format=False, with_sort_keys=True),
-    )
+    ),
 )
 def test_encrypt_batch_write_returns_plaintext_unprocessed_items_with_unknown_keys(items):
     crypto_config = get_dummy_crypto_config(None, None)
@@ -141,7 +134,7 @@ def test_encrypt_batch_write_returns_plaintext_unprocessed_items_with_unknown_ke
         get_test_items(standard_dict_format=False),
         get_test_items(standard_dict_format=True, with_sort_keys=True),
         get_test_items(standard_dict_format=False, with_sort_keys=True),
-    )
+    ),
 )
 def test_encrypt_batch_write_returns_plaintext_unprocessed_items_with_unknown_signed_keys(items):
     crypto_config = get_dummy_crypto_config(None, None, sign_keys=True)