1# Copyright (c) 2013 Amazon.com, Inc. or its affiliates. All Rights Reserved 2# 3# Permission is hereby granted, free of charge, to any person obtaining a 4# copy of this software and associated documentation files (the 5# "Software"), to deal in the Software without restriction, including 6# without limitation the rights to use, copy, modify, merge, publish, dis- 7# tribute, sublicense, and/or sell copies of the Software, and to permit 8# persons to whom the Software is furnished to do so, subject to the fol- 9# lowing conditions: 10# 11# The above copyright notice and this permission notice shall be included 12# in all copies or substantial portions of the Software. 13# 14# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS 15# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- 16# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT 17# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, 18# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 19# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS 20# IN THE SOFTWARE. 21# 22 23from boto.compat import json 24from boto.kms.layer1 import KMSConnection 25from tests.unit import AWSMockServiceTestCase 26 27 28class TestKinesis(AWSMockServiceTestCase): 29 connection_class = KMSConnection 30 31 def default_body(self): 32 return b'{}' 33 34 def test_binary_input(self): 35 """ 36 This test ensures that binary is base64 encoded when it is sent to 37 the service. 38 """ 39 self.set_http_response(status_code=200) 40 data = b'\x00\x01\x02\x03\x04\x05' 41 self.service_connection.encrypt(key_id='foo', plaintext=data) 42 body = json.loads(self.actual_request.body) 43 self.assertEqual(body['Plaintext'], 'AAECAwQF') 44 45 def test_non_binary_input_for_blobs_fails(self): 46 """ 47 This test ensures that only binary is used for blob type parameters. 48 """ 49 self.set_http_response(status_code=200) 50 data = u'\u00e9' 51 with self.assertRaises(TypeError): 52 self.service_connection.encrypt(key_id='foo', plaintext=data) 53 54 def test_binary_ouput(self): 55 """ 56 This test ensures that the output is base64 decoded before 57 it is returned to the user. 58 """ 59 content = {'Plaintext': 'AAECAwQF'} 60 self.set_http_response(status_code=200, 61 body=json.dumps(content).encode('utf-8')) 62 response = self.service_connection.decrypt(b'some arbitrary value') 63 self.assertEqual(response['Plaintext'], b'\x00\x01\x02\x03\x04\x05') 64