mirror of
https://github.com/encounter/sceptre.git
synced 2026-03-30 11:37:13 -07:00
Sean Rankine
700 lines
28 KiB
Python
700 lines
28 KiB
Python
import logging
|
|
import yaml
|
|
import datetime
|
|
import os
|
|
import errno
|
|
|
|
from click.testing import CliRunner
|
|
from mock import Mock, patch, sentinel
|
|
import pytest
|
|
|
|
import sceptre.cli
|
|
from sceptre.cli import cli
|
|
from sceptre.exceptions import SceptreException
|
|
from sceptre.stack_status import StackStatus
|
|
from sceptre.stack_status import StackChangeSetStatus
|
|
|
|
|
|
class TestCli(object):
|
|
|
|
def setup_method(self, test_method):
|
|
self.runner = CliRunner()
|
|
|
|
@patch("sys.exit")
|
|
def test_catch_excecptions(self, mock_exit):
|
|
@sceptre.cli.catch_exceptions
|
|
def raises_exception():
|
|
raise SceptreException
|
|
|
|
raises_exception()
|
|
mock_exit.assert_called_once_with(1)
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_validate_template(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["validate-template", "dev", "vpc"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].validate_template\
|
|
.assert_called_with()
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_generate_template(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
result = self.runner.invoke(cli, ["generate-template", "dev", "vpc"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
|
|
assert result.output == "{0}\n".format(
|
|
mock_get_env.return_value.stacks["vpc"].template.body
|
|
)
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_lock_stack(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["lock-stack", "dev", "vpc"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].lock\
|
|
.assert_called_with()
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_unlock_stack(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["unlock-stack", "dev", "vpc"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].unlock\
|
|
.assert_called_with()
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_describe_env_resources(self, mock_get_env, mock_getcwd):
|
|
mock_get_env.return_value.describe_resources.return_value = {
|
|
"stack-name-1": {
|
|
"StackResources": [
|
|
{
|
|
"LogicalResourceId": "logical-resource-id",
|
|
"PhysicalResourceId": "physical-resource-id"
|
|
}
|
|
]
|
|
},
|
|
"stack-name-2": {
|
|
"StackResources": [
|
|
{
|
|
"LogicalResourceId": "logical-resource-id",
|
|
"PhysicalResourceId": "physical-resource-id"
|
|
}
|
|
]
|
|
}
|
|
}
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
result = self.runner.invoke(cli, ["describe-env-resources", "dev"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.describe_resources\
|
|
.assert_called_with()
|
|
# Assert that there is output
|
|
assert result.output
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_describe_stack_resources(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
mock_get_env.return_value.stacks["vpc"].describe_resources\
|
|
.return_value = {
|
|
"StackResources": [
|
|
{
|
|
"LogicalResourceId": "logical-resource-id",
|
|
"PhysicalResourceId": "physical-resource-id"
|
|
}
|
|
]
|
|
}
|
|
result = self.runner.invoke(
|
|
cli, ["describe-stack-resources", "dev", "vpc"]
|
|
)
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].describe_resources\
|
|
.assert_called_with()
|
|
# Assert that there is output.
|
|
assert result.output
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_create_stack(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["create-stack", "dev", "vpc"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].create\
|
|
.assert_called_with()
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_delete_stack(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["delete-stack", "dev", "vpc"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].delete\
|
|
.assert_called_with()
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_update_stack(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["update-stack", "dev", "vpc"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].update\
|
|
.assert_called_with()
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_launch_stack(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["launch-stack", "dev", "vpc"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].launch\
|
|
.assert_called_with()
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_launch_env(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["launch-env", "dev"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.launch.assert_called_with()
|
|
|
|
@patch("sceptre.cli.get_env")
|
|
def test_launch_env_returns_zero_correctly(self, mock_get_env):
|
|
mock_get_env.return_value.launch.return_value = dict(
|
|
(sentinel.stack_name, StackStatus.COMPLETE) for _ in range(5)
|
|
)
|
|
result = self.runner.invoke(cli, ["launch-env", "environment"])
|
|
assert result.exit_code == 0
|
|
|
|
@patch("sceptre.cli.get_env")
|
|
def test_launch_env_returns_non_zero_correctly(self, mock_get_env):
|
|
mock_get_env.return_value.launch.return_value = dict(
|
|
(sentinel.stack_name, StackStatus.FAILED) for _ in range(5)
|
|
)
|
|
result = self.runner.invoke(cli, ["launch-env", "environment"])
|
|
assert result.exit_code == 1
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_delete_env(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
mock_get_env.return_value.delete.return_value = \
|
|
sentinel.response
|
|
self.runner.invoke(cli, ["delete-env", "dev"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.delete.assert_called_with()
|
|
|
|
@patch("sceptre.cli.get_env")
|
|
def test_delete_env_returns_zero_correctly(self, mock_get_env):
|
|
mock_get_env.return_value.delete.return_value = dict(
|
|
(sentinel.stack_name, StackStatus.COMPLETE) for _ in range(5)
|
|
)
|
|
result = self.runner.invoke(cli, ["delete-env", "environment"])
|
|
assert result.exit_code == 0
|
|
|
|
@patch("sceptre.cli.get_env")
|
|
def test_delete_env_returns_non_zero_correctly(self, mock_get_env):
|
|
mock_get_env.return_value.delete.return_value = dict(
|
|
(sentinel.stack_name, StackStatus.FAILED) for _ in range(5)
|
|
)
|
|
result = self.runner.invoke(cli, ["delete-env", "environment"])
|
|
assert result.exit_code == 1
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_continue_update_rollback(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["continue-update-rollback", "dev", "vpc"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].\
|
|
continue_update_rollback.assert_called_with()
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_create_change_set(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(
|
|
cli, ["create-change-set", "dev", "vpc", "cs1"]
|
|
)
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].create_change_set\
|
|
.assert_called_with("cs1")
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_delete_change_set(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(
|
|
cli, ["delete-change-set", "dev", "vpc", "cs1"]
|
|
)
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].delete_change_set\
|
|
.assert_called_with("cs1")
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_describe_change_set(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
mock_get_env.return_value.stacks["vpc"].describe_change_set\
|
|
.return_value = {
|
|
"ChangeSetName": "change-set-1",
|
|
"Changes": [
|
|
{
|
|
"ResourceChange": {
|
|
"ResourceType": "AWS::EC2::InternetGateway",
|
|
"Replacement": "True",
|
|
"PhysicalResourceId": "igw-04a59561",
|
|
"Details": [],
|
|
"Action": "Remove",
|
|
"Scope": [],
|
|
"LogicalResourceId": "InternetGateway"
|
|
}
|
|
}
|
|
],
|
|
"CreationTime": "2017-01-20 14:10:25.239000+00:00",
|
|
"ExecutionStatus": "AVAILABLE",
|
|
"StackName": "example-dev-vpc",
|
|
"Status": "CREATE_COMPLETE"
|
|
}
|
|
result = self.runner.invoke(
|
|
cli, ["describe-change-set", "dev", "vpc", "cs1"]
|
|
)
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].describe_change_set\
|
|
.assert_called_with("cs1")
|
|
assert yaml.safe_load(result.output) == {
|
|
"ChangeSetName": "change-set-1",
|
|
"Changes": [
|
|
{
|
|
"ResourceChange": {
|
|
"ResourceType": "AWS::EC2::InternetGateway",
|
|
"Replacement": "True",
|
|
"PhysicalResourceId": "igw-04a59561",
|
|
"Action": "Remove",
|
|
"LogicalResourceId": "InternetGateway",
|
|
"Scope": []
|
|
}
|
|
}
|
|
],
|
|
"CreationTime": "2017-01-20 14:10:25.239000+00:00",
|
|
"ExecutionStatus": "AVAILABLE",
|
|
"StackName": "example-dev-vpc",
|
|
"Status": "CREATE_COMPLETE"
|
|
}
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_describe_change_set_with_verbose_flag(
|
|
self, mock_get_env, mock_getcwd
|
|
):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
mock_get_env.return_value.stacks["vpc"].describe_change_set\
|
|
.return_value = {
|
|
"Changes": [
|
|
{
|
|
"ResourceChange": {
|
|
"ResourceType": "AWS::EC2::InternetGateway",
|
|
"PhysicalResourceId": "igw-04a59561",
|
|
"Details": [],
|
|
"Action": "Remove",
|
|
"Scope": [],
|
|
"LogicalResourceId": "InternetGateway"
|
|
}
|
|
}
|
|
]
|
|
}
|
|
result = self.runner.invoke(
|
|
cli, ["describe-change-set", "--verbose", "dev", "vpc", "cs1"]
|
|
)
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].describe_change_set\
|
|
.assert_called_with("cs1")
|
|
assert yaml.safe_load(result.output) == {
|
|
"Changes": [
|
|
{
|
|
"ResourceChange": {
|
|
"ResourceType": "AWS::EC2::InternetGateway",
|
|
"PhysicalResourceId": "igw-04a59561",
|
|
"Details": [],
|
|
"Action": "Remove",
|
|
"Scope": [],
|
|
"LogicalResourceId": "InternetGateway"
|
|
}
|
|
}
|
|
]
|
|
}
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_execute_change_set(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["execute-change-set", "dev", "vpc", "cs1"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].execute_change_set\
|
|
.assert_called_with("cs1")
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_list_change_sets(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["list-change-sets", "dev", "vpc"])
|
|
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].list_change_sets\
|
|
.assert_called_with()
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.uuid1")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_update_with_change_set_with_input_yes(
|
|
self, mock_get_env, mock_uuid1, mock_getcwd
|
|
):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
mock_get_env.return_value.stacks["vpc"].wait_for_cs_completion\
|
|
.return_value = StackChangeSetStatus.READY
|
|
mock_get_env.return_value.stacks["vpc"].describe_change_set\
|
|
.return_value = "description"
|
|
mock_uuid1().hex = "1"
|
|
self.runner.invoke(
|
|
cli, ["update-stack-cs", "dev", "vpc", "--verbose"], input="y"
|
|
)
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].create_change_set\
|
|
.assert_called_with("change-set-1")
|
|
mock_get_env.return_value.stacks["vpc"].wait_for_cs_completion\
|
|
.assert_called_with("change-set-1")
|
|
mock_get_env.return_value.stacks["vpc"].execute_change_set\
|
|
.assert_called_with("change-set-1")
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli._simplify_change_set_description")
|
|
@patch("sceptre.cli.uuid1")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_update_with_change_set_without_verbose_flag(
|
|
self, mock_get_environment, mock_uuid1,
|
|
mock_simplify_change_set_description, mock_getcwd
|
|
):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
mock_get_environment.return_value.stacks["vpc"].wait_for_cs_completion\
|
|
.return_value = StackChangeSetStatus.READY
|
|
mock_get_environment.return_value.stacks["vpc"].describe_change_set\
|
|
.return_value = "description"
|
|
mock_simplify_change_set_description.return_value = \
|
|
"simplified_description"
|
|
mock_uuid1().hex = "1"
|
|
response = self.runner.invoke(
|
|
cli, ["update-stack-cs", "dev", "vpc"], input="y"
|
|
)
|
|
assert "simplified_description" in response.output
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.uuid1")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_update_with_change_set_with_input_no(
|
|
self, mock_get_env, mock_uuid1, mock_getcwd
|
|
):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
mock_get_env.return_value.stacks["vpc"].wait_for_cs_completion\
|
|
.return_value = StackChangeSetStatus.READY
|
|
mock_get_env.return_value.stacks["vpc"].describe_change_set\
|
|
.return_value = "description"
|
|
mock_uuid1().hex = "1"
|
|
self.runner.invoke(
|
|
cli, ["update-stack-cs", "dev", "vpc", "--verbose"], input="n"
|
|
)
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].create_change_set\
|
|
.assert_called_with("change-set-1")
|
|
mock_get_env.return_value.stacks["vpc"].wait_for_cs_completion\
|
|
.assert_called_with("change-set-1")
|
|
mock_get_env.return_value.stacks["vpc"].delete_change_set\
|
|
.assert_called_with("change-set-1")
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.uuid1")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_update_with_change_set_with_status_defunct(
|
|
self, mock_get_env, mock_uuid1, mock_getcwd
|
|
):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
mock_get_env.return_value.stacks["vpc"].wait_for_cs_completion\
|
|
.return_value = StackChangeSetStatus.DEFUNCT
|
|
mock_get_env.return_value.stacks["vpc"].describe_change_set\
|
|
.return_value = "description"
|
|
mock_uuid1().hex = "1"
|
|
result = self.runner.invoke(
|
|
cli, ["update-stack-cs", "dev", "vpc", "--verbose"]
|
|
)
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].create_change_set\
|
|
.assert_called_with("change-set-1")
|
|
mock_get_env.return_value.stacks["vpc"].wait_for_cs_completion\
|
|
.assert_called_with("change-set-1")
|
|
assert result.exit_code == 1
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_describe_stack_outputs(self, mock_get_env, mock_getcwd):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, ["describe-stack-outputs", "dev", "vpc"])
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value.stacks["vpc"].describe_outputs\
|
|
.assert_called_with()
|
|
|
|
@patch("sceptre.cli.get_env")
|
|
def test_describe_stack_outputs_handles_envvar_flag(
|
|
self, mock_get_env
|
|
):
|
|
mock_get_env.return_value.stacks["vpc"].describe_outputs\
|
|
.return_value = [
|
|
{
|
|
"OutputKey": "key",
|
|
"OutputValue": "value"
|
|
}
|
|
]
|
|
result = self.runner.invoke(
|
|
cli, ["describe-stack-outputs", "--export=envvar", "dev", "vpc"]
|
|
)
|
|
assert result.output == "export SCEPTRE_key=value\n"
|
|
|
|
@patch("sceptre.cli.get_env")
|
|
def test_describe_env(self, mock_get_env):
|
|
mock_Environment = Mock()
|
|
mock_Environment.describe.return_value = {"stack": "status"}
|
|
mock_get_env.return_value = mock_Environment
|
|
|
|
result = self.runner.invoke(cli, ["describe-env", "dev"])
|
|
assert result.output == "stack: status\n\n"
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.get_env")
|
|
def test_set_stack_policy_with_file_flag(
|
|
self, mock_get_env, mock_getcwd
|
|
):
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
self.runner.invoke(cli, [
|
|
"set-stack-policy", "dev", "vpc",
|
|
"--policy-file=tests/fixtures/stack_policies/lock.json"
|
|
])
|
|
mock_Environment = Mock()
|
|
mock_get_env.assert_called_with(sentinel.cwd, "dev", {})
|
|
mock_get_env.return_value = mock_Environment
|
|
|
|
@patch("sceptre.cli.get_env")
|
|
def test_get_stack_policy_with_existing_policy(self, mock_get_env):
|
|
mock_get_env.return_value.stacks["vpc"].get_policy\
|
|
.return_value = {
|
|
"StackPolicyBody": "policy"
|
|
}
|
|
|
|
result = self.runner.invoke(cli, ["get-stack-policy", "dev", "vpc"])
|
|
assert result.output == "policy\n"
|
|
|
|
@patch("sceptre.cli.get_env")
|
|
def test_get_stack_policy_without_existing_policy(
|
|
self, mock_get_env
|
|
):
|
|
mock_get_env.return_value.stacks["vpc"].get_policy\
|
|
.return_value = {}
|
|
|
|
result = self.runner.invoke(cli, ["get-stack-policy", "dev", "vpc"])
|
|
assert result.output == "{}\n"
|
|
|
|
@patch("sceptre.cli.os.getcwd")
|
|
@patch("sceptre.cli.Environment")
|
|
def test_get_env(self, mock_Environment, mock_getcwd):
|
|
mock_Environment.return_value = sentinel.environment
|
|
mock_getcwd.return_value = sentinel.cwd
|
|
response = sceptre.cli.get_env(
|
|
sentinel.cwd, sentinel.environment_path, sentinel.options
|
|
)
|
|
mock_Environment.assert_called_once_with(
|
|
sceptre_dir=sentinel.cwd,
|
|
environment_path=sentinel.environment_path,
|
|
options=sentinel.options
|
|
)
|
|
assert response == sentinel.environment
|
|
|
|
def test_init_project_non_existant(self):
|
|
with self.runner.isolated_filesystem():
|
|
sceptre_dir = os.path.abspath('./example')
|
|
config_dir = os.path.join(sceptre_dir, "config")
|
|
template_dir = os.path.join(sceptre_dir, "templates")
|
|
region = "test-region"
|
|
os.environ["AWS_DEFAULT_REGION"] = region
|
|
defaults = {
|
|
"project_code": "example",
|
|
"region": region
|
|
}
|
|
|
|
result = self.runner.invoke(cli, ["init", "project", "example"])
|
|
assert not result.exception
|
|
assert os.path.isdir(config_dir)
|
|
assert os.path.isdir(template_dir)
|
|
|
|
with open(os.path.join(config_dir, "config.yaml")) as config_file:
|
|
config = yaml.load(config_file)
|
|
|
|
assert config == defaults
|
|
|
|
def test_init_project_already_exist(self):
|
|
with self.runner.isolated_filesystem():
|
|
sceptre_dir = os.path.abspath('./example')
|
|
config_dir = os.path.join(sceptre_dir, "config")
|
|
template_dir = os.path.join(sceptre_dir, "templates")
|
|
existing_config = {"Test": "Test"}
|
|
|
|
os.mkdir(sceptre_dir)
|
|
os.mkdir(config_dir)
|
|
os.mkdir(template_dir)
|
|
|
|
config_filepath = os.path.join(config_dir, "config.yaml")
|
|
with open(config_filepath, 'w') as config_file:
|
|
yaml.dump(existing_config, config_file)
|
|
|
|
result = self.runner.invoke(cli, ["init", "project", "example"])
|
|
assert result.exit_code == 1
|
|
assert result.output == 'Folder \"example\" already exists.\n'
|
|
assert os.path.isdir(config_dir)
|
|
assert os.path.isdir(template_dir)
|
|
|
|
with open(os.path.join(config_dir, "config.yaml")) as config_file:
|
|
config = yaml.load(config_file)
|
|
assert existing_config == config
|
|
|
|
@pytest.mark.parametrize("environment,config_structure,stdin,result", [
|
|
(
|
|
"A",
|
|
{"": {}},
|
|
'y\nA\nA\n', {"project_code": "A", "region": "A"}
|
|
),
|
|
(
|
|
"A",
|
|
{"": {"project_code": "top", "region": "top"}},
|
|
'y\n\n\n', {}
|
|
),
|
|
(
|
|
"A",
|
|
{"": {"project_code": "top", "region": "top"}},
|
|
'y\nA\nA\n', {"project_code": "A", "region": "A"}
|
|
),
|
|
(
|
|
"A/A",
|
|
{
|
|
"": {"project_code": "top", "region": "top"},
|
|
"A": {"project_code": "A", "region": "A"},
|
|
},
|
|
'y\nA/A\nA/A\n', {"project_code": "A/A", "region": "A/A"}
|
|
),
|
|
(
|
|
"A/A",
|
|
{
|
|
"": {"project_code": "top", "region": "top"},
|
|
"A": {"project_code": "A", "region": "A"},
|
|
},
|
|
'y\nA\nA\n', {}
|
|
)
|
|
])
|
|
def test_init_environment(
|
|
self, environment, config_structure, stdin, result
|
|
):
|
|
with self.runner.isolated_filesystem():
|
|
sceptre_dir = os.path.abspath('./example')
|
|
config_dir = os.path.join(sceptre_dir, "config")
|
|
os.makedirs(config_dir)
|
|
|
|
env_dir = os.path.join(sceptre_dir, "config", environment)
|
|
for env_path, config in config_structure.items():
|
|
path = os.path.join(config_dir, env_path)
|
|
try:
|
|
os.makedirs(path)
|
|
except OSError as e:
|
|
if e.errno == errno.EEXIST and os.path.isdir(path):
|
|
pass
|
|
else:
|
|
raise
|
|
|
|
filepath = os.path.join(path, "config.yaml")
|
|
with open(filepath, 'w') as config_file:
|
|
yaml.safe_dump(
|
|
config, stream=config_file, default_flow_style=False
|
|
)
|
|
|
|
os.chdir(sceptre_dir)
|
|
|
|
cmd_result = self.runner.invoke(
|
|
cli, ["init", "env", environment],
|
|
input=stdin
|
|
)
|
|
|
|
if result:
|
|
with open(os.path.join(env_dir, "config.yaml")) as config_file:
|
|
config = yaml.load(config_file)
|
|
assert config == result
|
|
else:
|
|
assert cmd_result.output.endswith(
|
|
"No config.yaml file needed - covered by parent config.\n"
|
|
)
|
|
|
|
def test_setup_logging_with_debug(self):
|
|
logger = sceptre.cli.setup_logging(True, False)
|
|
assert logger.getEffectiveLevel() == logging.DEBUG
|
|
assert logging.getLogger("botocore").getEffectiveLevel() == \
|
|
logging.INFO
|
|
|
|
# Silence logging for the rest of the tests
|
|
logger.setLevel(logging.CRITICAL)
|
|
|
|
def test_setup_logging_without_debug(self):
|
|
logger = sceptre.cli.setup_logging(False, False)
|
|
assert logger.getEffectiveLevel() == logging.INFO
|
|
assert logging.getLogger("botocore").getEffectiveLevel() == \
|
|
logging.CRITICAL
|
|
|
|
# Silence logging for the rest of the tests
|
|
logger.setLevel(logging.CRITICAL)
|
|
|
|
@patch("sceptre.cli.click.echo")
|
|
def test_write_with_yaml_format(self, mock_echo):
|
|
sceptre.cli.write({"key": "value"}, "yaml")
|
|
mock_echo.assert_called_once_with("key: value\n")
|
|
|
|
@patch("sceptre.cli.click.echo")
|
|
def test_write_with_json_format(self, mock_echo):
|
|
sceptre.cli.write({"key": "value"}, "json")
|
|
mock_echo.assert_called_once_with('{"key": "value"}')
|
|
|
|
@patch("sceptre.cli.click.echo")
|
|
def test_write_status_with_colour(self, mock_echo):
|
|
sceptre.cli.write("stack: CREATE_COMPLETE", no_colour=False)
|
|
mock_echo.assert_called_once_with(
|
|
"stack: \x1b[32mCREATE_COMPLETE\x1b[0m"
|
|
)
|
|
|
|
@patch("sceptre.cli.click.echo")
|
|
def test_write_status_without_colour(self, mock_echo):
|
|
sceptre.cli.write("stack: CREATE_COMPLETE", no_colour=True)
|
|
mock_echo.assert_called_once_with("stack: CREATE_COMPLETE")
|
|
|
|
@patch("sceptre.cli.StackStatusColourer.colour")
|
|
@patch("sceptre.cli.Formatter.format")
|
|
def test_ColouredFormatter_format_with_string(
|
|
self, mock_format, mock_colour
|
|
):
|
|
mock_format.return_value = sentinel.response
|
|
mock_colour.return_value = sentinel.coloured_response
|
|
coloured_formatter = sceptre.cli.ColouredFormatter()
|
|
response = coloured_formatter.format("string")
|
|
mock_format.assert_called_once_with("string")
|
|
mock_colour.assert_called_once_with(sentinel.response)
|
|
assert response == sentinel.coloured_response
|
|
|
|
def test_CustomJsonEncoder_with_non_json_serialisable_object(self):
|
|
encoder = sceptre.cli.CustomJsonEncoder()
|
|
response = encoder.encode(datetime.datetime(2016, 5, 3))
|
|
assert response == '"2016-05-03 00:00:00"'
|