Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
env_vars['AWS_CONFIG_FILE'] = 'does-not-exist-foo'
f.write(
'[from_env_var]\n'
'aws_access_key_id=enva\n'
'aws_secret_access_key=envb\n'
'\n'
'[from_argument]\n'
'aws_access_key_id=proa\n'
'aws_secret_access_key=prob\n'
)
f.flush()
# Now we set the current profile via env var:
env_vars['AWS_PROFILE'] = 'from_env_var'
# If we specify the --profile argument, that
# value should win over the AWS_PROFILE env var.
p = aws('configure list --profile from_argument',
env_vars=env_vars)
# 1. We should see the profile name being set.
self.assertIn('from_argument', p.stdout)
# 2. The creds should be profa/profb, which come
# from the "from_argument" profile.
self.assertIn('proa', p.stdout)
self.assertIn('prob', p.stdout)
def test_traceback_printed_when_debug_on(self):
p = aws('ec2 describe-instances --filters BADKEY=foo --debug')
self.assertIn('Traceback (most recent call last):', p.stderr, p.stderr)
# Also should see DEBUG statements:
self.assertIn('DEBUG', p.stderr, p.stderr)
def test_describe_instances_with_filter(self):
command = self.prefix + ' --filters Name=instance-id,Values='
command += 'malformed-id'
result = aws(command)
reservations = result.json["Reservations"]
self.assertEqual(len(reservations), 0)
def test_query(self):
bucket_name = self.create_bucket()
self.put_object(bucket_name, 'foo.txt', contents=b'bar')
p = aws('s3api list-objects --bucket %s '
'--query Contents[].Key --output json' % bucket_name)
self.assert_no_errors(p)
response_json = p.json
self.assertEqual(response_json, ['foo.txt'])
def test_set_with_a_url(self):
p = aws('configure set endpoint http://www.example.com',
env_vars=self.env_vars)
self.assert_no_errors(p)
self.assertEqual(
'[default]\n'
'endpoint = http://www.example.com\n',
self.get_config_file_contents())
def test_get_nested_attribute(self):
self.set_config_file_contents(
'[default]\n'
's3 =\n'
' signature_version = v4\n'
)
p = aws('configure get default.s3.signature_version',
env_vars=self.env_vars)
self.assertEqual(p.stdout.strip(), 'v4')
p = aws('configure get default.bad.doesnotexist',
env_vars=self.env_vars)
self.assertEqual(p.rc, 1)
self.assertEqual(p.stdout, '')
def test_cli_input_json_exta_args(self):
# Check that the object can be found.
p = aws('s3api head-object --cli-input-json file://%s --region %s'
% (self.temp_file, self.region))
self.assertEqual(p.rc, 0)
# Override the ``key`` argument. Should produce a failure because
# the key ``bar`` does not exist.
p = aws('s3api head-object --key bar --cli-input-json file://%s '
'--region %s'
% (self.temp_file, self.region))
self.assertEqual(p.rc, 255)
self.assertIn('Not Found', p.stderr)
def test_describe_snapshot_with_snapshot_id(self):
command = self.prefix + ' --snapshot-ids malformed-id'
result = aws(command)
self.assertIn('InvalidParameterValue', result.stderr)
def assert_s3_read_only_profile(self, profile_name):
# Calls to S3 should succeed
command = 's3api list-buckets --profile %s' % profile_name
result = aws(command, env_vars=self.environ)
self.assertEqual(result.rc, 0, result.stderr)
# Calls to other services should not
command = 'iam list-groups --profile %s' % profile_name
result = aws(command, env_vars=self.environ)
self.assertNotEqual(result.rc, 0, result.stdout)
self.assertIn('AccessDenied', result.stderr)
def test_describe_snapshots_with_filter(self):
command = self.prefix
command += ' --filters Name=snapshot-id,Values=malformed-id'
result = aws(command)
snapshots = result.json['Snapshots']
self.assertEqual(len(snapshots), 0)