How to use the chalice.deploy.deployer.ApplicationGraphBuilder function in chalice

def test_can_build_rest_api_with_authorizer(self, rest_api_app):
        @rest_api_app.authorizer()
        def my_auth(auth_request):
            pass

        @rest_api_app.route('/auth', authorizer=my_auth)
        def needs_auth():
            return {'foo': 'bar'}

        config = self.create_config(rest_api_app,
                                    app_name='rest-api-app',
                                    autogen_policy=True)
        builder = ApplicationGraphBuilder()
        application = builder.build(config, stage_name='dev')
        rest_api = application.resources[0]
        assert len(rest_api.authorizers) == 1
        assert isinstance(rest_api.authorizers[0], models.LambdaFunction)

def test_can_create_websocket_app_missing_connect(
            self, websocket_app_without_connect):
        config = self.create_config(websocket_app_without_connect,
                                    app_name='websocket-app',
                                    autogen_policy=True)
        builder = ApplicationGraphBuilder()
        application = builder.build(config, stage_name='dev')
        assert len(application.resources) == 1
        websocket_api = application.resources[0]
        assert isinstance(websocket_api, models.WebsocketAPI)
        assert websocket_api.resource_name == 'websocket_api'
        assert sorted(websocket_api.routes) == sorted(
            ['$default', '$disconnect'])
        assert websocket_api.api_gateway_stage == 'api'

        connect_function = websocket_api.connect_function
        assert connect_function is None

        message_function = websocket_api.message_function
        assert message_function.resource_name == 'websocket_message'
        assert message_function.handler == 'app.message'

def test_can_create_sns_event_handler(self, sns_event_app):
        config = self.create_config(sns_event_app,
                                    app_name='s3-event-app',
                                    autogen_policy=True)
        builder = ApplicationGraphBuilder()
        application = builder.build(config, stage_name='dev')
        assert len(application.resources) == 1
        sns_event = application.resources[0]
        assert isinstance(sns_event, models.SNSLambdaSubscription)
        assert sns_event.resource_name == 'handler-sns-subscription'
        assert sns_event.topic == 'mytopic'
        lambda_function = sns_event.lambda_function
        assert lambda_function.resource_name == 'handler'
        assert lambda_function.handler == 'app.handler'

def test_exception_raised_when_missing_vpc_params(self, lambda_app):
        @lambda_app.lambda_function()
        def foo(event, context):
            pass

        builder = ApplicationGraphBuilder()
        config = self.create_config(lambda_app,
                                    iam_role_arn='role:arn',
                                    security_group_ids=['sg1', 'sg2'],
                                    subnet_ids=[])
        with pytest.raises(ChaliceBuildError):
            builder.build(config, stage_name='dev')

def test_no_inject_is_already_injected(self, sqs_event_app):
        @sqs_event_app.on_sqs_message(queue='second-queue')
        def second_handler(event):
            pass

        config = Config.create(chalice_app=sqs_event_app,
                               autogen_policy=True,
                               project_dir='.')
        builder = ApplicationGraphBuilder()
        application = builder.build(config, stage_name='dev')
        event_sources = application.resources
        role = event_sources[0].lambda_function.role
        role.policy.document = {'Statement': []}
        injector = LambdaEventSourcePolicyInjector()
        injector.handle(config, event_sources[0])
        injector.handle(config, event_sources[1])
        # Even though we have two queue handlers, we only need to
        # inject the policy once.
        assert role.policy.document == {
            'Statement': [SQS_EVENT_SOURCE_POLICY.copy()],
        }

def test_vpc_trait_added_when_vpc_configured(self, lambda_app):
        @lambda_app.lambda_function()
        def foo(event, context):
            pass

        builder = ApplicationGraphBuilder()
        config = self.create_config(lambda_app,
                                    autogen_policy=True,
                                    security_group_ids=['sg1', 'sg2'],
                                    subnet_ids=['sn1', 'sn2'])
        application = builder.build(config, stage_name='dev')

        policy = application.resources[0].role.policy
        assert policy == models.AutoGenIAMPolicy(
            document=models.Placeholder.BUILD_STAGE,
            traits=set([models.RoleTraits.VPC_NEEDED]),
        )

def test_autogen_policy_for_function(self, lambda_app):
        # This test is just a sanity test that verifies all the params
        # for an ManagedIAMRole.  The various combinations for role
        # configuration is all tested via RoleTestCase.
        config = self.create_config(lambda_app, autogen_policy=True)
        builder = ApplicationGraphBuilder()
        application = builder.build(config, stage_name='dev')
        function = application.resources[0]
        role = function.role
        # We should have linked a ManagedIAMRole
        assert isinstance(role, models.ManagedIAMRole)
        assert role == models.ManagedIAMRole(
            resource_name='default-role',
            role_name='lambda-only-dev',
            trust_policy=LAMBDA_TRUST_POLICY,
            policy=models.AutoGenIAMPolicy(models.Placeholder.BUILD_STAGE),
        )

def setup_method(self):
        self.resource_builder = package.ResourceBuilder(
            application_builder=ApplicationGraphBuilder(),
            deps_builder=DependencyBuilder(),
            build_stage=mock.Mock(spec=BuildStage)
        )
        self.template_gen = self.template_gen_factory(Config())

def create_app_packager(
        config, package_format='cloudformation', merge_template=None):
    # type: (Config, str, Optional[str]) -> AppPackager
    osutils = OSUtils()
    ui = UI()
    application_builder = ApplicationGraphBuilder()
    deps_builder = DependencyBuilder()
    post_processors = []  # type: List[TemplatePostProcessor]
    generator = None  # type: Union[None, TemplateGenerator]

    if package_format == 'cloudformation':
        build_stage = create_build_stage(
            osutils, ui, CFNSwaggerGenerator())
        post_processors.extend([
            SAMCodeLocationPostProcessor(osutils=osutils),
            TemplateMergePostProcessor(
                osutils=osutils,
                merger=TemplateDeepMerger(),
                merge_template=merge_template)])
        generator = SAMTemplateGenerator(config)
    else:
        build_stage = create_build_stage(

def create_default_deployer(session, config, ui):
    # type: (Session, Config, UI) -> Deployer
    client = TypedAWSClient(session)
    osutils = OSUtils()
    return Deployer(
        application_builder=ApplicationGraphBuilder(),
        deps_builder=DependencyBuilder(),
        build_stage=create_build_stage(
            osutils, UI(), TemplatedSwaggerGenerator(),
        ),
        plan_stage=PlanStage(
            osutils=osutils, remote_state=RemoteState(
                client, config.deployed_resources(config.chalice_stage)),
        ),
        sweeper=ResourceSweeper(),
        executor=Executor(client, ui),
        recorder=ResultsRecorder(osutils=osutils),
    )