12
12
# language governing permissions and limitations under the License.
13
13
"""Unit test suite for aws_encryption_sdk.streaming_client.StreamEncryptor"""
14
14
import io
15
- import unittest
16
15
17
16
import pytest
18
17
import six
36
35
pytestmark = [pytest .mark .unit , pytest .mark .local ]
37
36
38
37
39
- class TestStreamEncryptor (unittest .TestCase ):
40
- def setUp (self ):
38
+ class TestStreamEncryptor (object ):
39
+ @pytest .fixture (autouse = True )
40
+ def apply_fixtures (self ):
41
41
# Create mock key provider
42
42
self .mock_key_provider = MagicMock ()
43
43
self .mock_key_provider .__class__ = MasterKeyProvider
@@ -144,8 +144,8 @@ def setUp(self):
144
144
# Set up serialize_frame patch
145
145
self .mock_serialize_frame_patcher = patch ("aws_encryption_sdk.streaming_client.serialize_frame" )
146
146
self .mock_serialize_frame = self .mock_serialize_frame_patcher .start ()
147
-
148
- def tearDown ( self ):
147
+ yield
148
+ # Run tearDown
149
149
self .mock_content_type_patcher .stop ()
150
150
self .mock_validate_frame_length_patcher .stop ()
151
151
self .mock_message_id_patcher .stop ()
@@ -173,14 +173,15 @@ def test_init(self):
173
173
assert test_encryptor .content_type is sentinel .content_type
174
174
175
175
def test_init_non_framed_message_too_large (self ):
176
- with six . assertRaisesRegex ( self , SerializationError , "Source too large for non-framed message" ) :
176
+ with pytest . raises ( SerializationError ) as excinfo :
177
177
StreamEncryptor (
178
178
source = io .BytesIO (self .plaintext ),
179
179
key_provider = self .mock_key_provider ,
180
180
frame_length = 0 ,
181
181
algorithm = self .mock_algorithm ,
182
182
source_length = aws_encryption_sdk .internal .defaults .MAX_NON_FRAMED_SIZE + 1 ,
183
183
)
184
+ excinfo .match ("Source too large for non-framed message" )
184
185
185
186
def test_prep_message_no_master_keys (self ):
186
187
self .mock_key_provider .master_keys_for_encryption .return_value = sentinel .primary_master_key , set ()
@@ -190,9 +191,9 @@ def test_prep_message_no_master_keys(self):
190
191
frame_length = self .mock_frame_length ,
191
192
source_length = 5 ,
192
193
)
193
-
194
- with six .assertRaisesRegex (self , MasterKeyProviderError , "No Master Keys available from Master Key Provider" ):
194
+ with pytest .raises (MasterKeyProviderError ) as excinfo :
195
195
test_encryptor ._prep_message ()
196
+ excinfo .match ("No Master Keys available from Master Key Provider" )
196
197
197
198
def test_prep_message_primary_master_key_not_in_master_keys (self ):
198
199
self .mock_key_provider .master_keys_for_encryption .return_value = (
@@ -206,8 +207,9 @@ def test_prep_message_primary_master_key_not_in_master_keys(self):
206
207
source_length = 5 ,
207
208
)
208
209
209
- with six . assertRaisesRegex ( self , MasterKeyProviderError , "Primary Master Key not in provided Master Keys" ) :
210
+ with pytest . raises ( MasterKeyProviderError ) as excinfo :
210
211
test_encryptor ._prep_message ()
212
+ excinfo .match ("Primary Master Key not in provided Master Keys" )
211
213
212
214
def test_prep_message_algorithm_change (self ):
213
215
self .mock_encryption_materials .algorithm = Algorithm .AES_256_GCM_IV12_TAG16
@@ -217,13 +219,11 @@ def test_prep_message_algorithm_change(self):
217
219
algorithm = Algorithm .AES_128_GCM_IV12_TAG16 ,
218
220
source_length = 128 ,
219
221
)
220
-
221
- with six .assertRaisesRegex (
222
- self ,
223
- ActionNotAllowedError ,
224
- "Cryptographic materials manager provided algorithm suite differs from algorithm suite in request.*" ,
225
- ):
222
+ with pytest .raises (ActionNotAllowedError ) as excinfo :
226
223
test_encryptor ._prep_message ()
224
+ excinfo .match (
225
+ "Cryptographic materials manager provided algorithm suite differs from algorithm suite in request.*"
226
+ )
227
227
228
228
@patch ("aws_encryption_sdk.streaming_client.EncryptionMaterialsRequest" )
229
229
@patch ("aws_encryption_sdk.streaming_client.derive_data_encryption_key" )
@@ -396,8 +396,9 @@ def test_read_bytes_to_non_framed_body_too_large(self):
396
396
test_encryptor .bytes_read = aws_encryption_sdk .internal .defaults .MAX_NON_FRAMED_SIZE
397
397
test_encryptor ._StreamEncryptor__unframed_plaintext_cache = pt_stream
398
398
399
- with six . assertRaisesRegex ( self , SerializationError , "Source too large for non-framed message" ) :
399
+ with pytest . raises ( SerializationError ) as excinfo :
400
400
test_encryptor ._read_bytes_to_non_framed_body (5 )
401
+ excinfo .match ("Source too large for non-framed message" )
401
402
402
403
def test_read_bytes_to_non_framed_body_close (self ):
403
404
test_encryptor = StreamEncryptor (source = io .BytesIO (self .plaintext ), key_provider = self .mock_key_provider )
@@ -485,8 +486,9 @@ def test_read_bytes_unsupported_type(self, mock_read_non_framed, mock_read_frame
485
486
test_encryptor ._encryption_materials = self .mock_encryption_materials
486
487
test_encryptor ._header = MagicMock ()
487
488
test_encryptor .content_type = None
488
- with six . assertRaisesRegex ( self , NotSupportedError , "Unsupported content type" ) :
489
+ with pytest . raises ( NotSupportedError ) as excinfo :
489
490
test_encryptor ._read_bytes (5 )
491
+ excinfo .match ("Unsupported content type" )
490
492
assert not mock_read_non_framed .called
491
493
assert not mock_read_framed .called
492
494
0 commit comments