CoCalc -- test_globalargs.py

GitHub Repository: aws/aws-cli
Path: blob/develop/tests/unit/customizations/test_globalargs.py
²⁶²⁴ views
1
# Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
2
#
3
# Licensed under the Apache License, Version 2.0 (the "License"). You
4
# may not use this file except in compliance with the License. A copy of
5
# the License is located at
6
#
7
#     http://aws.amazon.com/apache2.0/
8
#
9
# or in the "license" file accompanying this file. This file is
10
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
11
# ANY KIND, either express or implied. See the License for the specific
12
# language governing permissions and limitations under the License.
13
from unittest.mock import patch, Mock
14

15
from botocore.session import get_session
16
from botocore import UNSIGNED
17
import os
18

19
from awscli.testutils import mock, unittest, capture_output
20
from awscli.customizations import globalargs
21

22

23
class FakeParsedArgs(object):
24
    def __init__(self, **kwargs):
25
        self.__dict__.update(kwargs)
26

27
    def __getattr__(self, arg):
28
        return None
29

30

31
class FakeSession(object):
32

33
    def __init__(self, session_vars=None, config_file_vars=None):
34
        if session_vars is None:
35
            session_vars = {}
36
        self.session_var_map = session_vars
37
        if config_file_vars is None:
38
            config_file_vars = {}
39
        self.config_file_vars = config_file_vars
40

41
    def get_config_variable(self, name, methods=('env', 'config'),
42
                            default=None):
43
        value = None
44
        config_name, envvar_name = self.session_var_map[name]
45
        if methods is not None:
46
            if 'env' in methods and value is None:
47
                value = os.environ.get(envvar_name)
48
            if 'config' in methods and value is None:
49
                value = self.config_file_vars.get(config_name)
50
        else:
51
            value = default
52
        return value
53

54

55
class TestGlobalArgsCustomization(unittest.TestCase):
56

57
    def test_parse_query(self):
58
        parsed_args = FakeParsedArgs(query='foo.bar')
59
        globalargs.resolve_types(parsed_args)
60
        # Assert that it looks like a jmespath parsed expression.
61
        self.assertFalse(isinstance(parsed_args.query, str))
62
        self.assertTrue(hasattr(parsed_args.query, 'search'))
63

64
    def test_parse_query_error_message(self):
65
        # Invalid jmespath expression.
66
        parsed_args = FakeParsedArgs(query='foo.bar.')
67
        with self.assertRaises(ValueError):
68
            globalargs.resolve_types(parsed_args)
69
            globalargs.resolve_types('query')
70

71
    def test_parse_verify_ssl_default_value(self):
72
        with mock.patch('os.environ', {}):
73
            parsed_args = FakeParsedArgs(verify_ssl=True, ca_bundle=None)
74
            session_var_map = {'ca_bundle': ('ca_bundle', 'AWS_CA_BUNDLE')}
75
            session = FakeSession(session_vars=session_var_map)
76
            globalargs.resolve_verify_ssl(parsed_args, session)
77
            # None, so that botocore can apply it's default logic.
78
            self.assertIsNone(parsed_args.verify_ssl)
79

80
    def test_parse_verify_ssl_verify_turned_off(self):
81
        with mock.patch('os.environ', {}):
82
            parsed_args = FakeParsedArgs(verify_ssl=False, ca_bundle=None)
83
            session_var_map = {'ca_bundle': ('ca_bundle', 'AWS_CA_BUNDLE')}
84
            session = FakeSession(session_vars=session_var_map)
85
            globalargs.resolve_verify_ssl(parsed_args, session)
86
            self.assertFalse(parsed_args.verify_ssl)
87

88
    def test_cli_overrides_cert_bundle(self):
89
        environ = {}
90
        with mock.patch('os.environ', environ):
91
            parsed_args = FakeParsedArgs(
92
                verify_ssl=True,
93
                ca_bundle='/path/to/cli_bundle.pem')
94
            config_file_vars = {}
