How to use the runway.cfngin.plan.Step function in runway

"""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)

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])
        plan = Plan(description="Test", graph=graph)
        with self.assertRaises(PlanFailed):
            plan.execute(walk)

        calls.sort()

        self.assertEqual(calls, ['namespace-db.1', 'namespace-vpc.1'])

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'])
        context.put_persistent_graph.assert_not_called()

"""
        vpc = Stack(
            definition=generate_definition('vpc', 1),
            context=self.context)
        bastion = Stack(
            definition=generate_definition('bastion', 1, requires=[vpc.name]),
            locked=True,
            context=self.context)

        calls = []

        def fn(stack, status=None):
            calls.append(stack.fqn)
            return COMPLETE

        graph = Graph.from_steps([Step(vpc, fn), Step(bastion, fn)])
        plan = Plan(description="Test", graph=graph)
        plan.execute(walk)

        self.assertEqual(calls, ['namespace-vpc.1', 'namespace-bastion.1'])

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)

        graph = Graph.from_steps([vpc_step, bastion_step])
        plan = Plan(description="Test", graph=graph)
        plan.execute(walk)

        self.assertEqual(calls, ['namespace-vpc.1', 'namespace-bastion.1'])

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),
                                  Step(bastion, _launch_stack)])
        plan = Plan(description="Test", graph=graph, context=context)
        plan.context._persistent_graph_lock_code = plan.lock_code
        plan.execute(walk)

        # the order these are appended changes between python2/3
        self.assertIn('namespace-vpc.1', calls)
        self.assertIn('namespace-bastion.1', calls)
        self.assertIn('namespace-removed.1', calls)
        context.put_persistent_graph.assert_called()

        # order is different between python2/3 so can't compare dicts
        result_graph_dict = context.persistent_graph.to_dict()
        self.assertEqual(2, len(result_graph_dict))
        self.assertEqual(set(), result_graph_dict.get('vpc.1'))

def test_plan_targeted(self):
        """Test plan targeted."""
        context = Context(config=self.config)
        vpc = Stack(
            definition=generate_definition('vpc', 1),
            context=context)
        bastion = Stack(
            definition=generate_definition('bastion', 1, requires=[vpc.name]),
            context=context)
        context.stack_names = [vpc.name]
        graph = Graph.from_steps([Step(vpc, fn=None), Step(bastion, fn=None)])
        plan = Plan(description="Test", graph=graph, context=context)

        self.assertEqual({vpc.name: set()}, plan.graph.to_dict())

for ind_node, dep_nodes in persist_graph.dag.graph.items():
            if ind_node not in config_stack_names:
                inverse_steps.append(
                    Step.from_stack_name(ind_node, self.context,
                                         requires=list(dep_nodes),
                                         fn=self._destroy_stack,
                                         watch_func=(self._tail_stack if
                                                     tail else None))
                )

        graph.add_steps(inverse_steps)

        # invert what is going to be destroyed to retain dependencies
        graph = graph.transposed()

        steps = [Step(stack, fn=self._launch_stack,
                      watch_func=(self._tail_stack if tail else None))
                 for stack in self.context.get_stacks()]

        steps += [Step(target, fn=target_fn)
                  for target in self.context.get_targets()]

        graph.add_steps(steps)

        return Plan(
            context=self.context,
            description=self.DESCRIPTION,
            graph=graph
        )