How to use the runway.cfngin.plan.Graph.from_steps function in runway

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

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

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

def test_run_persist(self, mock_execute, mock_unlock, mock_lock,
                         mock_graph_tags):
        """Test run persist."""
        mock_graph_tags.return_value = {}
        context = self._get_context(
            extra_config_args={'persistent_graph_key': 'test.json'}
        )
        context._persistent_graph = Graph.from_steps(
            [Step.from_stack_name('removed', context)]
        )
        destroy_action = destroy.Action(context=context)
        destroy_action.run(force=True)

        mock_graph_tags.assert_called_once()
        mock_lock.assert_called_once()
        mock_execute.assert_called_once()
        mock_unlock.assert_called_once()

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'])

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:
                return SKIPPED
            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'])

def test_from_steps(self):
        """Test from steps."""
        graph = Graph.from_steps(self.steps)

        self.assertEqual(self.steps, list(graph.steps.values()))
        self.assertEqual([step.name for step in self.steps],
                         list(graph.steps.keys()))
        self.assertEqual(self.graph_dict_expected, graph.to_dict())

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'])

Args:
        graph1 (:class:`Graph`): Graph that ``graph2`` will
            be merged into.
        graph2 (:class:`Graph`): Graph that will be merged
            into ``graph1``.

    Returns:
        :class:`Graph`: A combined graph.

    """
    merged_graph_dict = merge_map(graph1.to_dict().copy(),
                                  graph2.to_dict())
    steps = [graph1.steps.get(name, graph2.steps.get(name))
             for name in merged_graph_dict.keys()]
    return Graph.from_steps(steps)

"""
        tail = self._tail_stack if tail else None

        def target_fn(*_args, **_kwargs):
            """Target function."""
            return COMPLETE

        steps = [
            Step(stack, fn=self._stack_action, watch_func=tail)
            for stack in self.context.get_stacks()]

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

        graph = Graph.from_steps(steps)

        if include_persistent_graph and self.context.persistent_graph:
            persist_steps = Step.from_persistent_graph(
                self.context.persistent_graph.to_dict(),
                self.context,
                fn=self._stack_action,
                watch_func=tail
            )
            persist_graph = Graph.from_steps(persist_steps)
            graph = merge_graphs(graph, persist_graph)

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