more unit tests

Author: lucasmcdonald3

src/aws_encryption_sdk/streaming_client.py

@@ -73,8 +73,9 @@
 try:
     # pylint should pass even if the MPL isn't installed
     # noqa pylint: disable=import-error
-    from aws_cryptographic_materialproviders.mpl.client import AwsCryptographicMaterialProviders
+    from aws_cryptographic_materialproviders.mpl import AwsCryptographicMaterialProviders
     from aws_cryptographic_materialproviders.mpl.config import MaterialProvidersConfig
+    from aws_cryptographic_materialproviders.mpl.errors import AwsCryptographicMaterialProvidersException
     from aws_cryptographic_materialproviders.mpl.models import CreateDefaultCryptographicMaterialsManagerInput
     from aws_cryptographic_materialproviders.mpl.references import IKeyring
     _HAS_MPL = True
@@ -147,9 +148,6 @@ def _has_mpl_attrs_post_init(self):
         """If the MPL is present in the runtime, perform MPL-specific post-init logic
         to validate the new object has a valid state.
         """
-        if not hasattr(self, "keyring"):
-            self._no_mpl_attrs_post_init()
-            return
         if not exactly_one_arg_is_not_none(self.materials_manager, self.key_provider, self.keyring):
             raise TypeError("Exactly one of keyring, materials_manager, or key_provider must be provided")
         if self.materials_manager is None:
@@ -159,21 +157,21 @@ def _has_mpl_attrs_post_init(self):
                     master_key_provider=self.key_provider
                 )
             elif self.keyring is not None:
