Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_execute_plan_no_persist(self):
"""Test execute plan with no persistent graph."""
context = Context(config=self.config)
context.put_persistent_graph = mock.MagicMock()
vpc = Stack(
definition=generate_definition('vpc', 1),
context=context)
bastion = Stack(
definition=generate_definition('bastion', 1, requires=[vpc.name]),
context=context)
calls = []
def _launch_stack(stack, status=None):
calls.append(stack.fqn)
return COMPLETE
graph = Graph.from_steps([Step(vpc, _launch_stack),
Step(bastion, _launch_stack)])
plan = Plan(description="Test", graph=graph, context=context)
plan.execute(walk)
self.assertEqual(calls, ['namespace-vpc.1', 'namespace-bastion.1'])
def test_execute_plan_cancelled(self):
"""Test execute plan cancelled."""
vpc = Stack(
definition=generate_definition('vpc', 1),
context=self.context)
bastion = Stack(
definition=generate_definition('bastion', 1, requires=[vpc.name]),
context=self.context)
calls = []
def fn(stack, status=None):
calls.append(stack.fqn)
if stack.fqn == vpc_step.name:
raise CancelExecution
return COMPLETE
vpc_step = Step(vpc, fn)
bastion_step = Step(bastion, fn)
def test_execute_plan(self):
"""Test execute plan."""
context = Context(config=self.config)
context.put_persistent_graph = mock.MagicMock()
vpc = Stack(
definition=generate_definition('vpc', 1),
context=context)
bastion = Stack(
definition=generate_definition('bastion', 1, requires=[vpc.name]),
context=context)
removed = Stack(
definition=generate_definition('removed', 1, requires=[]),
context=context)
context._persistent_graph = Graph.from_steps([removed])
calls = []
def _launch_stack(stack, status=None):
calls.append(stack.fqn)
return COMPLETE
def _destroy_stack(stack, status=None):
calls.append(stack.fqn)
return COMPLETE
def test_execute_plan_failed(self):
"""Test execute plan failed."""
vpc = Stack(
definition=generate_definition('vpc', 1),
context=self.context)
bastion = Stack(
definition=generate_definition('bastion', 1, requires=[vpc.name]),
context=self.context)
db = Stack(
definition=generate_definition('db', 1),
context=self.context)
calls = []
def fn(stack, status=None):
calls.append(stack.fqn)
if stack.name == vpc_step.name:
return FAILED
return COMPLETE
vpc_step = Step(vpc, fn)
bastion_step = Step(bastion, fn)
db_step = Step(db, fn)
graph = Graph.from_steps([vpc_step, bastion_step, db_step])
def test_build_graph_cyclic_dependencies(self):
"""Test build graph cyclic dependencies."""
vpc = Stack(
definition=generate_definition(
'vpc', 1),
context=self.context)
db = Stack(
definition=generate_definition(
'db', 1, requires=['app.1']),
context=self.context)
app = Stack(
definition=generate_definition(
'app', 1, requires=['db.1']),
context=self.context)
with self.assertRaises(GraphError) as expected:
Graph.from_steps([Step(vpc, None), Step(db, None), Step(app, None)])
message = ("Error detected when adding 'db.1' "
"as a dependency of 'app.1': graph is "
"not acyclic")
self.assertEqual(str(expected.exception), message)
def test_plan_reverse(self):
"""Test plan reverse."""
vpc = Stack(
definition=generate_definition('vpc', 1),
context=self.context)
bastion = Stack(
definition=generate_definition('bastion', 1, requires=[vpc.name]),
context=self.context)
graph = Graph.from_steps([Step(vpc, fn=None), Step(bastion, fn=None)])
plan = Plan(description="Test", graph=graph, reverse=True)
# order is different between python2/3 so can't compare dicts
result_graph_dict = plan.graph.to_dict()
self.assertEqual(set(), result_graph_dict.get('bastion.1'))
self.assertEqual(set(['bastion.1']), result_graph_dict.get('vpc.1'))
def test_execute_plan_filtered(self):
"""Test execute plan filtered."""
vpc = Stack(
definition=generate_definition('vpc', 1),
context=self.context)
db = Stack(
definition=generate_definition('db', 1, requires=[vpc.name]),
context=self.context)
app = Stack(
definition=generate_definition('app', 1, requires=[db.name]),
context=self.context)
calls = []
def fn(stack, status=None):
calls.append(stack.fqn)
return COMPLETE
context = mock.MagicMock()
def test_execute_plan(self):
"""Test execute plan."""
context = Context(config=self.config)
context.put_persistent_graph = mock.MagicMock()
vpc = Stack(
definition=generate_definition('vpc', 1),
context=context)
bastion = Stack(
definition=generate_definition('bastion', 1, requires=[vpc.name]),
context=context)
removed = Stack(
definition=generate_definition('removed', 1, requires=[]),
context=context)
context._persistent_graph = Graph.from_steps([removed])
calls = []
def _launch_stack(stack, status=None):
calls.append(stack.fqn)
return COMPLETE
def _destroy_stack(stack, status=None):
calls.append(stack.fqn)
return COMPLETE
graph = Graph.from_steps([Step(removed, _destroy_stack),
Step(vpc, _launch_stack),
def test_execute_plan_filtered(self):
"""Test execute plan filtered."""
vpc = Stack(
definition=generate_definition('vpc', 1),
context=self.context)
db = Stack(
definition=generate_definition('db', 1, requires=[vpc.name]),
context=self.context)
app = Stack(
definition=generate_definition('app', 1, requires=[db.name]),
context=self.context)
calls = []
def fn(stack, status=None):
calls.append(stack.fqn)
return COMPLETE
context = mock.MagicMock()
context.persistent_graph_locked = False
context.stack_names = ['db.1']
graph = Graph.from_steps([Step(vpc, fn), Step(db, fn), Step(app, fn)])
This function will be ran multiple times until the step
is "done".
watch_func (Callable): an optional function that will be
called to "tail" the step action.
Returns:
:class:`Step`
"""
# pylint: disable=import-outside-toplevel
from runway.cfngin.config import Stack as StackConfig
from runway.cfngin.stack import Stack
stack_def = StackConfig({'name': stack_name,
'requires': requires or []})
stack = Stack(stack_def, context)
return cls(stack, fn=fn, watch_func=watch_func)