1import json 2import copy 3import tempfile 4 5from tests.unit import AWSMockServiceTestCase 6from boto.glacier.layer1 import Layer1 7 8 9class GlacierLayer1ConnectionBase(AWSMockServiceTestCase): 10 connection_class = Layer1 11 12 def setUp(self): 13 super(GlacierLayer1ConnectionBase, self).setUp() 14 self.json_header = [('Content-Type', 'application/json')] 15 self.vault_name = u'examplevault' 16 self.vault_arn = 'arn:aws:glacier:us-east-1:012345678901:vaults/' + \ 17 self.vault_name 18 self.vault_info = {u'CreationDate': u'2012-03-16T22:22:47.214Z', 19 u'LastInventoryDate': u'2012-03-21T22:06:51.218Z', 20 u'NumberOfArchives': 2, 21 u'SizeInBytes': 12334, 22 u'VaultARN': self.vault_arn, 23 u'VaultName': self.vault_name} 24 25 26class GlacierVaultsOperations(GlacierLayer1ConnectionBase): 27 28 def test_create_vault_parameters(self): 29 self.set_http_response(status_code=201) 30 self.service_connection.create_vault(self.vault_name) 31 32 def test_list_vaults(self): 33 content = {u'Marker': None, 34 u'RequestId': None, 35 u'VaultList': [self.vault_info]} 36 self.set_http_response(status_code=200, header=self.json_header, 37 body=json.dumps(content).encode('utf-8')) 38 api_response = self.service_connection.list_vaults() 39 self.assertDictEqual(content, api_response) 40 41 def test_describe_vaults(self): 42 content = copy.copy(self.vault_info) 43 content[u'RequestId'] = None 44 self.set_http_response(status_code=200, header=self.json_header, 45 body=json.dumps(content).encode('utf-8')) 46 api_response = self.service_connection.describe_vault(self.vault_name) 47 self.assertDictEqual(content, api_response) 48 49 def test_delete_vault(self): 50 self.set_http_response(status_code=204) 51 self.service_connection.delete_vault(self.vault_name) 52 53 54class GlacierJobOperations(GlacierLayer1ConnectionBase): 55 56 def setUp(self): 57 super(GlacierJobOperations, self).setUp() 58 self.job_content = 'abc' * 1024 59 60 def test_initiate_archive_job(self): 61 content = {u'Type': u'archive-retrieval', 62 u'ArchiveId': u'AAABZpJrTyioDC_HsOmHae8EZp_uBSJr6cnGOLKp_XJCl-Q', 63 u'Description': u'Test Archive', 64 u'SNSTopic': u'Topic', 65 u'JobId': None, 66 u'Location': None, 67 u'RequestId': None} 68 self.set_http_response(status_code=202, header=self.json_header, 69 body=json.dumps(content).encode('utf-8')) 70 api_response = self.service_connection.initiate_job(self.vault_name, 71 self.job_content) 72 self.assertDictEqual(content, api_response) 73 74 def test_get_archive_output(self): 75 header = [('Content-Type', 'application/octet-stream')] 76 self.set_http_response(status_code=200, header=header, 77 body=self.job_content) 78 response = self.service_connection.get_job_output(self.vault_name, 79 'example-job-id') 80 self.assertEqual(self.job_content, response.read()) 81 82 83class GlacierUploadArchiveResets(GlacierLayer1ConnectionBase): 84 def test_upload_archive(self): 85 fake_data = tempfile.NamedTemporaryFile() 86 fake_data.write(b'foobarbaz') 87 # First seek to a non zero offset. 88 fake_data.seek(2) 89 self.set_http_response(status_code=201) 90 # Simulate reading the request body when we send the request. 91 self.service_connection.connection.request.side_effect = \ 92 lambda *args: fake_data.read() 93 self.service_connection.upload_archive('vault_name', fake_data, 'linear_hash', 94 'tree_hash') 95 # Verify that we seek back to the original offset after making 96 # a request. This ensures that if we need to resend the request we're 97 # back at the correct location within the file. 98 self.assertEqual(fake_data.tell(), 2) 99