95
            session_var_map = {'ca_bundle': ('ca_bundle', 'AWS_CA_BUNDLE')}
96
            session = FakeSession(
97
                session_vars=session_var_map,
98
                config_file_vars=config_file_vars)
99
            globalargs.resolve_verify_ssl(parsed_args, session)
100
            self.assertEqual(parsed_args.verify_ssl, '/path/to/cli_bundle.pem')
101

102
    def test_cli_overrides_env_cert_bundle(self):
103
        environ = {
104
            'AWS_CA_BUNDLE': '/path/to/env_bundle.pem',
105
        }
106
        with mock.patch('os.environ', environ):
107
            parsed_args = FakeParsedArgs(
108
                verify_ssl=True,
109
                ca_bundle='/path/to/cli_bundle.pem')
110
            config_file_vars = {}
111
            session_var_map = {'ca_bundle': ('ca_bundle', 'AWS_CA_BUNDLE')}
112
            session = FakeSession(
113
                session_vars=session_var_map,
114
                config_file_vars=config_file_vars)
115
            globalargs.resolve_verify_ssl(parsed_args, session)
116
            self.assertEqual(parsed_args.verify_ssl, '/path/to/cli_bundle.pem')
117

118
    def test_no_verify_ssl_overrides_cli_cert_bundle(self):
119
        environ = {
120
            'AWS_CA_BUNDLE': '/path/to/env_bundle.pem',
121
        }
122
        with mock.patch('os.environ', environ):
123
            parsed_args = FakeParsedArgs(
124
                verify_ssl=False,
125
                ca_bundle='/path/to/cli_bundle.pem')
126
            config_file_vars = {}
127
            session_var_map = {'ca_bundle': ('ca_bundle', 'AWS_CA_BUNDLE')}
128
            session = FakeSession(
129
                session_vars=session_var_map,
130
                config_file_vars=config_file_vars)
131
            globalargs.resolve_verify_ssl(parsed_args, session)
132
            self.assertFalse(parsed_args.verify_ssl)
133

134
    def test_no_sign_request_if_option_specified(self):
135
        args = FakeParsedArgs(sign_request=False)
136
        session = mock.Mock()
137
        with mock.patch('awscli.customizations.globalargs._update_default_client_config') as mock_update:
138
            globalargs.no_sign_request(args, session)
139
            mock_update.assert_called_once_with(session, 'signature_version', UNSIGNED)
140

141
    def test_request_signed_by_default(self):
142
        args = FakeParsedArgs(sign_request=True)
143
        session = mock.Mock()
144
        globalargs.no_sign_request(args, session)
145
        self.assertFalse(session.register.called)
146

147
    def test_invalid_endpoint_url(self):
148
        # Invalid jmespath expression.
149
        parsed_args = FakeParsedArgs(endpoint_url='missing-scheme.com')
150
        with self.assertRaises(ValueError):
151
            globalargs.resolve_types(parsed_args)
152

153
    def test_valid_endpoint_url(self):
154
        parsed_args = FakeParsedArgs(endpoint_url='http://custom-endpoint.com')
155
        globalargs.resolve_types(parsed_args)
156
        self.assertEqual(parsed_args.endpoint_url,
157
                         'http://custom-endpoint.com')
158

159
    def test_cli_read_timeout(self):
160
        parsed_args = FakeParsedArgs(read_timeout='60')
161
        session = get_session()
162
        globalargs.resolve_cli_read_timeout(parsed_args, session)
163
        self.assertEqual(parsed_args.read_timeout, 60)
164
        self.assertEqual(
165
            session.get_default_client_config().read_timeout, 60)
166

167
    def test_cli_connect_timeout(self):
168
        parsed_args = FakeParsedArgs(connect_timeout='60')
169
        session = get_session()
170
        globalargs.resolve_cli_connect_timeout(parsed_args, session)
171
        self.assertEqual(parsed_args.connect_timeout, 60)
172
        self.assertEqual(
173
            session.get_default_client_config().connect_timeout, 60)
