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

def test_generate_plan_persist_destroy(self, mock_graph_tags):
        """Test generate plan persist destroy."""
        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)]
        )
        build_action = build.Action(context=context)
        plan = build_action._Action__generate_plan()

        self.assertIsInstance(plan, Plan)
        self.assertEqual(build.Action.DESCRIPTION, plan.description)
        mock_graph_tags.assert_called_once()
        # order is different between python2/3 so can't compare dicts
        result_graph_dict = plan.graph.to_dict()
        self.assertEqual(5, len(result_graph_dict))
        self.assertEqual(set(), result_graph_dict['other'])
        self.assertEqual(set(), result_graph_dict['removed'])
        self.assertEqual(set(), result_graph_dict['vpc'])
        self.assertEqual(set(['vpc']), result_graph_dict['bastion'])
        self.assertEqual(set(['bastion', 'vpc']), result_graph_dict['db'])

def test_put_persistent_graph(self):
        """Return 'None' when put is successful."""
        code = '0000'
        context = Context(config=self.persist_graph_config)
        context._s3_bucket_verified = True
        graph_dict = {
            'stack1': [],
            'stack2': ['stack1']
        }
        context._persistent_graph = Graph.from_dict(graph_dict, context)
        stubber = Stubber(context.s3_conn)
        expected_params = {'Body': json.dumps(graph_dict, indent=4),
                           'ServerSideEncryption': 'AES256',
                           'ACL': 'bucket-owner-full-control',
                           'ContentType': 'application/json',
                           'Tagging': '{}={}'.format(
                               context._persistent_graph_lock_tag,
                               code)}
        expected_params.update(context.persistent_graph_location)

        stubber.add_response('get_object_tagging',
                             {'TagSet': gen_tagset(
                                 {context._persistent_graph_lock_tag: code}
                             )},
                             context.persistent_graph_location)
        stubber.add_response('put_object', {}, expected_params)

def test_unlock_persistent_graph_no_object(self):
        """Return 'None' when object does not exist.

        This can occur if the object is deleted by 'put_persistent_graph'.

        """
        code = '0000'
        context = Context(config=self.persist_graph_config)
        context._s3_bucket_verified = True
        context._persistent_graph = Graph()
        stubber = Stubber(context.s3_conn)
        expected_params = context.persistent_graph_location.copy()
        expected_params.update({'ResponseContentType': 'application/json'})

        stubber.add_client_error(
            'get_object', 'NoSuchKey',
            expected_params=expected_params
        )

        with stubber:
            assert context.unlock_persistent_graph(code)
            stubber.assert_no_pending_responses()

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)]
        )
        build_action = build.Action(context=context)
        build_action.run()

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

def test_lock_persistent_graph_no_object(self):
        """Error raised when when there is no object to lock."""
        code = '0000'
        context = Context(config=self.persist_graph_config)
        context._s3_bucket_verified = True
        context._persistent_graph = Graph()
        stubber = Stubber(context.s3_conn)
        expected_params = {'Tagging': {
            'TagSet': gen_tagset({context._persistent_graph_lock_tag: code})
        }}
        expected_params.update(context.persistent_graph_location)

        stubber.add_client_error(
            'get_object_tagging', 'NoSuchKey',
            expected_params=context.persistent_graph_location
        )
        stubber.add_client_error('put_object_tagging', 'NoSuchKey',
                                 expected_params=expected_params)

        with stubber:
            with self.assertRaises(PersistentGraphCannotLock):
                context.lock_persistent_graph(code)

def test_from_dict(self):
        """Test from dict."""
        graph = Graph.from_dict(self.graph_dict, self.context)

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

def test_lock_persistent_graph(self):
        """Return 'None' when lock is successful."""
        code = '0000'
        context = Context(config=self.persist_graph_config)
        context._s3_bucket_verified = True
        context._persistent_graph = Graph()
        stubber = Stubber(context.s3_conn)
        expected_params = {'Tagging': {
            'TagSet': gen_tagset({context._persistent_graph_lock_tag: code})
        }}
        expected_params.update(context.persistent_graph_location)

        stubber.add_response('get_object_tagging', {'TagSet': []},
                             context.persistent_graph_location)
        stubber.add_response('put_object_tagging', {}, expected_params)

        with stubber:
            self.assertIsNone(context.lock_persistent_graph(code))
            stubber.assert_no_pending_responses()

def test_generate_plan_with_persist_no_lock_req(self, mock_stack_action,
                                                    mock_tags):
        """Test generate plan with persist no lock req."""
        mock_stack_action.return_value = MagicMock()
        mock_tags.return_value = {}
        context = mock_context(namespace='test',
                               extra_config_args=self.config_persist,
                               region=self.region)
        persist_step = Step.from_stack_name('removed', context)
        context._persistent_graph = Graph.from_steps([persist_step])
        action = BaseAction(context=context,
                            provider_builder=MockProviderBuilder(
                                self.provider, region=self.region))

        plan = action._generate_plan(include_persistent_graph=True,
                                     require_unlocked=False)

        self.assertIsInstance(plan, Plan)
        mock_tags.assert_called_once()
        # order is different between python2/3 so can't compare dicts
        result_graph_dict = plan.graph.to_dict()
        self.assertEqual(3, len(result_graph_dict))
        self.assertEqual(set(), result_graph_dict['stack1'])
        self.assertEqual(set(['stack1']), result_graph_dict['stack2'])
        self.assertEqual(set(), result_graph_dict['removed'])
        self.assertEqual(BaseAction.DESCRIPTION, plan.description)

def test_add_steps(self):
        """Test add steps."""
        graph = Graph()
        graph.add_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())

graph but are no longer in the "local" graph will be deleted.
        If not using a persistent graph. the default method for creating
        a plan is used.

        Args:
            tail (Union[bool, Callable]): An optional function to call
                to tail the stack progress.

        Returns:
            :class:`runway.cfngin.plan.Plan`: The resulting plan object.

        """
        if not self.context.persistent_graph:
            return self._generate_plan(tail)

        graph = Graph()
        config_stack_names = [stack.name for stack in
                              self.context.get_stacks()]
        inverse_steps = []
        persist_graph = self.context.persistent_graph.transposed()

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

        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