Skip to content

Commit

Permalink
Update Build Action tests for external stack changes
Browse files Browse the repository at this point in the history
  • Loading branch information
danielkza committed Oct 14, 2018
1 parent 49e318b commit ee1ac01
Show file tree
Hide file tree
Showing 2 changed files with 25 additions and 42 deletions.
2 changes: 1 addition & 1 deletion stacker/plan.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@


def log_step(step):
msg = "%s: %s" % (step, step.status.name)
msg = "%s: %s" % (step.name, step.status.name)
if step.status.reason:
msg += " (%s)" % (step.status.reason)
color_code = COLOR_CODES.get(step.status.code, 37)
Expand Down
65 changes: 24 additions & 41 deletions stacker/tests/actions/test_build.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,12 +13,14 @@
_handle_missing_parameters,
UsePreviousParameterValue,
)
from stacker.blueprints.base import Blueprint
from stacker.blueprints.variables.types import CFNString
from stacker.context import Context, Config
from stacker.exceptions import StackDidNotChange, StackDoesNotExist
from stacker.providers.base import BaseProvider
from stacker.providers.aws.default import Provider
from stacker.session_cache import get_session
from stacker.stack import Stack
from stacker.status import (
NotSubmittedStatus,
COMPLETE,
Expand All @@ -28,7 +30,11 @@
FAILED
)

from ..factories import MockThreadingEvent, MockProviderBuilder
from ..factories import (
MockThreadingEvent,
MockProviderBuilder,
generate_definition
)


def mock_stack_parameters(parameters):
Expand Down Expand Up @@ -155,22 +161,6 @@ def test_execute_plan_when_outline_not_specified(self):
build_action.run(outline=False)
self.assertEqual(mock_generate_plan().execute.call_count, 1)

def test_should_update(self):
test_scenarios = (
dict(blueprint=None, locked=False, force=False, result=False),
dict(blueprint="BLUEPRINT", locked=False, force=False,
result=True),
dict(blueprint="BLUEPRINT", locked=False, force=True, result=True),
dict(blueprint="BLUEPRINT", locked=True, force=False,
result=False),
dict(blueprint="BLUEPRINT", locked=True, force=True, result=True)
)
for t in test_scenarios:
mock_stack = mock.MagicMock(
["blueprint", "locked", "force", "name"],
name='test-stack', **t)
self.assertEqual(build.should_update(mock_stack), t['result'])

def test_should_ensure_cfn_bucket(self):
test_scenarios = [
dict(outline=False, dump=False, result=True),
Expand All @@ -191,21 +181,18 @@ def test_should_ensure_cfn_bucket(self):
e.args += ("scenario", str(scenario))
raise

def test_should_submit(self):
test_scenarios = (
dict(blueprint=None, enabled=False, result=True),
dict(blueprint="BLUEPRINT", enabled=False, result=False),
dict(blueprint="BLUEPRINT", enabled=True, result=True),
)

for t in test_scenarios:
mock_stack = mock.MagicMock(
["blueprint", "enabled", "name"],
name='test-stack', **t)
self.assertEqual(build.should_submit(mock_stack), t['result'])
class TestLaunchStack(TestBuildAction):
def _get_stack(self):
stack = Stack(definition=generate_definition("vpc", 1),
context=self.context,)

blueprint_mock = mock.patch.object(type(stack), 'blueprint',
spec=Blueprint, rendered='{}')
self.addCleanup(blueprint_mock.stop)
blueprint_mock.start()
return stack

class TestLaunchStack(TestBuildAction):
def setUp(self):
self.context = self._get_context()
self.session = get_session(region=None)
Expand All @@ -215,13 +202,7 @@ def setUp(self):
self.build_action = build.Action(self.context,
provider_builder=provider_builder,
cancel=MockThreadingEvent())

self.stack = mock.MagicMock()
self.stack.region = None
self.stack.name = 'vpc'
self.stack.fqn = 'vpc'
self.stack.blueprint.rendered = '{}'
self.stack.locked = False
self.stack = self._get_stack()
self.stack_status = None

plan = self.build_action._generate_plan()
Expand All @@ -233,14 +214,16 @@ def patch_object(*args, **kwargs):
self.addCleanup(m.stop)
m.start()

def get_stack(name, *args, **kwargs):
if name != self.stack.name or not self.stack_status:
raise StackDoesNotExist(name)
def get_stack(fqn, *args, **kwargs):
if fqn != self.stack.fqn or not self.stack_status:
raise StackDoesNotExist(fqn)

tags = [{'Key': key, 'Value': value}
for (key, value) in self.stack.tags.items()]
return {'StackName': self.stack.name,
'StackStatus': self.stack_status,
'Outputs': [],
'Tags': []}
'Outputs': {},
'Tags': tags}

patch_object(self.provider, 'get_stack', side_effect=get_stack)
patch_object(self.provider, 'update_stack')
Expand Down

0 comments on commit ee1ac01

Please sign in to comment.