tests that do not need to talk to KMS should not need the CMK environment variable to be set: some of the integ tests were looking for this in setup instead of execution, triggering failures even if the pytest markers for those tests were not requested

Author: mattsb42-aws

test/integration/integration_test_utils.py

@@ -16,6 +16,7 @@
 from aws_encryption_sdk.key_providers.kms import KMSMasterKeyProvider
 
 AWS_KMS_KEY_ID = 'AWS_ENCRYPTION_SDK_PYTHON_INTEGRATION_TEST_AWS_KMS_KEY_ID'
+_KMS_MKP = None
 
 
 def get_cmk_arn():
@@ -32,9 +33,17 @@ def get_cmk_arn():
     raise ValueError('KMS CMK ARN provided for integration tests much be a key not an alias')
 
 
-def setup_kms_master_key_provider():
+def setup_kms_master_key_provider(cache=True):
     """Reads the test_values config file and builds the requested KMS Master Key Provider."""
+    global _KMS_MKP  # pylint: disable=global-statement
+    if cache and _KMS_MKP is not None:
+        return _KMS_MKP
+
     cmk_arn = get_cmk_arn()
     kms_master_key_provider = KMSMasterKeyProvider()
     kms_master_key_provider.add_master_key(cmk_arn)
+
+    if cache:
+        _KMS_MKP = kms_master_key_provider
+
     return kms_master_key_provider

test/integration/test_i_thread_safety.py

@@ -22,7 +22,7 @@
 from six.moves import queue  # six.moves confuses pylint: disable=import-error
 
 import aws_encryption_sdk
-from .integration_test_utils import setup_kms_master_key_provider
+from .integration_test_utils import get_cmk_arn, setup_kms_master_key_provider
 
 pytestmark = [pytest.mark.integ]
 
@@ -49,7 +49,7 @@ def crypto_thread_worker(crypto_function, start_pause, input_value, output_queue
     :param cache: Cache to use with master key provider (optional)
     """
     time.sleep(start_pause)
-    kms_master_key_provider = setup_kms_master_key_provider()
+    kms_master_key_provider = setup_kms_master_key_provider(cache=False)
     if cache is None:
         # For simplicity, always use a caching CMM; just use a null cache if no cache is specified.
         cache = aws_encryption_sdk.NullCryptoMaterialsCache()
@@ -105,6 +105,8 @@ def random_pause_time(max_seconds=3):
 
 def test_threading_loop():
     """Test thread safety of client."""
+    # Check for the CMK ARN first to fail fast if it is not available
+    get_cmk_arn()
     rounds = 20
     plaintext_inputs = [
         dict(input_value=copy.copy(PLAINTEXT), start_pause=random_pause_time())
@@ -130,6 +132,8 @@ def test_threading_loop():
 
 def test_threading_loop_with_common_cache():
     """Test thread safety of client while using common cryptographic materials cache across all threads."""
+    # Check for the CMK ARN first to fail fast if it is not available
+    get_cmk_arn()
     rounds = 20
     cache = aws_encryption_sdk.LocalCryptoMaterialsCache(capacity=40)
     plaintext_inputs = [

test/integration/test_i_xcompat_kms.py

@@ -34,7 +34,6 @@ def _file_root():
 
 
 def _generate_test_cases():
-    kms_key_provider = setup_kms_master_key_provider()
     try:
         root_dir = os.path.abspath(file_root())
     except Exception:  # pylint: disable=broad-except
@@ -65,22 +64,21 @@ def _generate_test_cases():
             if key['provider_id'] == 'aws-kms' and key['decryptable']:
                 _test_cases.append((
                     os.path.join(base_dir, test_case['plaintext']['filename']),
-                    os.path.join(base_dir, test_case['ciphertext']['filename']),
-                    kms_key_provider
+                    os.path.join(base_dir, test_case['ciphertext']['filename'])
                 ))
                 break
     return _test_cases
 
 
-@pytest.mark.parametrize('plaintext_filename,ciphertext_filename,key_provider', _generate_test_cases())
-def test_decrypt_from_file(plaintext_filename, ciphertext_filename, key_provider):
+@pytest.mark.parametrize('plaintext_filename, ciphertext_filename', _generate_test_cases())
+def test_decrypt_from_file(plaintext_filename, ciphertext_filename):
     """Tests decrypt from known good files."""
     with open(ciphertext_filename, 'rb') as infile:
         ciphertext = infile.read()
     with open(plaintext_filename, 'rb') as infile:
         plaintext = infile.read()
     decrypted_ciphertext, _header = aws_encryption_sdk.decrypt(
         source=ciphertext,
-        key_provider=key_provider
+        key_provider=setup_kms_master_key_provider()
     )
     assert decrypted_ciphertext == plaintext