text: import sys
  import unittest
  from datetime import datetime
  
  import mock
  
  from libcloud.common.aws import AWSRequestSignerAlgorithmV4
  from libcloud.common.aws import SignedAWSConnection
  from libcloud.common.aws import UNSIGNED_PAYLOAD
  from libcloud.test import LibcloudTestCase
  
  
  class EC2MockDriver(object):
      region_name = 'my_region'
  
  
  class AWSRequestSignerAlgorithmV4TestCase(LibcloudTestCase):
  
      def setUp(self):
          SignedAWSConnection.driver = EC2MockDriver()
          SignedAWSConnection.service_name = 'my_service'
          SignedAWSConnection.version = '2013-10-15'
          self.connection = SignedAWSConnection('my_key', 'my_secret')
  
          self.signer = AWSRequestSignerAlgorithmV4(access_key='my_key',
                                                    access_secret='my_secret',
                                                    version='2013-10-15',
                                                    connection=self.connection)
  
          SignedAWSConnection.action = '/my_action/'
          SignedAWSConnection.driver = EC2MockDriver()
  
          self.now = datetime(2015, 3, 4, hour=17, minute=34, second=52)
  
      def test_v4_signature(self):
          params = {
              'Action': 'DescribeInstances',
              'Version': '2013-10-15'
          }
          headers = {
              'Host': 'ec2.eu-west-1.amazonaws.com',
              'Accept-Encoding': 'gzip,deflate',
              'X-AMZ-Date': '20150304T173452Z',
              'User-Agent': 'libcloud/0.17.0 (Amazon EC2 (eu-central-1)) '
          }
          dt = self.now
          sig = self.signer._get_authorization_v4_header(params=params,
                                                         headers=headers,
                                                         dt=dt,
                                                         method='GET',
                                                         path='/my_action/')
          self.assertEqual(sig, 'AWS4-HMAC-SHA256 '
                                'Credential=my_key/20150304/my_region/my_service/aws4_request, '
                                'SignedHeaders=accept-encoding;host;user-agent;x-amz-date, '
                                'Signature=f9868f8414b3c3f856c7955019cc1691265541f5162b9b772d26044280d39bd3')
  
      def test_v4_signature_contains_user_id(self):
          sig = self.signer._get_authorization_v4_header(params={}, headers={},
                                                         dt=self.now)
          self.assertIn('Credential=my_key/', sig)
  
      def test_v4_signature_contains_credential_scope(self):
          with mock.patch('libcloud.common.aws.AWSRequestSignerAlgorithmV4._get_credential_scope') as mock_get_creds:
              mock_get_creds.return_value = 'my_credential_scope'
              sig = self.signer._get_authorization_v4_header(params={}, headers={}, dt=self.now)
  
          self.assertIn('Credential=my_key/my_credential_scope, ', sig)
  
      def test_v4_signature_contains_signed_headers(self):
          with mock.patch('libcloud.common.aws.AWSRequestSignerAlgorithmV4._get_signed_headers') as mock_get_headers:
              mock_get_headers.return_value = 'my_signed_headers'
              sig = self.signer._get_authorization_v4_header({}, {}, self.now,
                                                             method='GET',
                                                             path='/')
          self.assertIn('SignedHeaders=my_signed_headers, ', sig)
  
      def test_v4_signature_contains_signature(self):
          with mock.patch('libcloud.common.aws.AWSRequestSignerAlgorithmV4._get_signature') as mock_get_signature:
              mock_get_signature.return_value = 'my_signature'
              sig = self.signer._get_authorization_v4_header({}, {}, self.now)
          self.assertIn('Signature=my_signature', sig)
  
      def test_get_signature_(self):
          def _sign(key, msg, hex=False):
              if hex:
                  return 'H|%s|%s' % (key, msg)
              else:
                  return '%s|%s' % (key, msg)
  
          with mock.patch('libcloud.common.aws.AWSRequestSignerAlgorithmV4._get_key_to_sign_with') as mock_get_key:
              with mock.patch('libcloud.common.aws.AWSRequestSignerAlgorithmV4._get_string_to_sign') as mock_get_string:
                  with mock.patch('libcloud.common.aws._sign', new=_sign):
                      mock_get_key.return_value = 'my_signing_key'
                      mock_get_string.return_value = 'my_string_to_sign'
                      sig = self.signer._get_signature({}, {}, self.now,
                                                       method='GET', path='/', data=None)
  
          self.assertEqual(sig, 'H|my_signing_key|my_string_to_sign')
  
      def test_get_string_to_sign(self):
          with mock.patch('hashlib.sha256') as mock_sha256:
              mock_sha256.return_value.hexdigest.return_value = 'chksum_of_canonical_request'
              to_sign = self.signer._get_string_to_sign({}, {}, self.now,
                                                        method='GET', path='/', data=None)
  
          self.assertEqual(to_sign,
                           'AWS4-HMAC-SHA256\n'
                           '20150304T173452Z\n'
                           '20150304/my_region/my_service/aws4_request\n'
                           'chksum_of_canonical_request')
  
      def test_get_key_to_sign_with(self):
          def _sign(key, msg, hex=False):
              return '%s|%s' % (key, msg)
  
          with mock.patch('libcloud.common.aws._sign', new=_sign):
              key = self.signer._get_key_to_sign_with(self.now)
  
          self.assertEqual(key, 'AWS4my_secret|20150304|my_region|my_service|aws4_request')
  
      def test_get_signed_headers_contains_all_headers_lowercased(self):
          headers = {'Content-Type': 'text/plain', 'Host': 'my_host', 'X-Special-Header': ''}
          signed_headers = self.signer._get_signed_headers(headers)
  
          self.assertIn('content-type', signed_headers)
          self.assertIn('host', signed_headers)
          self.assertIn('x-special-header', signed_headers)
  
      def test_get_signed_headers_concats_headers_sorted_lexically(self):
          headers = {'Host': 'my_host', 'X-Special-Header': '', '1St-Header': '2', 'Content-Type': 'text/plain'}
          signed_headers = self.signer._get_signed_headers(headers)
  
          self.assertEqual(signed_headers, '1st-header;content-type;host;x-special-header')
  
      def test_get_credential_scope(self):
          scope = self.signer._get_credential_scope(self.now)
          self.assertEqual(scope, '20150304/my_region/my_service/aws4_request')
  
      def test_get_canonical_headers_joins_all_headers(self):
          headers = {
              'accept-encoding': 'gzip,deflate',
              'host': 'my_host',
          }
          self.assertEqual(self.signer._get_canonical_headers(headers),
                           'accept-encoding:gzip,deflate\n'
                           'host:my_host\n')
  
      def test_get_canonical_headers_sorts_headers_lexically(self):
          headers = {
              'accept-encoding': 'gzip,deflate',
              'host': 'my_host',
              '1st-header': '2',
              'x-amz-date': '20150304T173452Z',
              'user-agent': 'my-ua'
          }
          self.assertEqual(self.signer._get_canonical_headers(headers),
                           '1st-header:2\n'
                           'accept-encoding:gzip,deflate\n'
                           'host:my_host\n'
                           'user-agent:my-ua\n'
                           'x-amz-date:20150304T173452Z\n')
  
      def test_get_canonical_headers_lowercases_headers_names(self):
          headers = {
              'Accept-Encoding': 'GZIP,DEFLATE',
              'User-Agent': 'My-UA'
          }
          self.assertEqual(self.signer._get_canonical_headers(headers),
                           'accept-encoding:GZIP,DEFLATE\n'
                           'user-agent:My-UA\n')
  
      def test_get_canonical_headers_trims_header_values(self):
          # TODO: according to AWS spec (and RFC 2616 Section 4.2.) excess whitespace
          # from inside non-quoted strings should be stripped. Now we only strip the
          # start and end of the string. See
          # http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
          headers = {
              'accept-encoding': '   gzip,deflate',
              'user-agent': 'libcloud/0.17.0 '
          }
          self.assertEqual(self.signer._get_canonical_headers(headers),
                           'accept-encoding:gzip,deflate\n'
                           'user-agent:libcloud/0.17.0\n')
  
      def test_get_request_params_joins_params_sorted_lexically(self):
          self.assertEqual(self.signer._get_request_params({
              'Action': 'DescribeInstances',
              'Filter.1.Name': 'state',
              'Version': '2013-10-15'
          }),
              'Action=DescribeInstances&Filter.1.Name=state&Version=2013-10-15')
  
      def test_get_canonical_headers_allow_numeric_header_value(self):
          headers = {
              'Accept-Encoding': 'gzip,deflate',
              'Content-Length': 314
          }
          self.assertEqual(self.signer._get_canonical_headers(headers),
                           'accept-encoding:gzip,deflate\n'
                           'content-length:314\n')
  
      def test_get_request_params_allows_integers_as_value(self):
          self.assertEqual(self.signer._get_request_params({'Action': 'DescribeInstances', 'Port': 22}),
                           'Action=DescribeInstances&Port=22')
  
      def test_get_request_params_urlquotes_params_keys(self):
          self.assertEqual(self.signer._get_request_params({'Action+Reaction': 'DescribeInstances'}),
                           'Action%2BReaction=DescribeInstances')
  
      def test_get_request_params_urlquotes_params_values(self):
          self.assertEqual(self.signer._get_request_params({
              'Action': 'DescribeInstances&Addresses',
              'Port-Range': '2000 3000'
          }),
              'Action=DescribeInstances%26Addresses&Port-Range=2000%203000')
  
      def test_get_request_params_urlquotes_params_values_allows_safe_chars_in_value(self):
          # http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html
          self.assertEqual('Action=a~b.c_d-e',
                           self.signer._get_request_params({'Action': 'a~b.c_d-e'}))
  
      def test_get_payload_hash_returns_digest_of_empty_string_for_GET_requests(self):
          SignedAWSConnection.method = 'GET'
          self.assertEqual(self.signer._get_payload_hash(method='GET'),
                           'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855')
  
      def test_get_payload_hash_with_data_for_PUT_requests(self):
          SignedAWSConnection.method = 'PUT'
          self.assertEqual(self.signer._get_payload_hash(method='PUT', data='DUMMY'),
                           'ceec12762e66397b56dad64fd270bb3d694c78fb9cd665354383c0626dbab013')
  
      def test_get_payload_hash_with_empty_data_for_POST_requests(self):
          SignedAWSConnection.method = 'POST'
          self.assertEqual(self.signer._get_payload_hash(method='POST'),
                           UNSIGNED_PAYLOAD)
  
      def test_get_canonical_request(self):
          req = self.signer._get_canonical_request(
              {'Action': 'DescribeInstances', 'Version': '2013-10-15'},
              {'Accept-Encoding': 'gzip,deflate', 'User-Agent': 'My-UA'},
              method='GET',
              path='/my_action/',
              data=None
          )
          self.assertEqual(req, 'GET\n'
                                '/my_action/\n'
                                'Action=DescribeInstances&Version=2013-10-15\n'
                                'accept-encoding:gzip,deflate\n'
                                'user-agent:My-UA\n'
                                '\n'
                                'accept-encoding;user-agent\n'
                                'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855')
  
      def test_post_canonical_request(self):
          req = self.signer._get_canonical_request(
              {'Action': 'DescribeInstances', 'Version': '2013-10-15'},
              {'Accept-Encoding': 'gzip,deflate', 'User-Agent': 'My-UA'},
              method='POST',
              path='/my_action/',
              data='{}'
          )
          self.assertEqual(req, 'POST\n'
                                '/my_action/\n'
                                'Action=DescribeInstances&Version=2013-10-15\n'
                                'accept-encoding:gzip,deflate\n'
                                'user-agent:My-UA\n'
                                '\n'
                                'accept-encoding;user-agent\n'
                                '44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a')
  
  if __name__ == '__main__':
      sys.exit(unittest.main())
  