1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
|
#!/usr/bin/env python
# Copyright 2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
import base64
import datetime
import awscli.customizations.ec2.bundleinstance
from awscli.compat import StringIO
from awscli.testutils import BaseAWSCommandParamsTest, mock
class TestBundleInstance(BaseAWSCommandParamsTest):
prefix = 'ec2 bundle-instance'
POLICY = (
b'{"expiration": "2013-08-10T00:00:00.000000Z",'
b'"conditions": [{"bucket": "mybucket"},{"acl": '
b'"ec2-bundle-read"},["starts-with", "$key", "foobar"]]}'
)
POLICY_SIGNATURE = 'ynxybUMv9YuGbPl7HZ8AFJW/2t0='
def setUp(self):
super(TestBundleInstance, self).setUp()
# This mocks out datetime.datetime.utcnow() so that it always
# returns the same datetime object. This is because this value
# is embedded into the policy file that is generated and we
# don't what the policy or its signature to change each time
# we run the test.
self.datetime_patcher = mock.patch.object(
awscli.customizations.ec2.bundleinstance.datetime,
'datetime',
mock.Mock(wraps=datetime.datetime),
)
mocked_datetime = self.datetime_patcher.start()
mocked_datetime.utcnow.return_value = datetime.datetime(2013, 8, 9)
def tearDown(self):
super(TestBundleInstance, self).tearDown()
self.datetime_patcher.stop()
def test_no_policy_provided(self):
args = ' --instance-id i-12345678 --owner-akid AKIAIOSFODNN7EXAMPLE'
args += ' --owner-sak wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY'
args += ' --bucket mybucket --prefix foobar'
args_list = (self.prefix + args).split()
result = {
'InstanceId': 'i-12345678',
'Storage': {
'S3': {
'Bucket': 'mybucket',
'Prefix': 'foobar',
'AWSAccessKeyId': 'AKIAIOSFODNN7EXAMPLE',
'UploadPolicy': self.POLICY,
'UploadPolicySignature': self.POLICY_SIGNATURE,
}
},
}
self.assert_params_for_cmd(args_list, result)
def test_policy_provided(self):
policy = '{"notarealpolicy":true}'
base64policy = (
base64.encodebytes(policy.encode('latin-1'))
.strip()
.decode('utf-8')
)
policy_signature = 'a5SmoLOxoM0MHpOdC25nE7KIafg='
args = ' --instance-id i-12345678 --owner-akid AKIAIOSFODNN7EXAMPLE'
args += ' --owner-sak wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY'
args += ' --bucket mybucket --prefix foobar --policy %s' % base64policy
args_list = (self.prefix + args).split()
result = {
'InstanceId': 'i-12345678',
'Storage': {
'S3': {
'Bucket': 'mybucket',
'Prefix': 'foobar',
'AWSAccessKeyId': 'AKIAIOSFODNN7EXAMPLE',
'UploadPolicy': b'{"notarealpolicy":true}',
'UploadPolicySignature': policy_signature,
}
},
}
self.assert_params_for_cmd(args_list, result)
def test_both(self):
captured = StringIO()
json = """{"S3":{"Bucket":"foobar","Prefix":"fiebaz"}}"""
args = (
' --instance-id i-12345678 --owner-akid blah --owner-sak blah --storage %s'
% json
)
args_list = (self.prefix + args).split()
_, stderr, _ = self.assert_params_for_cmd(args_list, expected_rc=252)
self.assertIn('Mixing the --storage option', stderr)
|