-                # No CMM, provided MPL keyring => create MPL's DefaultCryptographicMaterialsManager
-                if not isinstance(self.keyring, IKeyring):
-                    raise ValueError(f"Argument provided to keyring MUST be a {IKeyring}. \
-                                     Found {self.keyring.__class__.__name__}")
-
-                mat_prov: AwsCryptographicMaterialProviders = AwsCryptographicMaterialProviders(
-                    config=MaterialProvidersConfig()
-                )
-                cmm = mat_prov.create_default_cryptographic_materials_manager(
-                    CreateDefaultCryptographicMaterialsManagerInput(
-                        keyring=self.keyring
+                try:
+                    mat_prov: AwsCryptographicMaterialProviders = AwsCryptographicMaterialProviders(
+                        config=MaterialProvidersConfig()
                     )
-                )
-                cmm_handler: CryptoMaterialsManager = CryptoMaterialsManagerFromMPL(cmm)
-                self.materials_manager = cmm_handler
+                    cmm = mat_prov.create_default_cryptographic_materials_manager(
+                        CreateDefaultCryptographicMaterialsManagerInput(
+                            keyring=self.keyring
+                        )
+                    )
+                    cmm_handler: CryptoMaterialsManager = CryptoMaterialsManagerFromMPL(cmm)
+                    self.materials_manager = cmm_handler
+                except AwsCryptographicMaterialProvidersException as mpl_exception:
+                    # Wrap MPL error into the ESDK error type
+                    # so customers only have to catch ESDK error types.
+                    raise AWSEncryptionSDKClientError(mpl_exception)                    
 
     def _no_mpl_attrs_post_init(self):
         """If the MPL is NOT present in the runtime, perform post-init logic

test/unit/test_crypto_authentication_signer.py

@@ -12,7 +12,8 @@
 # language governing permissions and limitations under the License.
 """Unit test suite for ``aws_encryption_sdk.internal.crypto.authentication.Signer``."""
 import pytest
-from mock import MagicMock, sentinel
+from mock import MagicMock, sentinel, patch
+import cryptography.hazmat.primitives.serialization
 from pytest_mock import mocker  # noqa pylint: disable=unused-import
 
 import aws_encryption_sdk.internal.crypto.authentication
@@ -75,28 +76,72 @@ def test_f_signer_from_key_bytes():
 def test_f_signer_key_bytes():
     test = Signer(algorithm=ALGORITHM, key=VALUES["ecc_private_key_prime"])
     assert test.key_bytes() == VALUES["ecc_private_key_prime_private_bytes"]
+    
 
+def test_GIVEN_no_encoding_WHEN_signer_from_key_bytes_THEN_load_der_private_key(
+    patch_default_backend,
+    patch_build_hasher,
+    patch_ec
+):
+    mock_algorithm_info = MagicMock(return_value=sentinel.algorithm_info, spec=patch_ec.EllipticCurve)
+    _algorithm = MagicMock(signing_algorithm_info=mock_algorithm_info)
 
-def test_signer_from_key_bytes(patch_default_backend, patch_serialization, patch_build_hasher, patch_ec):
+    # Make a new patched serialization module for this test.
+    # The default patch introduces serialization as `serialization.Encoding.DER`
+    # from within the src, but is `Encoding.DER` in the test.
+    # This namespace change causes the src's `isinstance` checks to fail.
+    # Mock the `serialization.Encoding.DER`
+    with patch.object(cryptography.hazmat.primitives, "serialization"):
+        # Mock the `serialization.load_der_private_key`
+        with patch.object(aws_encryption_sdk.internal.crypto.authentication.serialization, "load_der_private_key") as mock_der:
+            Signer.from_key_bytes(
+                algorithm=_algorithm,
+                key_bytes=sentinel.key_bytes,
+            )
+
+            mock_der.assert_called_once_with(
+                data=sentinel.key_bytes, password=None, backend=patch_default_backend.return_value
+            )
+
+    
+def test_GIVEN_PEM_encoding_WHEN_signer_from_key_bytes_THEN_load_pem_private_key(
+    patch_default_backend,
+    patch_serialization,
+    patch_build_hasher,
+    patch_ec
+):
     mock_algorithm_info = MagicMock(return_value=sentinel.algorithm_info, spec=patch_ec.EllipticCurve)
     _algorithm = MagicMock(signing_algorithm_info=mock_algorithm_info)
 
-    # Explicitly pass in patched serialization module.
-    # Patching the module introduces namespace issues
-    # which causes the method's `isinstance` checks to fail
-    # by changing the namespace from `serialization.Encoding.DER` to `Encoding.DER`.
     signer = Signer.from_key_bytes(
         algorithm=_algorithm,
         key_bytes=sentinel.key_bytes,
-        encoding=patch_serialization.Encoding.DER
+        encoding=patch_serialization.Encoding.PEM
     )
 
-    patch_serialization.load_der_private_key.assert_called_once_with(
+    patch_serialization.load_pem_private_key.assert_called_once_with(
         data=sentinel.key_bytes, password=None, backend=patch_default_backend.return_value
     )
     assert isinstance(signer, Signer)
     assert signer.algorithm is _algorithm
-    assert signer.key is patch_serialization.load_der_private_key.return_value
+    assert signer.key is patch_serialization.load_pem_private_key.return_value
+
+
+def test_GIVEN_unrecognized_encoding_WHEN_signer_from_key_bytes_THEN_raise_ValueError(
+    patch_default_backend,
+    patch_serialization,
+    patch_build_hasher,
+    patch_ec
+):
+    mock_algorithm_info = MagicMock(return_value=sentinel.algorithm_info, spec=patch_ec.EllipticCurve)
+    _algorithm = MagicMock(signing_algorithm_info=mock_algorithm_info)
+
+    with pytest.raises(ValueError):
+        signer = Signer.from_key_bytes(
+            algorithm=_algorithm,
+            key_bytes=sentinel.key_bytes,
+            encoding="not an encoding"
+        )
 
 
 def test_signer_key_bytes(patch_default_backend, patch_serialization, patch_build_hasher, patch_ec):

test/unit/test_streaming_client_configs.py

@@ -15,6 +15,7 @@
 
 import pytest
 import six
+from mock import patch
 
 from aws_encryption_sdk import CommitmentPolicy
 from aws_encryption_sdk.internal.defaults import ALGORITHM, FRAME_LENGTH, LINE_LENGTH
@@ -28,6 +29,22 @@
 pytestmark = [pytest.mark.unit, pytest.mark.local]
 
 
+# Check if MPL is installed, and skip tests based on its installation status
+# Ideally, this logic would be based on mocking imports and testing logic,
+# but doing that introduces errors that cause other tests to fail.
+try:
+    from aws_cryptographic_materialproviders.mpl.references import (
+        IKeyring,
+    )
+    HAS_MPL = True
+
+    from aws_encryption_sdk.materials_managers.mpl.cmm import (
+        CryptoMaterialsManagerFromMPL,
+    )
+except ImportError:
+    HAS_MPL = False
+
+
 class FakeCryptoMaterialsManager(CryptoMaterialsManager):
     def get_encryption_materials(self, request):
         return
@@ -42,6 +59,14 @@ class FakeMasterKeyProvider(MasterKeyProvider):
 
     def _new_master_key(self, key_id):
         return
+    
+if HAS_MPL:
+    class FakeKeyring(IKeyring):
+        def on_encrypt(self, param):
+            return 
+        
+        def on_decrypt(self, param):
+            return 
 
 
 BASE_KWARGS = dict(
@@ -126,6 +151,18 @@ def test_client_config_defaults():
     assert test.max_encrypted_data_keys is None
 
 
+@pytest.mark.skipif(not HAS_MPL, reason="Test should only be executed with MPL in installation")
+def test_client_config_with_mpl_attr():
+    test = _ClientConfig(**BASE_KWARGS)
+    assert hasattr(test, "keyring")
+
+
+@pytest.mark.skipif(HAS_MPL, reason="Test should only be executed without MPL in installation")
+def test_client_config_no_mpl():
+    test = _ClientConfig(**BASE_KWARGS)
+    assert not hasattr(test, "keyring")
+
+
 def test_encryptor_config_defaults():
     test = EncryptorConfig(**BASE_KWARGS)
     assert test.encryption_context == {}
@@ -154,3 +191,62 @@ def test_client_config_converts(kwargs, stream_type):
     assert isinstance(test.source, stream_type)
     if test.key_provider is not None:
         assert isinstance(test.materials_manager, DefaultCryptoMaterialsManager)
+
+
+@pytest.mark.skipif(HAS_MPL, reason="Test should only be executed without MPL in installation")
+@patch.object(_ClientConfig, "_no_mpl_attrs_post_init")
+def test_GIVEN_no_mpl_WHEN_attrs_post_init_THEN_calls_no_mpl_method(
+    mock_no_mpl_attrs_post_init,
+):
+    _ClientConfig(**BASE_KWARGS)
+    mock_no_mpl_attrs_post_init.assert_called_once_with()
+
+
+@pytest.mark.skipif(not HAS_MPL, reason="Test should only be executed with MPL in installation")
+@patch.object(_ClientConfig, "_has_mpl_attrs_post_init")
+def test_GIVEN_has_mpl_WHEN_attrs_post_init_THEN_calls_no_mpl_method(
+    _has_mpl_attrs_post_init,
+):
+    _ClientConfig(**BASE_KWARGS)
+    _has_mpl_attrs_post_init.assert_called_once_with()
+
+
+@pytest.mark.parametrize(
+    "kwargs, stream_type",
+    (
+        (dict(source=b"", materials_manager=FakeCryptoMaterialsManager()), io.BytesIO),
+        (dict(source=b"", key_provider=FakeMasterKeyProvider()), io.BytesIO),
+        (dict(source="", materials_manager=FakeCryptoMaterialsManager()), io.BytesIO),
+        (dict(source=io.BytesIO(), materials_manager=FakeCryptoMaterialsManager()), io.BytesIO),
+        (dict(source=six.StringIO(), materials_manager=FakeCryptoMaterialsManager()), six.StringIO),
+        (dict(source=b"", keyring=FakeKeyring()), io.BytesIO),
+    ),
+)
+@pytest.mark.skipif(not HAS_MPL, reason="Test should only be executed with MPL in installation")
+def test_client_configs_with_mpl(
+    kwargs,
+    stream_type
+):
+    kwargs["commitment_policy"] = CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT
+
+    test = _ClientConfig(**kwargs)
+    
+    # In all cases, config should have a materials manager
+    assert test.materials_manager is not None
+    
+    # If materials manager was provided, it should be directly used
+    if hasattr(kwargs, "materials_manager"):
+        assert kwargs["materials_manager"] == test.materials_manager
+
+    # If MPL keyring was provided, it should be wrapped in MPL materials manager
+    if hasattr(kwargs, "keyring"):
+        assert test.keyring is not None
+        assert test.keyring == kwargs["keyring"]
+        assert isinstance(test.keyring, IKeyring)
+        assert isinstance(test.materials_manager, CryptoMaterialsManagerFromMPL)
+
+    # If native key_provider was provided, it should be wrapped in native materials manager
+    if hasattr(kwargs, "key_provider"):
+        assert test.key_provider is not None
+        assert test.key_provider == kwargs["key_provider"]
+        assert isinstance(test.materials_manager, DefaultCryptoMaterialsManager)