174

175
    def test_cli_read_timeout_for_blocking(self):
176
        parsed_args = FakeParsedArgs(read_timeout='0')
177
        session = get_session()
178
        globalargs.resolve_cli_read_timeout(parsed_args, session)
179
        self.assertEqual(parsed_args.read_timeout, None)
180
        self.assertEqual(
181
            session.get_default_client_config().read_timeout, None)
182

183
    def test_cli_connect_timeout_for_blocking(self):
184
        parsed_args = FakeParsedArgs(connect_timeout='0')
185
        session = get_session()
186
        globalargs.resolve_cli_connect_timeout(parsed_args, session)
187
        self.assertEqual(parsed_args.connect_timeout, None)
188
        self.assertEqual(
189
            session.get_default_client_config().connect_timeout, None)
190

191
    def test_register_feature_id(self):
192
        parsed_args = FakeParsedArgs(v2_debug=True)
193
        session = get_session()
194
        globalargs.detect_migration_breakage(
195
            parsed_args,
196
            session,
197
            [],
198
        )
199
        # Verify the correct feature ID is registered during the
200
        # provide-client-params event.
201
        with (mock.patch('awscli.customizations.globalargs.register_feature_id')
202
              as mock_register_feature_id):
203
            session.emit(
204
                'provide-client-params.s3.ListBuckets',
205
                params={},
206
                model={},
207
            )
208
        mock_register_feature_id.assert_any_call(
209
            'CLI_V1_TO_V2_MIGRATION_DEBUG_MODE'
210
        )
211

212
    def test_ecr_login_v2_debug(self):
213
        parsed_args = FakeParsedArgs(command='ecr', v2_debug=True)
214
        remaining_args = ['get-login']
215
        session = get_session()
216
        with capture_output() as output:
217
            globalargs.detect_migration_breakage(
218
                parsed_args,
219
                session,
220
                remaining_args,
221
            )
222
        # Verify the expected warning is printed
223
        self.assertIn(
224
            'AWS CLI v2 UPGRADE WARNING: The `ecr get-login` command has '
225
            'been removed in AWS CLI v2.',
226
            output.stderr.getvalue()
227
        )
228

229
    def test_ecr_login_v2_debug_env_var(self):
230
        parsed_args = FakeParsedArgs(command='ecr')
231
        remaining_args = ['get-login']
232
        session = get_session()
233
        env = {'AWS_CLI_UPGRADE_DEBUG_MODE': 'true'}
234
        with capture_output() as output:
235
            with mock.patch('os.environ', env):
236
                globalargs.detect_migration_breakage(
237
                    parsed_args,
238
                    session,
239
                    remaining_args,
240
                )
241
        # Verify the expected warning is printed
242
        self.assertIn(
243
            'AWS CLI v2 UPGRADE WARNING: The `ecr get-login` command has '
244
            'been removed in AWS CLI v2.',
245
            output.stderr.getvalue()
246
        )
247

248
    def test_v2_debug_python_utf8_env_var(self):
249
        parsed_args = FakeParsedArgs(v2_debug=True)
250
        session = get_session()
251
        environ = {'PYTHONUTF8': '1'}
252
        with mock.patch('os.environ', environ):
253
            with capture_output() as output:
254
                globalargs.detect_migration_breakage(parsed_args, session, [])
255
                self.assertIn(
256
                    'The AWS CLI v2 does not support The `PYTHONUTF8` and '
257
                    '`PYTHONIOENCODING` environment variables, and instead '
258
                    'uses the `AWS_CLI_FILE_ENCODING` variable',
259
                    output.stderr.getvalue()
260
                )
261

262
    def test_v2_debug_python_utf8_resolved_env_var(self):
263
        parsed_args = FakeParsedArgs(v2_debug=True)
264
        session = get_session()
265
        environ = {'PYTHONUTF8': '1', 'AWS_CLI_FILE_ENCODING': 'UTF-8'}
266
        with mock.patch('os.environ', environ):
