1###################################################################### 2# 3# File: test/unit/v1/test_copy_manager.py 4# 5# Copyright 2021 Backblaze Inc. All Rights Reserved. 6# 7# License https://www.backblaze.com/using_b2_code.html 8# 9###################################################################### 10 11from ..test_base import TestBase 12 13from .deps_exception import InvalidAuthToken, Unauthorized, SSECKeyIdMismatchInCopy 14from .deps import MetadataDirectiveMode 15from .deps import EncryptionAlgorithm, EncryptionSetting, EncryptionMode, EncryptionKey, SSE_NONE, SSE_B2_AES 16from b2sdk.transfer.outbound.copy_manager import CopyManager 17from b2sdk.http_constants import SSE_C_KEY_ID_FILE_INFO_KEY_NAME 18 19SSE_C_AES = EncryptionSetting( 20 mode=EncryptionMode.SSE_C, 21 algorithm=EncryptionAlgorithm.AES256, 22 key=EncryptionKey(secret=b'some_key', key_id='some-id'), 23) 24SSE_C_AES_2 = EncryptionSetting( 25 mode=EncryptionMode.SSE_C, 26 algorithm=EncryptionAlgorithm.AES256, 27 key=EncryptionKey(secret=b'some_other_key', key_id='some-id-2'), 28) 29 30 31class TestCopyManager(TestBase): 32 def test_establish_sse_c_replace(self): 33 file_info = {'some_key': 'some_value'} 34 content_type = 'text/plain' 35 metadata_directive, new_file_info, new_content_type = CopyManager.establish_sse_c_file_metadata( 36 MetadataDirectiveMode.REPLACE, 37 destination_file_info=file_info, 38 destination_content_type=content_type, 39 destination_server_side_encryption=SSE_C_AES, 40 source_server_side_encryption=SSE_C_AES_2, 41 source_file_info=None, 42 source_content_type=None, 43 ) 44 self.assertEqual( 45 ( 46 MetadataDirectiveMode.REPLACE, { 47 'some_key': 'some_value', 48 SSE_C_KEY_ID_FILE_INFO_KEY_NAME: 'some-id' 49 }, content_type 50 ), (metadata_directive, new_file_info, new_content_type) 51 ) 52 53 def test_establish_sse_c_copy_no_enc(self): 54 file_info = {} 55 content_type = 'text/plain' 56 metadata_directive, new_file_info, new_content_type = CopyManager.establish_sse_c_file_metadata( 57 MetadataDirectiveMode.COPY, 58 destination_file_info=file_info, 59 destination_content_type=content_type, 60 destination_server_side_encryption=None, 61 source_server_side_encryption=None, 62 source_file_info=None, 63 source_content_type=None, 64 ) 65 self.assertEqual( 66 (MetadataDirectiveMode.COPY, {}, content_type), 67 (metadata_directive, new_file_info, new_content_type) 68 ) 69 70 def test_establish_sse_c_copy_b2(self): 71 file_info = {} 72 content_type = 'text/plain' 73 metadata_directive, new_file_info, new_content_type = CopyManager.establish_sse_c_file_metadata( 74 MetadataDirectiveMode.COPY, 75 destination_file_info=file_info, 76 destination_content_type=content_type, 77 destination_server_side_encryption=SSE_B2_AES, 78 source_server_side_encryption=None, 79 source_file_info=None, 80 source_content_type=None, 81 ) 82 self.assertEqual( 83 (MetadataDirectiveMode.COPY, {}, content_type), 84 (metadata_directive, new_file_info, new_content_type) 85 ) 86 87 def test_establish_sse_c_copy_same_key_id(self): 88 file_info = None 89 content_type = 'text/plain' 90 metadata_directive, new_file_info, new_content_type = CopyManager.establish_sse_c_file_metadata( 91 MetadataDirectiveMode.COPY, 92 destination_file_info=file_info, 93 destination_content_type=content_type, 94 destination_server_side_encryption=SSE_C_AES, 95 source_server_side_encryption=SSE_C_AES, 96 source_file_info=None, 97 source_content_type=None, 98 ) 99 self.assertEqual( 100 (MetadataDirectiveMode.COPY, None, content_type), 101 (metadata_directive, new_file_info, new_content_type) 102 ) 103 104 def test_establish_sse_c_copy_sources_given(self): 105 metadata_directive, new_file_info, new_content_type = CopyManager.establish_sse_c_file_metadata( 106 MetadataDirectiveMode.COPY, 107 destination_file_info=None, 108 destination_content_type=None, 109 destination_server_side_encryption=SSE_C_AES, 110 source_server_side_encryption=SSE_C_AES_2, 111 source_file_info={ 112 'some_key': 'some_value', 113 SSE_C_KEY_ID_FILE_INFO_KEY_NAME: 'some-id-2' 114 }, 115 source_content_type='text/plain', 116 ) 117 self.assertEqual( 118 ( 119 MetadataDirectiveMode.REPLACE, { 120 'some_key': 'some_value', 121 SSE_C_KEY_ID_FILE_INFO_KEY_NAME: 'some-id' 122 }, 'text/plain' 123 ), (metadata_directive, new_file_info, new_content_type) 124 ) 125 126 def test_establish_sse_c_copy_sources_unknown(self): 127 for source_file_info, source_content_type in [ 128 (None, None), 129 ({ 130 'a': 'b' 131 }, None), 132 (None, 'text/plain'), 133 ]: 134 with self.subTest( 135 source_file_info=source_file_info, source_content_type=source_content_type 136 ): 137 with self.assertRaises( 138 SSECKeyIdMismatchInCopy, 139 'attempting to copy file using MetadataDirectiveMode.COPY without providing source_file_info ' 140 'and source_content_type for differing sse_c_key_ids: source="some-id-2", destination="some-id"' 141 ): 142 CopyManager.establish_sse_c_file_metadata( 143 MetadataDirectiveMode.COPY, 144 destination_file_info=None, 145 destination_content_type=None, 146 destination_server_side_encryption=SSE_C_AES, 147 source_server_side_encryption=SSE_C_AES_2, 148 source_file_info=source_file_info, 149 source_content_type=source_content_type, 150 ) 151