def test_encoding_notification_messages(self):
    key = 'a@b'
    bucket_name = ('notif-enc-%s' % short_uid())
    queue_url = self.sqs_client.create_queue(QueueName='testQueue')['QueueUrl']
    queue_attributes = self.sqs_client.get_queue_attributes(QueueUrl=queue_url, AttributeNames=['QueueArn'])
    self._create_test_notification_bucket(queue_attributes, bucket_name=bucket_name)
    self.s3_client.put_object(Bucket=bucket_name, Key=key, Body='something')
    response = self.sqs_client.receive_message(QueueUrl=queue_url)
    self.assertEqual(json.loads(response['Messages'][0]['Body'])['Records'][0]['s3']['object']['key'], 'a%40b')
    self.s3_client.delete_objects(Bucket=bucket_name, Delete={'Objects': [{'Key': key}]})
def test_encoding_notification_messages(self):
    key = 'a@b'
    bucket_name = ('notif-enc-%s' % short_uid())
    queue_url = self.sqs_client.create_queue(QueueName='testQueue')['QueueUrl']
    queue_attributes = self.sqs_client.get_queue_attributes(QueueUrl=queue_url, AttributeNames=['QueueArn'])
    self._create_test_notification_bucket(queue_attributes, bucket_name=bucket_name)
    self.s3_client.put_object(Bucket=bucket_name, Key=key, Body='something')
    response = self.sqs_client.receive_message(QueueUrl=queue_url)
    self.assertEqual(json.loads(response['Messages'][0]['Body'])['Records'][0]['s3']['object']['key'], 'a%40b')
    self.s3_client.delete_objects(Bucket=bucket_name, Delete={'Objects': [{'Key': key}]})