267
            with capture_output() as output:
268
                globalargs.detect_migration_breakage(parsed_args, session, [])
269
                self.assertNotIn(
270
                    'AWS CLI v2 UPGRADE WARNING: The PYTHONUTF8 and '
271
                    'PYTHONIOENCODING environment variables are unsupported '
272
                    'in AWS CLI v2.',
273
                    output.stderr.getvalue()
274
                )
275

276
    def test_v2_debug_python_io_encoding_env_var(self):
277
        parsed_args = FakeParsedArgs(v2_debug=True)
278
        session = get_session()
279
        environ = {'PYTHONIOENCODING': 'UTF8'}
280
        with mock.patch('os.environ', environ):
281
            with capture_output() as output:
282
                globalargs.detect_migration_breakage(parsed_args, session, [])
283
                self.assertIn(
284
                    'The AWS CLI v2 does not support The `PYTHONUTF8` and '
285
                    '`PYTHONIOENCODING` environment variables, and instead '
286
                    'uses the `AWS_CLI_FILE_ENCODING` variable',
287
                    output.stderr.getvalue()
288
                )
289

290
    def test_v2_debug_s3_sigv2(self):
291
        parsed_args = FakeParsedArgs(v2_debug=True)
292
        session = get_session()
293
        globalargs.detect_migration_breakage(parsed_args, session, [])
294
        with capture_output() as output:
295
            session.emit(
296
                'choose-signer.s3.*',
297
                signing_name='s3',
298
                region_name='us-west-2',
299
                signature_version='v2',
300
                context={'auth_type': 'v2'},
301
            )
302
        self.assertIn(
303
            'AWS CLI v2 UPGRADE WARNING: The AWS CLI v2 only uses Signature '
304
            'v4 to authenticate Amazon S3 requests.',
305
            output.stderr.getvalue()
306
        )
307

308
    def test_v2_debug_s3_us_east_1(self):
309
        parsed_args = FakeParsedArgs(v2_debug=True, region='us-east-1')
310
        session = get_session()
311
        globalargs.detect_migration_breakage(parsed_args, session, [])
312
        def mock_get(key: str):
313
            if key == 'retries':
314
                return {'invocation-id': '012345'}
315
            return None
316

317
        with capture_output() as output:
318
            mock_request = Mock()
319
            mock_request.url = 'https://s3.amazonaws.com'
320
            mock_request.context.get.side_effect = mock_get
321
            mock_request.headers = {}
322

323
            session.emit(
324
                'request-created.s3.ListBuckets',
325
                request=mock_request,
326
                operation_name='ListBuckets',
327
            )
328
        self.assertIn(
329
            'AWS CLI v2 UPGRADE WARNING: When you configure AWS CLI v2 '
330
            'to use the `us-east-1` region, it uses the true regional '
331
            'endpoint rather than the global endpoint.',
332
            output.stderr.getvalue()
333
        )
334

335
    def test_v2_debug_s3api_us_east_1(self):
336
        parsed_args = FakeParsedArgs(v2_debug=True, region='us-east-1')
337
        session = get_session()
338
        globalargs.detect_migration_breakage(parsed_args, session, [])
339

340
        def mock_get(key: str):
341
            if key == 'retries':
342
                return {'invocation-id': '012345'}
343
            return None
344

345
        with capture_output() as output:
346
            mock_request = Mock()
347
            mock_request.url = 'https://s3.amazonaws.com'
348
            mock_request.context.get.side_effect = mock_get
349
            mock_request.headers = {}
350

351
            session.emit(
352
                'request-created.s3api.ListBuckets',
353
                request=mock_request,
354
                operation_name='ListBuckets',
355
            )
356
        self.assertIn(
357
            'AWS CLI v2 UPGRADE WARNING: When you configure AWS CLI v2 '
358
            'to use the `us-east-1` region, it uses the true regional '
359
            'endpoint rather than the global endpoint.',
360
            output.stderr.getvalue()
361
        )
362
Product

Resources

Company
