Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
109 changes: 109 additions & 0 deletions tests/integration/local/invoke/test_integrations_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import os
import copy
from unittest import skipIf
import tempfile

from nose_parameterized import parameterized
from subprocess import Popen, PIPE
Expand Down Expand Up @@ -233,6 +234,114 @@ def test_invoke_with_docker_network_of_host(self):
self.assertEquals(return_code, 0)


class TestUsingConfigFiles(InvokeIntegBase):
template = Path("template.yml")

def setUp(self):
self.config_dir = tempfile.mkdtemp()

def tearDown(self):
shutil.rmtree(self.config_dir, ignore_errors=True)

def test_default_profile_with_custom_configs(self):
profile = "default"
custom_config = self._create_config_file(profile)
custom_cred = self._create_cred_file(profile)

command_list = self.get_command_list("EchoEnvWithParameters",
template_path=self.template_path,
event_path=self.event_path)

env = os.environ.copy()
env.pop('AWS_DEFAULT_REGION', None)
env['AWS_CONFIG_FILE'] = custom_config
env['AWS_SHARED_CREDENTIALS_FILE'] = custom_cred

process = Popen(command_list, stdout=PIPE, env=env)
process.wait()
process_stdout = b"".join(process.stdout.readlines()).strip()
environ = json.loads(process_stdout.decode('utf-8'))

self.assertEquals(environ["AWS_DEFAULT_REGION"], 'us-west-1')
self.assertEquals(environ["AWS_REGION"], 'us-west-1')
self.assertEquals(environ["AWS_ACCESS_KEY_ID"], 'someaccesskeyid')
self.assertEquals(environ["AWS_SECRET_ACCESS_KEY"], 'shhhhhthisisasecret')
self.assertEquals(environ["AWS_SESSION_TOKEN"], 'sessiontoken')

def test_custom_profile_with_custom_configs(self):
custom_config = self._create_config_file("custom")
custom_cred = self._create_cred_file("custom")

command_list = self.get_command_list("EchoEnvWithParameters",
template_path=self.template_path,
event_path=self.event_path,
profile='custom')

env = os.environ.copy()
env.pop('AWS_DEFAULT_REGION', None)
env['AWS_CONFIG_FILE'] = custom_config
env['AWS_SHARED_CREDENTIALS_FILE'] = custom_cred

process = Popen(command_list, stdout=PIPE, env=env)
process.wait()
process_stdout = b"".join(process.stdout.readlines()).strip()
environ = json.loads(process_stdout.decode('utf-8'))

self.assertEquals(environ["AWS_DEFAULT_REGION"], 'us-west-1')
self.assertEquals(environ["AWS_REGION"], 'us-west-1')
self.assertEquals(environ["AWS_ACCESS_KEY_ID"], 'someaccesskeyid')
self.assertEquals(environ["AWS_SECRET_ACCESS_KEY"], 'shhhhhthisisasecret')
self.assertEquals(environ["AWS_SESSION_TOKEN"], 'sessiontoken')

def test_custom_profile_through_envrionment_variables(self):
# When using a custom profile in a custom location, you need both the config
# and credential file otherwise we fail to find a region or the profile (depending
# on which one is provided
custom_config = self._create_config_file("custom")

custom_cred = self._create_cred_file("custom")

command_list = self.get_command_list("EchoEnvWithParameters",
template_path=self.template_path,
event_path=self.event_path)

env = os.environ.copy()
env.pop('AWS_DEFAULT_REGION', None)
env['AWS_CONFIG_FILE'] = custom_config
env['AWS_SHARED_CREDENTIALS_FILE'] = custom_cred
env['AWS_PROFILE'] = "custom"

process = Popen(command_list, stdout=PIPE, env=env)
process.wait()
process_stdout = b"".join(process.stdout.readlines()).strip()
environ = json.loads(process_stdout.decode('utf-8'))

self.assertEquals(environ["AWS_DEFAULT_REGION"], 'us-west-1')
self.assertEquals(environ["AWS_REGION"], 'us-west-1')
self.assertEquals(environ["AWS_ACCESS_KEY_ID"], 'someaccesskeyid')
self.assertEquals(environ["AWS_SECRET_ACCESS_KEY"], 'shhhhhthisisasecret')
self.assertEquals(environ["AWS_SESSION_TOKEN"], 'sessiontoken')

def _create_config_file(self, profile):
if profile == "default":
config_file_content = "[{}]\noutput = json\nregion = us-west-1".format(profile)
else:
config_file_content = "[profile {}]\noutput = json\nregion = us-west-1".format(profile)

custom_config = os.path.join(self.config_dir, "customconfig")
with open(custom_config, "w") as file:
file.write(config_file_content)
return custom_config

def _create_cred_file(self, profile):
cred_file_content = "[{}]\naws_access_key_id = someaccesskeyid\naws_secret_access_key = shhhhhthisisasecret \
\naws_session_token = sessiontoken".format(profile)
custom_cred = os.path.join(self.config_dir, "customcred")
with open(custom_cred, "w") as file:
file.write(cred_file_content)
return custom_cred


@skipIf(SKIP_LAYERS_TESTS,
"Skip layers tests in Travis only")
class TestLayerVersion(InvokeIntegBase):
Expand Down